From ff80dff55fdffeae900272f86ea4ab9efc39bd7c Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Fri, 17 Jul 2026 13:30:51 +0000 Subject: [PATCH] feat(plugins): atomic multi-document storage batch (ctx.storage.batch) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds ctx.storage.batch([...ops]) — an all-or-nothing batch of conditional writes (insert / updateIf) across multiple documents and collections. Commits iff every op's guard passes; any guard failure rolls back the whole batch and reports { applied:false, failedIndex, reason, conflictField? }. - Core: factor buildInsertQuery / buildUpdateIfQuery / normalizeDeltaEntries out of insert/updateIf (behaviour-preserving); lift recoverConflictField to a free function; add applyPluginStorageBatch (pg/sqlite real transaction) and applyPluginStorageBatchD1 (raw env.DB.batch with interleaved zero-rows guard assertions + failure-path diagnosis). Reserve "batch" as a collection name. - StorageAccess intersection type + PluginContext.storage; ctx.storage.batch wired in createStorageAccess / createPluginStorageAccessor. - workerd + cloudflare bridges: storage/batch dispatch with per-op declared collection validation (anti-smuggling); identical BatchResult shape. - Tests: core contract suite (both dialects) incl. coupled no-oversell on Postgres; workerd bridge round-trip; real-D1 suite via vitest-pool-workers. --- .changeset/plugin-storage-atomic-batch.md | 9 + .changeset/plugin-storage-batch-cloudflare.md | 5 + .changeset/plugin-storage-batch-workerd.md | 5 + packages/cloudflare/package.json | 4 +- packages/cloudflare/src/sandbox/bridge.ts | 47 +- packages/cloudflare/src/sandbox/wrapper.ts | 3 + .../d1/migrations/0001_plugin_storage.sql | 36 + .../tests/d1/storage-batch.d1.test.ts | 321 ++++++++ packages/cloudflare/tests/d1/tsconfig.json | 7 + packages/cloudflare/tests/d1/vitest.config.ts | 35 + packages/cloudflare/tests/d1/worker.ts | 7 + packages/cloudflare/tests/d1/wrangler.jsonc | 14 + packages/cloudflare/tsconfig.json | 3 +- packages/cloudflare/vitest.config.ts | 28 + .../core/src/database/repositories/index.ts | 3 + .../database/repositories/plugin-storage.ts | 764 +++++++++++++++--- packages/core/src/index.ts | 14 + packages/core/src/plugins/context.ts | 16 +- packages/core/src/plugins/index.ts | 11 + packages/core/src/plugins/storage-indexes.ts | 26 + packages/core/src/plugins/types.ts | 109 ++- .../plugins/storage-batch-no-oversell.test.ts | 124 +++ .../integration/plugins/storage-batch.test.ts | 526 ++++++++++++ .../unit/plugins/storage-batch-access.test.ts | 79 ++ .../workerd/src/sandbox/bridge-handler.ts | 68 +- packages/workerd/src/sandbox/wrapper.ts | 3 + packages/workerd/test/bridge-handler.test.ts | 161 ++++ pnpm-lock.yaml | 26 +- 28 files changed, 2331 insertions(+), 123 deletions(-) create mode 100644 .changeset/plugin-storage-atomic-batch.md create mode 100644 .changeset/plugin-storage-batch-cloudflare.md create mode 100644 .changeset/plugin-storage-batch-workerd.md create mode 100644 packages/cloudflare/tests/d1/migrations/0001_plugin_storage.sql create mode 100644 packages/cloudflare/tests/d1/storage-batch.d1.test.ts create mode 100644 packages/cloudflare/tests/d1/tsconfig.json create mode 100644 packages/cloudflare/tests/d1/vitest.config.ts create mode 100644 packages/cloudflare/tests/d1/worker.ts create mode 100644 packages/cloudflare/tests/d1/wrangler.jsonc create mode 100644 packages/cloudflare/vitest.config.ts create mode 100644 packages/core/tests/integration/plugins/storage-batch-no-oversell.test.ts create mode 100644 packages/core/tests/integration/plugins/storage-batch.test.ts create mode 100644 packages/core/tests/unit/plugins/storage-batch-access.test.ts diff --git a/.changeset/plugin-storage-atomic-batch.md b/.changeset/plugin-storage-atomic-batch.md new file mode 100644 index 0000000000..c72c0ed064 --- /dev/null +++ b/.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). diff --git a/.changeset/plugin-storage-batch-cloudflare.md b/.changeset/plugin-storage-batch-cloudflare.md new file mode 100644 index 0000000000..7e3b2409ea --- /dev/null +++ b/.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. diff --git a/.changeset/plugin-storage-batch-workerd.md b/.changeset/plugin-storage-batch-workerd.md new file mode 100644 index 0000000000..2a588bc70f --- /dev/null +++ b/.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. diff --git a/packages/cloudflare/package.json b/packages/cloudflare/package.json index c91c22745b..b6bf2daf34 100644 --- a/packages/cloudflare/package.json +++ b/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", diff --git a/packages/cloudflare/src/sandbox/bridge.ts b/packages/cloudflare/src/sandbox/bridge.ts index d302c898c6..ddfb4f07ce 100644 --- a/packages/cloudflare/src/sandbox/bridge.ts +++ b/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 { + 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 // ========================================================================= diff --git a/packages/cloudflare/src/sandbox/wrapper.ts b/packages/cloudflare/src/sandbox/wrapper.ts index 3605e2e0f7..4897dd9e20 100644 --- a/packages/cloudflare/src/sandbox/wrapper.ts +++ b/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); } }); diff --git a/packages/cloudflare/tests/d1/migrations/0001_plugin_storage.sql b/packages/cloudflare/tests/d1/migrations/0001_plugin_storage.sql new file mode 100644 index 0000000000..2444552c65 --- /dev/null +++ b/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"]' +); diff --git a/packages/cloudflare/tests/d1/storage-batch.d1.test.ts b/packages/cloudflare/tests/d1/storage-batch.d1.test.ts new file mode 100644 index 0000000000..4532f7e8bf --- /dev/null +++ b/packages/cloudflare/tests/d1/storage-batch.d1.test.ts @@ -0,0 +1,321 @@ +/** + * Real-D1 tests for `applyPluginStorageBatchD1` — the raw `env.DB.batch()` + * atomic-batch path used by the Cloudflare `PluginBridge` in production. + * + * Runs inside a real workerd isolate with a real miniflare-backed D1 binding + * (via `@cloudflare/vitest-pool-workers`), against the ACTUAL `_plugin_storage` + * schema (see ./migrations). Proves the interleaved zero-rows-assertion + * mechanism genuinely rolls back on D1 (where a bare 0-row UPDATE would NOT), + * that `changes()` carries across batch statements, that RETURNING surfaces per + * op, and that the failure-path diagnosis reports the right failedIndex/reason. + */ + +import { applyD1Migrations, env } from "cloudflare:test"; +import { applyPluginStorageBatchD1, type BatchOp, type D1BatchBinding } from "emdash"; +import { beforeAll, beforeEach, describe, expect, it } from "vitest"; + +interface TestEnv { + DB: D1Database; + TEST_MIGRATIONS: Parameters[1]; +} +const testEnv = env as unknown as TestEnv; +const DB = () => testEnv.DB; +// The production executor takes the raw binding; D1Database is structurally a +// D1BatchBinding (prepare/bind/first/batch). +const runBatch = (ops: BatchOp[]) => + applyPluginStorageBatchD1(DB() as unknown as D1BatchBinding, "shop", ops); + +beforeAll(async () => { + await applyD1Migrations(testEnv.DB, testEnv.TEST_MIGRATIONS); +}); + +async function put(collection: string, id: string, data: unknown): Promise { + const now = new Date().toISOString(); + await DB() + .prepare( + "INSERT OR REPLACE INTO _plugin_storage (plugin_id, collection, id, data, created_at, updated_at) VALUES ('shop', ?, ?, ?, ?, ?)", + ) + .bind(collection, id, JSON.stringify(data), now, now) + .run(); +} + +async function read(collection: string, id: string): Promise | null> { + const row = await DB() + .prepare("SELECT data FROM _plugin_storage WHERE plugin_id='shop' AND collection=? AND id=?") + .bind(collection, id) + .first<{ data: string }>(); + return row ? JSON.parse(row.data) : null; +} + +beforeEach(async () => { + // Clear all rows between tests (keep the schema + unique index + tracking row). + await DB().prepare("DELETE FROM _plugin_storage WHERE plugin_id='shop'").run(); +}); + +describe("applyPluginStorageBatchD1 — real D1", () => { + it("guard-fail rolls back BOTH coupled writes on real D1 (the atomicity proof)", async () => { + await put("inventory", "widget", { on_hand: 1 }); + await put("reservations", "r1", { state: "pending", sku: "widget", qty: 2 }); + + const result = await runBatch([ + { + op: "updateIf", + collection: "inventory", + id: "widget", + where: { on_hand: { gte: 2 } }, + delta: { on_hand: { dec: 2 } }, + }, + { + op: "updateIf", + collection: "reservations", + id: "r1", + where: { state: "pending" }, + set: { state: "held" }, + }, + ]); + + expect(result).toEqual({ applied: false, failedIndex: 0, reason: "guard_failed" }); + // Neither moved — the flip did not happen because the decrement rolled back. + expect((await read("inventory", "widget"))?.on_hand).toBe(1); + expect((await read("reservations", "r1"))?.state).toBe("pending"); + }); + + it("changes() carries across statements within one env.DB.batch (load-bearing assumption)", async () => { + await put("inventory", "a", { on_hand: 10 }); + await put("inventory", "b", { on_hand: 10 }); + + const rows = await DB().batch<{ c: number }>([ + DB().prepare( + "UPDATE _plugin_storage SET data=json_set(data,'$.on_hand',json_extract(data,'$.on_hand')-1) WHERE plugin_id='shop' AND collection='inventory' AND id='a' AND json_extract(data,'$.on_hand')>=1", + ), + DB().prepare("SELECT changes() AS c"), + DB().prepare( + "UPDATE _plugin_storage SET data=json_set(data,'$.on_hand',json_extract(data,'$.on_hand')-1) WHERE plugin_id='shop' AND collection='inventory' AND id='b' AND json_extract(data,'$.on_hand')>=999", + ), + DB().prepare("SELECT changes() AS c"), + ]); + + expect(rows[1]?.results?.[0]?.c).toBe(1); + expect(rows[3]?.results?.[0]?.c).toBe(0); + }); + + it("happy path commits and returns RETURNING data per updateIf op", async () => { + await put("inventory", "widget", { on_hand: 5 }); + await put("reservations", "r1", { state: "pending", sku: "widget", qty: 2 }); + + const result = await runBatch([ + { + op: "updateIf", + collection: "inventory", + id: "widget", + where: { on_hand: { gte: 2 } }, + delta: { on_hand: { dec: 2 } }, + }, + { + op: "updateIf", + collection: "reservations", + id: "r1", + where: { state: "pending" }, + set: { state: "held" }, + }, + ]); + + expect(result.applied).toBe(true); + if (!result.applied) throw new Error("expected applied"); + expect(result.results[0]).toEqual({ + op: "updateIf", + applied: true, + data: { on_hand: 3 }, + }); + expect((result.results[1] as { data: { state: string } }).data.state).toBe("held"); + expect((await read("inventory", "widget"))?.on_hand).toBe(3); + expect((await read("reservations", "r1"))?.state).toBe("held"); + }); + + it("full reserve: claim insert ∧ decrement commit atomically on D1", async () => { + await put("inventory", "widget", { on_hand: 5 }); + + const result = await runBatch([ + { + op: "insert", + collection: "reservations", + id: "res-1", + data: { state: "held", sku: "widget", qty: 2, idempotency_key: "key-1" }, + }, + { + op: "updateIf", + collection: "inventory", + id: "widget", + where: { on_hand: { gte: 2 } }, + delta: { on_hand: { dec: 2 } }, + }, + ]); + + expect(result.applied).toBe(true); + if (!result.applied) throw new Error("expected applied"); + expect(result.results[0]).toEqual({ op: "insert", inserted: true }); + expect((await read("inventory", "widget"))?.on_hand).toBe(3); + expect((await read("reservations", "res-1"))?.state).toBe("held"); + }); + + it("diagnosis reports the correct failedIndex/reason when the SECOND op fails", async () => { + await put("inventory", "widget", { on_hand: 5 }); + await put("reservations", "r1", { state: "held", sku: "widget", qty: 2 }); + + const result = await runBatch([ + { + op: "updateIf", + collection: "inventory", + id: "widget", + where: { on_hand: { gte: 2 } }, + delta: { on_hand: { dec: 2 } }, + }, + { + op: "updateIf", + collection: "reservations", + id: "r1", + where: { state: "pending" }, + set: { state: "released" }, + }, + ]); + + expect(result).toEqual({ applied: false, failedIndex: 1, reason: "guard_failed" }); + // op0's decrement rolled back too. + expect((await read("inventory", "widget"))?.on_hand).toBe(5); + }); + + it("duplicate claim (same id) → failedIndex 0 reason exists, decrement rolled back", async () => { + await put("inventory", "widget", { on_hand: 5 }); + await put("reservations", "res-1", { + state: "held", + sku: "widget", + qty: 2, + idempotency_key: "k", + }); + + const result = await runBatch([ + { + op: "insert", + collection: "reservations", + id: "res-1", + data: { state: "held", sku: "widget", qty: 2, idempotency_key: "k2" }, + }, + { + op: "updateIf", + collection: "inventory", + id: "widget", + where: { on_hand: { gte: 2 } }, + delta: { on_hand: { dec: 2 } }, + }, + ]); + + expect(result).toEqual({ applied: false, failedIndex: 0, reason: "exists" }); + expect((await read("inventory", "widget"))?.on_hand).toBe(5); + }); + + it("unique_violation on idempotency_key (different id, same key) → failedIndex 0 + conflictField", async () => { + await put("inventory", "widget", { on_hand: 5 }); + await put("reservations", "res-1", { + state: "held", + sku: "widget", + qty: 2, + idempotency_key: "dup", + }); + + const result = await runBatch([ + { + op: "insert", + collection: "reservations", + id: "res-2", + data: { state: "held", sku: "widget", qty: 2, idempotency_key: "dup" }, + }, + { + op: "updateIf", + collection: "inventory", + id: "widget", + where: { on_hand: { gte: 2 } }, + delta: { on_hand: { dec: 2 } }, + }, + ]); + + expect(result).toEqual({ + applied: false, + failedIndex: 0, + reason: "unique_violation", + conflictField: "idempotency_key", + }); + expect((await read("inventory", "widget"))?.on_hand).toBe(5); + expect(await read("reservations", "res-2")).toBeNull(); + }); + + it("ifNotExists insert treats an existing row as a satisfied no-op and commits siblings", async () => { + await put("inventory", "widget", { on_hand: 5 }); + await put("reservations", "res-1", { + state: "held", + sku: "widget", + qty: 2, + idempotency_key: "k", + }); + + const result = await runBatch([ + { + op: "insert", + collection: "reservations", + id: "res-1", + ifNotExists: true, + data: { state: "held", sku: "widget", qty: 2, idempotency_key: "k" }, + }, + { + op: "updateIf", + collection: "inventory", + id: "widget", + where: { on_hand: { gte: 2 } }, + delta: { on_hand: { dec: 2 } }, + }, + ]); + + expect(result.applied).toBe(true); + if (!result.applied) throw new Error("expected applied"); + expect(result.results[0]).toEqual({ op: "insert", inserted: false, reason: "exists" }); + // The decrement DID apply. + expect((await read("inventory", "widget"))?.on_hand).toBe(3); + }); + + it("concurrent coupled reserve batches never throw and report well-formed results (diagnosis is graceful)", async () => { + await put("inventory", "widget", { on_hand: 3 }); + + const N = 8; + const results = await Promise.all( + Array.from({ length: N }, (_v, i) => + runBatch([ + { + op: "insert", + collection: "reservations", + id: `res-${i}`, + data: { state: "held", sku: "widget", qty: 1, idempotency_key: `key-${i}` }, + }, + { + op: "updateIf", + collection: "inventory", + id: "widget", + where: { on_hand: { gte: 1 } }, + delta: { on_hand: { dec: 1 } }, + }, + ]), + ), + ); + + // Every call resolved (no throw); every result is a well-formed BatchResult. + for (const r of results) { + expect(typeof r.applied).toBe("boolean"); + if (!r.applied) { + expect(typeof r.failedIndex).toBe("number"); + expect(["guard_failed", "exists", "unique_violation"]).toContain(r.reason); + } + } + // D1 serializes writes, so no oversell: on_hand never goes below 0. + const onHand = (await read("inventory", "widget"))?.on_hand as number; + expect(onHand).toBeGreaterThanOrEqual(0); + expect(onHand).toBeLessThanOrEqual(3); + }); +}); diff --git a/packages/cloudflare/tests/d1/tsconfig.json b/packages/cloudflare/tests/d1/tsconfig.json new file mode 100644 index 0000000000..175eda3b2b --- /dev/null +++ b/packages/cloudflare/tests/d1/tsconfig.json @@ -0,0 +1,7 @@ +{ + "extends": "../../tsconfig.json", + "compilerOptions": { + "types": ["@cloudflare/vitest-pool-workers/types"] + }, + "include": ["**/*.ts"] +} diff --git a/packages/cloudflare/tests/d1/vitest.config.ts b/packages/cloudflare/tests/d1/vitest.config.ts new file mode 100644 index 0000000000..4f66f3607e --- /dev/null +++ b/packages/cloudflare/tests/d1/vitest.config.ts @@ -0,0 +1,35 @@ +/** + * Real-D1 test project for the atomic storage batch (`applyPluginStorageBatchD1`). + * + * Runs inside a real workerd isolate via `@cloudflare/vitest-pool-workers` + * (miniflare-backed D1), against the ACTUAL `_plugin_storage` schema created by + * `migrations/0001_plugin_storage.sql`. Kept SEPARATE from the package's default + * node-pool suites (referenced as a project from the package `vitest.config.ts`) + * because the workers pool cannot host the plain node suites — the probe found a + * single pool-workers config silently drops them. + */ + +import { fileURLToPath } from "node:url"; + +import { cloudflareTest, readD1Migrations } from "@cloudflare/vitest-pool-workers"; +import { defineConfig } from "vitest/config"; + +const here = (p: string) => fileURLToPath(new URL(p, import.meta.url)); +const migrations = await readD1Migrations(here("./migrations")); + +export default defineConfig({ + test: { + name: "d1", + include: [here("./*.d1.test.ts")], + }, + plugins: [ + cloudflareTest({ + wrangler: { configPath: here("./wrangler.jsonc") }, + miniflare: { + bindings: { + TEST_MIGRATIONS: migrations, + }, + }, + }), + ], +}); diff --git a/packages/cloudflare/tests/d1/worker.ts b/packages/cloudflare/tests/d1/worker.ts new file mode 100644 index 0000000000..b987e3359c --- /dev/null +++ b/packages/cloudflare/tests/d1/worker.ts @@ -0,0 +1,7 @@ +// Minimal worker entry so vitest-pool-workers has a `main` to load. The tests +// drive `env.DB` directly; this fetch handler is unused. +export default { + fetch(): Response { + return new Response("d1-plugin-storage-batch"); + }, +}; diff --git a/packages/cloudflare/tests/d1/wrangler.jsonc b/packages/cloudflare/tests/d1/wrangler.jsonc new file mode 100644 index 0000000000..579a65ba5c --- /dev/null +++ b/packages/cloudflare/tests/d1/wrangler.jsonc @@ -0,0 +1,14 @@ +{ + "$schema": "node_modules/wrangler/config-schema.json", + "name": "d1-plugin-storage-batch", + "main": "./worker.ts", + "compatibility_date": "2026-02-24", + "compatibility_flags": ["nodejs_compat"], + "d1_databases": [ + { + "binding": "DB", + "database_name": "plugin-storage-batch", + "migrations_dir": "./migrations", + }, + ], +} diff --git a/packages/cloudflare/tsconfig.json b/packages/cloudflare/tsconfig.json index 39d063dda4..d0d8561e7c 100644 --- a/packages/cloudflare/tsconfig.json +++ b/packages/cloudflare/tsconfig.json @@ -13,5 +13,6 @@ "noImplicitOverride": true, "types": ["@cloudflare/workers-types", "astro/client"] }, - "include": ["src/**/*.ts", "tests/**/*.ts"] + "include": ["src/**/*.ts", "tests/**/*.ts"], + "exclude": ["tests/d1/**"] } diff --git a/packages/cloudflare/vitest.config.ts b/packages/cloudflare/vitest.config.ts new file mode 100644 index 0000000000..75fbca0a36 --- /dev/null +++ b/packages/cloudflare/vitest.config.ts @@ -0,0 +1,28 @@ +/** + * Two-project test setup: + * + * - `node` — the package's default suites, run in the standard node pool. + * - `d1` — real-D1 tests for the atomic storage batch, run inside a workerd + * isolate via `@cloudflare/vitest-pool-workers` (see + * `tests/d1/vitest.config.ts`). Kept as a SEPARATE project because a + * single pool-workers config would silently drop the node suites. + * + * `pnpm test` (`vitest run`) runs BOTH projects. + */ + +import { defineConfig } from "vitest/config"; + +export default defineConfig({ + test: { + projects: [ + { + test: { + name: "node", + include: ["tests/**/*.test.ts"], + exclude: ["tests/d1/**", "**/node_modules/**", "**/dist/**"], + }, + }, + "./tests/d1/vitest.config.ts", + ], + }, +}); diff --git a/packages/core/src/database/repositories/index.ts b/packages/core/src/database/repositories/index.ts index d67ccd8555..58091db39d 100644 --- a/packages/core/src/database/repositories/index.ts +++ b/packages/core/src/database/repositories/index.ts @@ -22,7 +22,10 @@ export { createPluginStorageAccessor, deleteAllPluginStorage, deletePluginCollection, + applyPluginStorageBatch, + applyPluginStorageBatchD1, } from "./plugin-storage.js"; +export type { D1BatchBinding, D1BatchStatement } from "./plugin-storage.js"; export { MenuGoneError, MenuRepository } from "./menu.js"; export type { Menu, diff --git a/packages/core/src/database/repositories/plugin-storage.ts b/packages/core/src/database/repositories/plugin-storage.ts index ff4a4ad0f4..cb17f1bf91 100644 --- a/packages/core/src/database/repositories/plugin-storage.ts +++ b/packages/core/src/database/repositories/plugin-storage.ts @@ -7,8 +7,18 @@ * @see PLUGIN-SYSTEM.md § Plugin Storage > Full API Reference */ -import type { Kysely, RawBuilder, SqlBool } from "kysely"; -import { sql } from "kysely"; +import { + DummyDriver, + SqliteAdapter, + SqliteIntrospector, + SqliteQueryCompiler, + Kysely, + sql, + type RawBuilder, + type SqlBool, + type UpdateQueryBuilder, + type InsertQueryBuilder, +} from "kysely"; import { buildWhereClause, @@ -26,10 +36,17 @@ import type { InsertResult, UpdateIfArgs, UpdateIfResult, + BatchOp, + BatchOpResult, + BatchResult, + BatchFailureReason, + NumericDelta, + StorageAccess, } from "../../plugins/types.js"; import { pluginDataWriteExpr } from "../dialect-helpers.js"; import { withTransaction } from "../transaction.js"; import type { Database } from "../types.js"; +import { validateIdentifier } from "../validate.js"; import { encodeCursor, decodeCursor } from "./types.js"; /** @@ -99,6 +116,188 @@ function buildRawWhereExpression(whereResult: { return sql`${sql.join(parts, sql.raw(""))}`; } +/** + * Best-effort recovery of the single field behind a unique-index violation. + * Prefers the `_plugin_indexes` tracking row (authoritative field list); falls + * back to parsing the `generateIndexName` format. Composite indexes yield + * `undefined`. + * + * Lifted to a module-level free function (from the former private method) so + * both `PluginStorageRepository.insert` and the batch executors can call it + * with an explicit `(db, pluginId, collection)` — behaviour-preserving. + */ +async function recoverConflictField( + db: Kysely, + pluginId: string, + collection: string, + indexName?: string, +): Promise { + if (!indexName) return undefined; + + const row = await db + .selectFrom("_plugin_indexes") + .select("fields") + .where("plugin_id", "=", pluginId) + .where("collection", "=", collection) + .where("index_name", "=", indexName) + .executeTakeFirst(); + + if (row) { + try { + const parsed: unknown = JSON.parse(row.fields); + if (Array.isArray(parsed) && parsed.length === 1 && typeof parsed[0] === "string") { + return parsed[0]; + } + } catch { + // fall through to name parsing + } + return undefined; + } + + // Fallback: uidx_plugin___ + const prefix = `uidx_plugin_${pluginId}_${collection}_`; + if (indexName.startsWith(prefix)) { + const field = indexName.slice(prefix.length); + if (SAFE_FIELD_NAME_RE.test(field)) return field; + } + return undefined; +} + +/** + * Build the `INSERT … ON CONFLICT (plugin_id, collection, id) DO NOTHING` + * statement for a single insert-once. Extracted from `insert` so the standalone + * method and the batch executor share ONE builder (pure — no execution here). + */ +function buildInsertQuery( + db: Kysely, + pluginId: string, + collection: string, + id: string, + data: unknown, +): InsertQueryBuilder< + Database, + "_plugin_storage", + { numInsertedOrUpdatedRows: bigint | undefined } +> { + const now = new Date().toISOString(); + const jsonData = JSON.stringify(data); + return db + .insertInto("_plugin_storage") + .values({ + plugin_id: pluginId, + collection, + id, + data: jsonData, + created_at: now, + updated_at: now, + }) + .onConflict((oc) => oc.columns(["plugin_id", "collection", "id"]).doNothing()); +} + +/** + * Normalize the `delta` map into signed integer entries, enforcing integer-only + * and the "not in both `set` and `delta`" rule at runtime. Extracted verbatim + * from `updateIf` so the standalone method and the batch reject floats / + * both-in-set-and-delta identically (a `TypeError` — a programmer error). + */ +function normalizeDeltaEntries( + delta: { [K in keyof T]?: NumericDelta } | undefined, + setEntries: Array<[string, unknown]>, +): Array<[string, number]> { + const deltaEntries: Array<[string, number]> = []; + if (!delta || Object.keys(delta).length === 0) return deltaEntries; + + const setFieldSet = new Set(setEntries.map(([field]) => field)); + for (const [field, spec] of Object.entries(delta)) { + if (spec === undefined) continue; + if (setFieldSet.has(field)) { + throw new Error(`updateIf: field "${field}" appears in both \`set\` and \`delta\`.`); + } + if (!isDeltaLike(spec)) { + throw new TypeError( + `updateIf: delta for "${field}" must be exactly one of { inc: number } or { dec: number }.`, + ); + } + const { inc, dec } = spec; + let signed: number; + if (typeof inc === "number" && typeof dec !== "number") { + signed = inc; + } else if (typeof dec === "number" && typeof inc !== "number") { + signed = -dec; + } else { + // Both present or neither present/numeric → ambiguous or invalid. + throw new TypeError( + `updateIf: delta for "${field}" must be exactly one of { inc: number } or { dec: number }.`, + ); + } + if (!Number.isInteger(signed)) { + throw new TypeError( + `updateIf: delta for "${field}" must be an integer (got ${String(inc ?? dec)}).`, + ); + } + deltaEntries.push([field, signed]); + } + return deltaEntries; +} + +/** + * Build the guarded `UPDATE … RETURNING data` statement for a single `updateIf`. + * Returns `{ query, empty }` where `empty` flags the empty-`in:[]` short-circuit + * (matches nothing → `applied:false`) so the caller never emits invalid + * `IN ()`. Extracted from `updateIf`; the delta validation runs through + * {@link normalizeDeltaEntries}. Pure — no execution here. + */ +function buildUpdateIfQuery( + db: Kysely, + pluginId: string, + collection: string, + id: string, + args: UpdateIfArgs, +): + | { empty: true; query: null } + | { + empty: false; + query: UpdateQueryBuilder; + } { + const { where, set, delta } = args; + + const setEntries: Array<[string, unknown]> = set ? Object.entries(set) : []; + const hasSet = setEntries.length > 0; + const hasDelta = delta !== undefined && Object.keys(delta).length > 0; + + if (!hasSet && !hasDelta) { + throw new Error("updateIf requires at least one of `set` or `delta`."); + } + + const deltaEntries = normalizeDeltaEntries(delta, setEntries); + + // Defensive empty-`in` guard: an empty `in: []` matches nothing. The shared + // where-translation would emit invalid `IN ()`; short-circuit to a no-op + // (matches nothing → applied:false) BEFORE building any SQL. + for (const value of Object.values(where)) { + if (isInFilter(value) && value.in.length === 0) { + return { empty: true, query: null }; + } + } + + const now = new Date().toISOString(); + const dataExpr = pluginDataWriteExpr(db, setEntries, deltaEntries); + + let query = db + .updateTable("_plugin_storage") + .set({ data: dataExpr, updated_at: now }) + .where("plugin_id", "=", pluginId) + .where("collection", "=", collection) + .where("id", "=", id); + + const whereResult = buildWhereClause(db, where); + if (whereResult.sql) { + query = query.where(buildRawWhereExpression(whereResult)); + } + + return { empty: false, query: query.returning("data") }; +} + /** * Plugin Storage Repository * @@ -372,30 +571,27 @@ export class PluginStorageRepository implements StorageCollection { - const now = new Date().toISOString(); - const jsonData = JSON.stringify(data); - try { - const result = await this.db - .insertInto("_plugin_storage") - .values({ - plugin_id: this.pluginId, - collection: this.collection, - id, - data: jsonData, - created_at: now, - updated_at: now, - }) - .onConflict((oc) => oc.columns(["plugin_id", "collection", "id"]).doNothing()) - .executeTakeFirst(); + const result = await buildInsertQuery( + this.db, + this.pluginId, + this.collection, + id, + data, + ).executeTakeFirst(); - const inserted = (result.numInsertedOrUpdatedRows ?? 0n) > 0n; + const inserted = (result?.numInsertedOrUpdatedRows ?? 0n) > 0n; if (inserted) return { inserted: true }; return { inserted: false, reason: "exists" }; } catch (error) { const classified = classifyUniqueViolation(error); if (!classified) throw error; - const conflictField = await this.recoverConflictField(classified.indexName); + const conflictField = await recoverConflictField( + this.db, + this.pluginId, + this.collection, + classified.indexName, + ); return conflictField ? { inserted: false, reason: "unique_violation", conflictField } : { inserted: false, reason: "unique_violation" }; @@ -417,119 +613,465 @@ export class PluginStorageRepository implements StorageCollection): Promise> { - const { where, set, delta } = args; + const built = buildUpdateIfQuery(this.db, this.pluginId, this.collection, id, args); + if (built.empty) return { applied: false }; - const setEntries: Array<[string, unknown]> = set ? Object.entries(set) : []; - const hasSet = setEntries.length > 0; - const hasDelta = delta !== undefined && Object.keys(delta).length > 0; + const row = await built.query.executeTakeFirst(); + if (!row) return { applied: false }; + // JSON.parse returns any; it flows into the T-typed `data` field directly. + const data: T = JSON.parse(row.data); + return { applied: true, data }; + } +} + +// ============================================================================= +// Atomic multi-document batch (ctx.storage.batch) +// ============================================================================= + +/** + * Internal sentinel thrown inside the transaction to force a rollback with a + * reported outcome. Caught OUTSIDE the transaction and turned into a + * `{ applied:false, … }` result (never surfaced to callers). `indexName` / + * `collection` are carried for `unique_violation` so `conflictField` can be + * recovered on the pool connection AFTER the transaction rolls back (a + * constraint error poisons the Postgres transaction, so the lookup must not run + * inside it). + */ +class BatchAbort extends Error { + constructor( + readonly failedIndex: number, + readonly reason: BatchFailureReason, + readonly indexName?: string, + readonly abortCollection?: string, + ) { + super(`batch aborted at op ${failedIndex}: ${reason}`); + this.name = "BatchAbort"; + } +} - if (!hasSet && !hasDelta) { - throw new Error("updateIf requires at least one of `set` or `delta`."); +/** True for an `updateIf` op whose guard contains an empty `in: []` (matches nothing). */ +function hasEmptyInGuard(op: BatchOp): boolean { + if (op.op !== "updateIf") return false; + for (const value of Object.values(op.where)) { + if (isInFilter(value) && value.in.length === 0) return true; + } + return false; +} + +/** + * Upper bound on ops per batch. + * + * The D1 executor emits up to TWO statements per guarded op (the write + its + * zero-rows assertion), so this caps a batch at ≤ ~2× statements. 50 keeps the + * worst case (~100 statements) comfortably inside Cloudflare D1's per-`batch()` + * budget and far under the SQLite/D1 bound-variable ceiling (32766) — each op's + * compiled statement binds only a handful of params. It is also generous for the + * intended coupled-write use case (a reserve is 2–3 ops). An over-limit batch is + * a programmer error (THROW), consistent with the other malformed-ops throws, + * rather than an opaque D1 failure deep in `env.DB.batch()`. + */ +const MAX_BATCH_OPS = 50; + +/** + * Validate the ops array shape up front (BEFORE any write), so a malformed op is + * a THROW with no partial commit — mirroring the single-op `insert`/`updateIf` + * discipline. Guard/uniqueness OUTCOMES are never validated here (they are + * reported, not thrown); only programmer errors throw. + */ +function assertBatchOpsValid(ops: BatchOp[]): void { + if (!Array.isArray(ops) || ops.length === 0) { + throw new Error("batch requires a non-empty array of ops."); + } + if (ops.length > MAX_BATCH_OPS) { + throw new Error(`batch exceeds the maximum of ${MAX_BATCH_OPS} ops (got ${ops.length}).`); + } + for (let i = 0; i < ops.length; i++) { + const op = ops[i]; + if (op === null || typeof op !== "object") { + throw new Error(`batch op ${i} must be an object.`); } + if (op.op !== "insert" && op.op !== "updateIf") { + throw new Error(`batch op ${i} has unknown op "${String((op as { op?: unknown }).op)}".`); + } + validateIdentifier(op.collection, `batch op ${i} collection name`); + if (typeof op.id !== "string" || op.id.length === 0) { + throw new Error(`batch op ${i} requires a non-empty string id.`); + } + if (op.op === "insert") { + // `data` is required — an `insert` with no data would otherwise fall + // through to a NOT NULL violation deep in the write. Fail up front, + // matching the insert/updateIf validation discipline. + if (op.data === undefined) { + throw new Error(`batch op ${i} (insert) requires \`data\`.`); + } + } else { + if (typeof op.where !== "object" || op.where === null) { + throw new Error(`batch op ${i} (updateIf) requires an object \`where\`.`); + } + const setEntries: Array<[string, unknown]> = op.set ? Object.entries(op.set) : []; + const hasSet = setEntries.length > 0; + const hasDelta = op.delta !== undefined && Object.keys(op.delta).length > 0; + if (!hasSet && !hasDelta) { + throw new Error(`batch op ${i} (updateIf) requires at least one of \`set\` or \`delta\`.`); + } + // Throws on float delta / field in both set & delta (programmer error). + normalizeDeltaEntries(op.delta, setEntries); + } + } +} - // Build the signed integer deltas, enforcing integer-only at runtime. - const deltaEntries: Array<[string, number]> = []; - if (hasDelta) { - const setFieldSet = new Set(setEntries.map(([field]) => field)); - for (const [field, spec] of Object.entries(delta)) { - if (spec === undefined) continue; - if (setFieldSet.has(field)) { - throw new Error(`updateIf: field "${field}" appears in both \`set\` and \`delta\`.`); - } - if (!isDeltaLike(spec)) { - throw new TypeError( - `updateIf: delta for "${field}" must be exactly one of { inc: number } or { dec: number }.`, - ); - } - const { inc, dec } = spec; - let signed: number; - if (typeof inc === "number" && typeof dec !== "number") { - signed = inc; - } else if (typeof dec === "number" && typeof inc !== "number") { - signed = -dec; +/** + * Apply several conditional writes atomically on the pg / better-sqlite3 path. + * + * Runs every op inside ONE `db.transaction().execute(...)` (a real transaction — + * NOT the `withTransaction` fallback, which degrades to non-atomic on a + * no-transaction backend and would break all-or-nothing). Commits iff every op's + * guard passes; a failing op throws {@link BatchAbort} to roll back the whole + * transaction, and the outer catch turns that into `{ applied:false, … }`. A raw + * DB error re-throws. + * + * The D1 production sandbox path does NOT use this (D1 has no interactive + * transactions) — it uses {@link applyPluginStorageBatchD1}. + */ +export async function applyPluginStorageBatch( + db: Kysely, + pluginId: string, + ops: BatchOp[], +): Promise { + assertBatchOpsValid(ops); + + try { + const results = await db.transaction().execute(async (trx) => { + const out: BatchOpResult[] = []; + for (const [i, op] of ops.entries()) { + if (op.op === "insert") { + let inserted: boolean; + try { + const res = await buildInsertQuery( + trx, + pluginId, + op.collection, + op.id, + op.data, + ).executeTakeFirst(); + inserted = (res?.numInsertedOrUpdatedRows ?? 0n) > 0n; + } catch (error) { + const classified = classifyUniqueViolation(error); + if (!classified) throw error; // raw DB error — never swallowed + throw new BatchAbort(i, "unique_violation", classified.indexName, op.collection); + } + if (inserted) { + out.push({ op: "insert", inserted: true }); + } else if (op.ifNotExists) { + // Satisfied no-op — the row already exists, let the batch proceed. + out.push({ op: "insert", inserted: false, reason: "exists" }); + } else { + throw new BatchAbort(i, "exists"); + } } else { - // Both present or neither present/numeric → ambiguous or invalid. - throw new TypeError( - `updateIf: delta for "${field}" must be exactly one of { inc: number } or { dec: number }.`, - ); - } - if (!Number.isInteger(signed)) { - throw new TypeError( - `updateIf: delta for "${field}" must be an integer (got ${String(inc ?? dec)}).`, - ); + const built = buildUpdateIfQuery(trx, pluginId, op.collection, op.id, { + where: op.where, + set: op.set, + delta: op.delta, + }); + if (built.empty) throw new BatchAbort(i, "guard_failed"); + const row = await built.query.executeTakeFirst(); + if (!row) throw new BatchAbort(i, "guard_failed"); + out.push({ op: "updateIf", applied: true, data: JSON.parse(row.data) }); } - deltaEntries.push([field, signed]); } - } - - // Defensive empty-`in` guard: an empty `in: []` matches nothing. The shared - // where-translation would emit invalid `IN ()`; short-circuit to a no-op - // (matches nothing → applied:false) BEFORE building any SQL. - for (const value of Object.values(where)) { - if (isInFilter(value) && value.in.length === 0) { - return { applied: false }; + return out; + }); + return { applied: true, results }; + } catch (err) { + if (err instanceof BatchAbort) { + if (err.reason === "unique_violation" && err.indexName && err.abortCollection) { + const conflictField = await recoverConflictField( + db, + pluginId, + err.abortCollection, + err.indexName, + ); + return conflictField + ? { applied: false, failedIndex: err.failedIndex, reason: err.reason, conflictField } + : { applied: false, failedIndex: err.failedIndex, reason: err.reason }; } + return { applied: false, failedIndex: err.failedIndex, reason: err.reason }; } + throw err; + } +} - const now = new Date().toISOString(); - const dataExpr = pluginDataWriteExpr(this.db, setEntries, deltaEntries); +// ── D1 executor (raw env.DB.batch) ──────────────────────────────────────── +// +// Cloudflare D1 has no interactive transactions; the only atomic primitive is +// `env.DB.batch([...])`, an implicit transaction that rolls back ONLY when a +// statement ERRORS. A guarded `UPDATE … WHERE guard` matching 0 rows does not +// error, so we interleave an assertion after each guarded write that raises when +// `changes() = 0`, forcing the whole batch to roll back. Verified on real D1 +// (see packages/cloudflare/tests/sandbox/*batch* + the probe): `changes()` +// carries across statements in one batch; `abs(-9223372036854775808)` overflows +// → SQLITE_ERROR → whole-batch rollback. - let query = this.db - .updateTable("_plugin_storage") - .set({ data: dataExpr, updated_at: now }) - .where("plugin_id", "=", this.pluginId) - .where("collection", "=", this.collection) - .where("id", "=", id); +/** Minimal structural view of the raw `D1Database` binding (avoids a hard dep on `@cloudflare/workers-types` in core). */ +export interface D1BatchBinding { + prepare(query: string): D1BatchStatement; + batch(statements: D1BatchStatement[]): Promise; +} +export interface D1BatchStatement { + bind(...values: unknown[]): D1BatchStatement; + first>(colName?: string): Promise; +} +interface D1BatchRow { + results?: Array>; + meta?: { changes?: number }; +} - const whereResult = buildWhereClause(this.db, where); +/** + * `changes() = 0` → integer-overflow → `SQLITE_ERROR`, rolling back the whole + * `env.DB.batch`. No JSON1 dependency. (`json('')` "malformed JSON" is a proven + * fallback if a future D1 build changes overflow behaviour.) + */ +const D1_ASSERT_APPLIED = "SELECT CASE WHEN changes()=0 THEN abs(-9223372036854775808) ELSE 1 END"; +/** Unconditional error (no `changes()` dependency) used to force rollback for an unsatisfiable guard. */ +const D1_ASSERT_ALWAYS = "SELECT abs(-9223372036854775808)"; + +/** + * Compile-only Kysely (SQLite dialect, DummyDriver — never executes) used to + * translate the shared query builders into `{ sql, parameters }` for D1. A + * SQLite adapter makes `pluginDataWriteExpr` / `buildWhereClause` emit the + * SQLite JSON forms (`json_set` / `json_extract`), NOT the Postgres `jsonb` + * syntax — exactly what D1 needs. + */ +let compileDb: Kysely | null = null; +function getCompileDb(): Kysely { + compileDb ??= new Kysely({ + dialect: { + createAdapter: () => new SqliteAdapter(), + createDriver: () => new DummyDriver(), + createQueryCompiler: () => new SqliteQueryCompiler(), + createIntrospector: (d) => new SqliteIntrospector(d), + }, + }); + return compileDb; +} + +/** Compile an `EXISTS(SELECT 1 … WHERE pk [AND guard])` diagnosis probe (parameterized — reuses buildWhereClause). */ +function compileExistsProbe( + pluginId: string, + collection: string, + id: string, + where?: WhereClause, +): { sql: string; parameters: unknown[] } { + const parts = [ + "SELECT EXISTS(SELECT 1 FROM _plugin_storage WHERE plugin_id = ? AND collection = ? AND id = ?", + ]; + const parameters: unknown[] = [pluginId, collection, id]; + if (where && Object.keys(where).length > 0) { + const whereResult = buildWhereClause(getCompileDb(), where); if (whereResult.sql) { - query = query.where(buildRawWhereExpression(whereResult)); + parts.push(` AND ${whereResult.sql}`); + parameters.push(...whereResult.params); } - - const row = await query.returning("data").executeTakeFirst(); - if (!row) return { applied: false }; - // JSON.parse returns any; it flows into the T-typed `data` field directly. - const data: T = JSON.parse(row.data); - return { applied: true, data }; } + parts.push(") AS e"); + return { sql: parts.join(""), parameters }; +} - /** - * Best-effort recovery of the single field behind a unique-index violation. - * Prefers the `_plugin_indexes` tracking row (authoritative field list); - * falls back to parsing the `generateIndexName` format. Composite indexes - * yield `undefined`. - */ - private async recoverConflictField(indexName?: string): Promise { - if (!indexName) return undefined; - - const row = await this.db - .selectFrom("_plugin_indexes") - .select("fields") - .where("plugin_id", "=", this.pluginId) - .where("collection", "=", this.collection) - .where("index_name", "=", indexName) - .executeTakeFirst(); +/** Read a truthy `EXISTS(...)` result column from a D1 `.first()` row. */ +function existsRowTrue(row: Record | null): boolean { + if (!row) return false; + const e = row.e; + return e === 1 || e === true || e === "1"; +} - if (row) { - try { - const parsed: unknown = JSON.parse(row.fields); - if (Array.isArray(parsed) && parsed.length === 1 && typeof parsed[0] === "string") { - return parsed[0]; +/** + * Failure-path diagnosis (D1 only). The atomic guarantee already held (nothing + * committed); this read-only pass re-derives `failedIndex` / `reason` for the + * report. It NEVER throws or loops: under a concurrent restock every guard may + * be satisfiable again, in which case we return a well-formed generic fallback. + * Best-effort by construction — the committed state is always correct; only the + * report can drift (TOCTOU between rollback and diagnosis). + */ +async function diagnoseD1Failure( + d1: D1BatchBinding, + pluginId: string, + ops: BatchOp[], + classified: { indexName?: string } | null, +): Promise { + for (const [i, op] of ops.entries()) { + if (op.op === "insert") { + const probe = compileExistsProbe(pluginId, op.collection, op.id); + const row = await d1 + .prepare(probe.sql) + .bind(...probe.parameters) + .first(); + const exists = existsRowTrue(row); + if (op.ifNotExists) { + // ifNotExists never fails on a PK conflict; only a unique violation can. + if (classified) { + const conflictField = await recoverConflictFieldD1( + d1, + pluginId, + op.collection, + classified, + ); + return conflictField + ? { applied: false, failedIndex: i, reason: "unique_violation", conflictField } + : { applied: false, failedIndex: i, reason: "unique_violation" }; } - } catch { - // fall through to name parsing + continue; + } + if (exists) return { applied: false, failedIndex: i, reason: "exists" }; + // PK absent but the insert still failed → a declared unique violation. + if (classified) { + const conflictField = await recoverConflictFieldD1(d1, pluginId, op.collection, classified); + return conflictField + ? { applied: false, failedIndex: i, reason: "unique_violation", conflictField } + : { applied: false, failedIndex: i, reason: "unique_violation" }; + } + } else { + if (hasEmptyInGuard(op)) return { applied: false, failedIndex: i, reason: "guard_failed" }; + const probe = compileExistsProbe(pluginId, op.collection, op.id, op.where); + const row = await d1 + .prepare(probe.sql) + .bind(...probe.parameters) + .first(); + if (!existsRowTrue(row)) return { applied: false, failedIndex: i, reason: "guard_failed" }; + } + } + // No op currently fails (a concurrent restock re-satisfied every guard). + // Return a well-formed best-effort fallback rather than throwing/looping. + if (classified) { + for (const [i, op] of ops.entries()) { + if (op.op === "insert") { + const conflictField = await recoverConflictFieldD1(d1, pluginId, op.collection, classified); + return conflictField + ? { applied: false, failedIndex: i, reason: "unique_violation", conflictField } + : { applied: false, failedIndex: i, reason: "unique_violation" }; } - return undefined; } + } + return { applied: false, failedIndex: 0, reason: "guard_failed" }; +} - // Fallback: uidx_plugin___ - const prefix = `uidx_plugin_${this.pluginId}_${this.collection}_`; - if (indexName.startsWith(prefix)) { - const field = indexName.slice(prefix.length); - if (SAFE_FIELD_NAME_RE.test(field)) return field; +/** conflictField recovery over the raw D1 binding (mirrors {@link recoverConflictField}). */ +async function recoverConflictFieldD1( + d1: D1BatchBinding, + pluginId: string, + collection: string, + classified: { indexName?: string }, +): Promise { + const indexName = classified.indexName; + if (!indexName) return undefined; + const row = await d1 + .prepare( + "SELECT fields FROM _plugin_indexes WHERE plugin_id = ? AND collection = ? AND index_name = ?", + ) + .bind(pluginId, collection, indexName) + .first<{ fields?: string }>(); + if (row && typeof row.fields === "string") { + try { + const parsed: unknown = JSON.parse(row.fields); + if (Array.isArray(parsed) && parsed.length === 1 && typeof parsed[0] === "string") { + return parsed[0]; + } + } catch { + // fall through } return undefined; } + const prefix = `uidx_plugin_${pluginId}_${collection}_`; + if (indexName.startsWith(prefix)) { + const field = indexName.slice(prefix.length); + if (SAFE_FIELD_NAME_RE.test(field)) return field; + } + return undefined; +} + +/** + * Apply an atomic batch on the Cloudflare D1 path via raw `env.DB.batch()` with + * interleaved zero-rows assertions (§5.2). Compiles each op's statement through + * the shared SQLite-dialect builders, submits `[op0, assert0, op1, assert1, …]` + * in one `batch()` (an implicit transaction), and on failure runs a read-only + * {@link diagnoseD1Failure} pass to report `failedIndex` / `reason`. + * + * Returns the SAME `BatchResult` shape as {@link applyPluginStorageBatch}. + */ +export async function applyPluginStorageBatchD1( + d1: D1BatchBinding, + pluginId: string, + ops: BatchOp[], +): Promise { + assertBatchOpsValid(ops); + const cdb = getCompileDb(); + + const statements: D1BatchStatement[] = []; + // For each op, remember where its main statement lands so we can read + // RETURNING data / changes() back from the batch results in ops order. + const positions: Array< + { kind: "insert"; pos: number; ifNotExists: boolean } | { kind: "updateIf"; pos: number } + > = []; + + for (const op of ops) { + if (op.op === "insert") { + const compiled = buildInsertQuery(cdb, pluginId, op.collection, op.id, op.data).compile(); + const pos = statements.length; + statements.push(d1.prepare(compiled.sql).bind(...compiled.parameters)); + positions.push({ kind: "insert", pos, ifNotExists: op.ifNotExists === true }); + // A non-ifNotExists insert must affect a row; assert it did (0 rows ⇒ + // PK conflict ⇒ roll back). An ifNotExists insert gets NO assertion + // (0 rows is a satisfied no-op). + if (!op.ifNotExists) statements.push(d1.prepare(D1_ASSERT_APPLIED)); + } else { + const built = buildUpdateIfQuery(cdb, pluginId, op.collection, op.id, { + where: op.where, + set: op.set, + delta: op.delta, + }); + if (built.empty) { + // Unsatisfiable guard (empty in:[]): force a whole-batch rollback. + statements.push(d1.prepare(D1_ASSERT_ALWAYS)); + positions.push({ kind: "updateIf", pos: -1 }); + } else { + const compiled = built.query.compile(); + const pos = statements.length; + statements.push(d1.prepare(compiled.sql).bind(...compiled.parameters)); + positions.push({ kind: "updateIf", pos }); + statements.push(d1.prepare(D1_ASSERT_APPLIED)); + } + } + } + + let batchRows: D1BatchRow[]; + try { + batchRows = await d1.batch(statements); + } catch (err) { + return diagnoseD1Failure(d1, pluginId, ops, classifyUniqueViolation(err)); + } + + // Success: every guard passed. Build results[] in ops order. + const results: BatchOpResult[] = []; + for (const meta of positions) { + const row = meta.pos >= 0 ? batchRows[meta.pos] : undefined; + if (meta.kind === "insert") { + const changed = (row?.meta?.changes ?? 0) > 0; + if (changed) { + results.push({ op: "insert", inserted: true }); + } else { + // Committed with 0 rows ⇒ satisfied ifNotExists no-op (row existed). + results.push({ op: "insert", inserted: false, reason: "exists" }); + } + } else { + const data = row?.results?.[0]?.data; + results.push({ + op: "updateIf", + applied: true, + data: typeof data === "string" ? JSON.parse(data) : data, + }); + } + } + return { applied: true, results }; } /** @@ -542,7 +1084,7 @@ export function createPluginStorageAccessor( string, { indexes: Array; uniqueIndexes?: Array } >, -): Record { +): StorageAccess { const accessor: Record = {}; for (const [collectionName, config] of Object.entries(storageConfig)) { @@ -555,7 +1097,9 @@ export function createPluginStorageAccessor( ); } - return accessor; + return Object.assign(accessor, { + batch: (ops: BatchOp[]) => applyPluginStorageBatch(db, pluginId, ops), + }); } /** diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 488fe2789f..672ef015fa 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -13,11 +13,14 @@ export { ContentRepository, MediaRepository, PluginStorageRepository, + applyPluginStorageBatch, + applyPluginStorageBatchD1, UserRepository, OptionsRepository, EmDashValidationError, InvalidCursorError, } from "./database/repositories/index.js"; +export type { D1BatchBinding, D1BatchStatement } from "./database/repositories/index.js"; export type { ContentItem, ContentSeo, @@ -242,6 +245,17 @@ export type { PluginContext, PluginStorageConfig, StorageCollection, + StorageAccess, + BatchOp, + BatchInsertOp, + BatchUpdateIfOp, + BatchOpResult, + BatchResult, + BatchFailureReason, + InsertResult, + NumericDelta, + UpdateIfArgs, + UpdateIfResult, KVAccess, ContentAccess, MediaAccess, diff --git a/packages/core/src/plugins/context.ts b/packages/core/src/plugins/context.ts index 044fd69012..1aa6a70cba 100644 --- a/packages/core/src/plugins/context.ts +++ b/packages/core/src/plugins/context.ts @@ -11,7 +11,10 @@ import { ulid } from "ulidx"; import { ContentRepository } from "../database/repositories/content.js"; import { MediaRepository } from "../database/repositories/media.js"; import { OptionsRepository } from "../database/repositories/options.js"; -import { PluginStorageRepository } from "../database/repositories/plugin-storage.js"; +import { + PluginStorageRepository, + applyPluginStorageBatch, +} from "../database/repositories/plugin-storage.js"; import { SeoRepository } from "../database/repositories/seo.js"; import { TaxonomyRepository, type Taxonomy } from "../database/repositories/taxonomy.js"; import { UserRepository } from "../database/repositories/user.js"; @@ -33,6 +36,8 @@ import type { PluginContext, PluginStorageConfig, StorageCollection, + StorageAccess, + BatchOp, KVAccess, CronAccess, EmailAccess, @@ -151,7 +156,7 @@ export function createStorageAccess( db: Kysely, pluginId: string, storageConfig: T, -): Record { +): StorageAccess { const storage: Record = {}; for (const [collectionName, config] of Object.entries(storageConfig)) { @@ -159,7 +164,12 @@ export function createStorageAccess( storage[collectionName] = createStorageCollection(db, pluginId, collectionName, allIndexes); } - return storage; + // Attach the cross-collection atomic batch primitive alongside the + // per-collection accessors. `Object.assign` yields exactly + // `Record & { batch }` = StorageAccess. + return Object.assign(storage, { + batch: (ops: BatchOp[]) => applyPluginStorageBatch(db, pluginId, ops), + }); } // ============================================================================= diff --git a/packages/core/src/plugins/index.ts b/packages/core/src/plugins/index.ts index 75b5879a4d..4089b3faa0 100644 --- a/packages/core/src/plugins/index.ts +++ b/packages/core/src/plugins/index.ts @@ -104,6 +104,17 @@ export type { // Context APIs PluginContext, StorageCollection, + StorageAccess, + BatchOp, + BatchInsertOp, + BatchUpdateIfOp, + BatchOpResult, + BatchResult, + BatchFailureReason, + InsertResult, + NumericDelta, + UpdateIfArgs, + UpdateIfResult, KVAccess, ContentAccess, ContentAccessWithWrite, diff --git a/packages/core/src/plugins/storage-indexes.ts b/packages/core/src/plugins/storage-indexes.ts index 43ec2e4eb5..3b6c6606a3 100644 --- a/packages/core/src/plugins/storage-indexes.ts +++ b/packages/core/src/plugins/storage-indexes.ts @@ -17,6 +17,29 @@ import { validatePluginIdentifier, } from "../database/validate.js"; +/** + * Collection names that collide with a method on the `ctx.storage` access + * object (the {@link StorageAccess} intersection type). `"batch"` matches the + * collection-name regex `/^[a-z][a-z0-9_]*$/`, so a plugin declaring a + * collection named `batch` would be silently SHADOWED by `ctx.storage.batch()`. + * Reject such names at declaration time. + */ +const RESERVED_COLLECTION_NAMES = new Set(["batch"]); + +/** + * Validate a plugin storage collection name: a safe identifier AND not a + * reserved access-object method name. Throws with a clear message otherwise. + */ +export function assertCollectionNameAllowed(collection: string): void { + validateIdentifier(collection, "collection name"); + if (RESERVED_COLLECTION_NAMES.has(collection)) { + throw new Error( + `Storage collection name "${collection}" is reserved — it collides with the ` + + `ctx.storage.${collection}() method. Rename the collection.`, + ); + } +} + /** * Generate a deterministic index name. * Unique indexes use a `uidx_` prefix to avoid collisions with regular indexes on the same fields. @@ -120,6 +143,9 @@ export async function createStorageIndexes( created: string[]; errors: Array<{ index: string; error: string }>; }> { + // Reject reserved / malformed collection names at declaration (install) time. + assertCollectionNameAllowed(collection); + const normalized = normalizeIndexes(indexes); const uniqueNormalized = options?.uniqueIndexes ? normalizeIndexes(options.uniqueIndexes) : []; const uniqueSet = new Set(uniqueNormalized.map((f) => f.join(","))); diff --git a/packages/core/src/plugins/types.ts b/packages/core/src/plugins/types.ts index bef790cbc6..2952f98844 100644 --- a/packages/core/src/plugins/types.ts +++ b/packages/core/src/plugins/types.ts @@ -190,6 +190,82 @@ export interface UpdateIfArgs { */ export type UpdateIfResult = { applied: true; data: T } | { applied: false }; +/** + * A single operation in an atomic {@link StorageAccess.batch}. + * + * `insert` and `updateIf` reuse the exact `where` / `set` / `delta` shapes of + * the single-document {@link StorageCollection.insert} / `updateIf` primitives, + * so a `{ dec: n }` delta can never be mistaken for a wholesale set and the + * per-op float / both-in-set-and-delta guards apply identically. + */ +export interface BatchInsertOp { + op: "insert"; + collection: string; + id: string; + data: T; + /** + * Idempotent-claim mode. When `true`, an already-present row (PK conflict, + * `reason: "exists"`) does NOT fail the batch — the op is a satisfied no-op + * and the batch proceeds. A real unique-index violation on a non-`id` field + * STILL fails the batch. Default `false` (exists ⇒ the whole batch fails). + */ + ifNotExists?: boolean; +} + +export interface BatchUpdateIfOp { + op: "updateIf"; + collection: string; + id: string; + /** Guard evaluated in-SQL; identical semantics to {@link UpdateIfArgs.where}. */ + where: WhereClause; + /** Wholesale field values merged into the stored JSON document. */ + set?: Partial; + /** Per-field integer deltas applied in-SQL (`COALESCE(base, 0) ± n`). */ + delta?: { [K in keyof T]?: NumericDelta }; +} + +/** A single op in a {@link StorageAccess.batch} — insert or guarded update. */ +export type BatchOp = BatchInsertOp | BatchUpdateIfOp; + +/** + * Per-op success result in a committed batch. `results[i]` corresponds to + * `ops[i]`. + */ +export type BatchOpResult = + | { op: "insert"; inserted: true } + | { op: "insert"; inserted: false; reason: "exists" } + | { op: "updateIf"; applied: true; data: unknown }; + +/** + * Why a batch rolled back. + * - `guard_failed` — an `updateIf` op matched 0 rows (row absent OR guard false). + * - `exists` — an `insert` op hit a PK conflict and `ifNotExists` was not set. + * - `unique_violation` — an `insert` op hit a declared unique index on a non-`id` + * field. + */ +export type BatchFailureReason = "guard_failed" | "exists" | "unique_violation"; + +/** + * Result of an atomic {@link StorageAccess.batch}. + * + * Commits **iff** every op's guard passes; otherwise the WHOLE batch rolls back + * and reports the first failing op via `failedIndex` + `reason`. `conflictField` + * is populated for `unique_violation` (best-effort — single-field indexes). + * + * Guard / uniqueness *outcomes* are reported here, never thrown. Malformed ops + * (float delta, field in both `set` & `delta`, unknown op, empty ops array, + * `updateIf` with neither `set` nor `delta`) THROW — they are programmer errors. + * A raw DB error re-throws (never swallowed). + */ +export type BatchResult = + | { applied: true; results: BatchOpResult[] } + | { + applied: false; + failedIndex: number; + reason: BatchFailureReason; + conflictField?: string; + }; + /** * Storage collection interface - the API exposed to plugins * No async iterators - all operations return promises with pagination @@ -228,6 +304,27 @@ export interface StorageCollection { updateIf(id: string, args: UpdateIfArgs): Promise>; } +/** + * The storage **access** object handed to plugins as `ctx.storage`. + * + * It keeps the existing per-collection index signature (`ctx.storage.`) + * AND adds the cross-collection {@link StorageAccess.batch} primitive. Modeled + * as an intersection (`Record & { batch }`) rather + * than a single interface with a conflicting index signature — the same + * mixed-map shape the codebase uses elsewhere. A plugin collection literally + * named `batch` is rejected at declaration time (it would be shadowed by this + * method). + */ +export type StorageAccess = Record & { + /** + * Apply several conditional writes (`insert` / `updateIf`) across multiple + * documents and collections **all-or-nothing**: commits only if EVERY op's + * guard passes, otherwise rolls back the whole batch and reports which op + * failed. Atomic on Postgres, SQLite (better-sqlite3), and Cloudflare D1. + */ + batch(ops: BatchOp[]): Promise; +}; + /** * Plugin storage context - typed based on declared collections */ @@ -546,8 +643,16 @@ export interface PluginContext; + /** + * Storage collections plus the atomic {@link StorageAccess.batch} primitive. + * Per-collection accessors (`ctx.storage.`) keep their declared-key + * typing; `batch` is additive. Structurally equivalent to {@link StorageAccess} + * (what the runtime builds) while preserving `TStorage` key typing for native + * plugins. + */ + storage: PluginStorage & { + batch(ops: BatchOp[]): Promise; + }; /** Key-value store for config and state */ kv: KVAccess; diff --git a/packages/core/tests/integration/plugins/storage-batch-no-oversell.test.ts b/packages/core/tests/integration/plugins/storage-batch-no-oversell.test.ts new file mode 100644 index 0000000000..0438f9f4e2 --- /dev/null +++ b/packages/core/tests/integration/plugins/storage-batch-no-oversell.test.ts @@ -0,0 +1,124 @@ +/** + * NO OVERSELL under concurrency — for the COUPLED atomic batch. + * + * Seed stock M, fire N > M concurrent reserve batches (each: claim insert ∧ + * guarded decrement ∧ flip) and assert exactly M commit, final on_hand is 0, + * exactly M reservations are `held`, and NO reservation is `held` without its + * decrement (the invariant `held ⟺ a durable decrement`). + * + * As with the single-op no-oversell test: + * - **better-sqlite3 [sqlite]** serializes writes in-process → proves the batch + * SQL / transaction is CORRECT, but not the race. + * - **Postgres** is the true concurrent race: N connections contend on the same + * inventory row; the guarded decrement inside each transaction serializes so + * exactly M observe `on_hand >= 1`. + */ + +import type { Kysely } from "kysely"; +import { it, expect, beforeEach, afterEach } from "vitest"; + +import { + PluginStorageRepository, + applyPluginStorageBatch, +} from "../../../src/database/repositories/plugin-storage.js"; +import type { Database } from "../../../src/database/types.js"; +import { createStorageIndexes } from "../../../src/plugins/storage-indexes.js"; +import type { BatchOp } from "../../../src/plugins/types.js"; +import { + describeEachDialect, + setupForDialect, + teardownForDialect, + type DialectTestContext, +} from "../../utils/test-db.js"; + +interface Inventory { + on_hand: number; +} +interface Reservation { + state: string; + sku: string; + qty: number; + idempotency_key: string; +} + +const PLUGIN = "shop"; + +describeEachDialect("Plugin storage atomic batch — no oversell", (dialect) => { + let ctx: DialectTestContext; + let db: Kysely; + + beforeEach(async () => { + ctx = await setupForDialect(dialect); + db = ctx.db; + await createStorageIndexes(db, PLUGIN, "reservations", ["state"], { + uniqueIndexes: ["idempotency_key"], + }); + // Raised hook timeout: the shared test-PG DB-create/migrate provisioning + // reproducibly times out at the default 10s ~1-in-3 under back-to-back + // runs (provisioning contention, NOT the feature — the batch tx completes + // in ~1.4s and leaks no connections). + }, 30000); + + afterEach(async () => { + await teardownForDialect(ctx); + }); + + it( + dialect === "sqlite" + ? "exactly M of N coupled reserve batches commit (better-sqlite3 serializes in-process → proves SQL correctness, NOT the race)" + : "exactly M of N coupled reserve batches commit under real concurrent connections (the true no-oversell race)", + async () => { + const inventory = new PluginStorageRepository(db, PLUGIN, "inventory", []); + const reservations = new PluginStorageRepository(db, PLUGIN, "reservations", [ + "idempotency_key", + "state", + ]); + + const M = 5; + const N = 20; + await inventory.insert("widget", { on_hand: M }); + + const results = await Promise.all( + Array.from({ length: N }, (_v, i) => { + const ops: BatchOp[] = [ + { + op: "insert", + collection: "reservations", + id: `res-${i}`, + data: { state: "held", sku: "widget", qty: 1, idempotency_key: `key-${i}` }, + }, + { + op: "updateIf", + collection: "inventory", + id: "widget", + where: { on_hand: { gte: 1 } }, + delta: { on_hand: { dec: 1 } }, + }, + ]; + return applyPluginStorageBatch(db, PLUGIN, ops); + }), + ); + + const applied = results.filter((r) => r.applied).length; + const failed = results.filter((r) => !r.applied); + + expect(applied).toBe(M); + expect(failed).toHaveLength(N - M); + // Every failure is a guard_failed decrement (the claim rolled back too). + for (const f of failed) { + if (f.applied) continue; + expect(f.reason).toBe("guard_failed"); + } + + // Stock fully drained, never negative. + expect((await inventory.get("widget"))?.on_hand).toBe(0); + + // Exactly M reservations exist and are held (the invariant: no held + // reservation without its decrement — a rolled-back batch leaves no row). + const held = await reservations.count({ state: "held" } as never); + expect(held).toBe(M); + const total = await reservations.count(); + expect(total).toBe(M); + }, + ); +}); diff --git a/packages/core/tests/integration/plugins/storage-batch.test.ts b/packages/core/tests/integration/plugins/storage-batch.test.ts new file mode 100644 index 0000000000..d6a046ddc8 --- /dev/null +++ b/packages/core/tests/integration/plugins/storage-batch.test.ts @@ -0,0 +1,526 @@ +/** + * Atomic multi-document batch primitive — `applyPluginStorageBatch` + * (the executor behind `ctx.storage.batch`). + * + * Runs on SQLite (always) and Postgres (when EMDASH_TEST_PG is set). The batch + * couples N conditional writes (`insert` / `updateIf`) across multiple documents + * and collections all-or-nothing: it commits iff EVERY op's guard passes, and + * any guard failure rolls back the WHOLE batch and reports which op failed. + * Guard/uniqueness OUTCOMES are reported (never thrown); malformed ops THROW. + */ + +import type { Kysely } from "kysely"; +import { it, expect, beforeEach, afterEach, describe } from "vitest"; + +import { + PluginStorageRepository, + applyPluginStorageBatch, +} from "../../../src/database/repositories/plugin-storage.js"; +import type { Database } from "../../../src/database/types.js"; +import { createStorageIndexes } from "../../../src/plugins/storage-indexes.js"; +import type { BatchOp } from "../../../src/plugins/types.js"; +import { + describeEachDialect, + setupForDialect, + teardownForDialect, + type DialectTestContext, +} from "../../utils/test-db.js"; + +interface Inventory { + on_hand: number; +} +interface Reservation { + state: string; + sku: string; + qty: number; + idempotency_key?: string; +} + +const PLUGIN = "shop"; + +describeEachDialect("Plugin storage atomic batch", (dialect) => { + let ctx: DialectTestContext; + let db: Kysely; + + beforeEach(async () => { + ctx = await setupForDialect(dialect); + db = ctx.db; + // Declared unique index on reservations.idempotency_key (the DB-enforced + // idempotency claim, mirroring the SQL adapter's ON CONFLICT). + await createStorageIndexes(db, PLUGIN, "reservations", [], { + uniqueIndexes: ["idempotency_key"], + }); + // Raised hook timeout: the shared test-PG DB-create/migrate provisioning + // can contend under back-to-back runs and blow the default 10s. The batch + // tx itself completes in ~1.4s — this only absorbs setup contention. + }, 30000); + + afterEach(async () => { + await teardownForDialect(ctx); + }); + + const inventory = () => new PluginStorageRepository(db, PLUGIN, "inventory", []); + const reservations = () => + new PluginStorageRepository(db, PLUGIN, "reservations", ["idempotency_key"]); + const batch = (ops: BatchOp[]) => applyPluginStorageBatch(db, PLUGIN, ops); + + // 1 — both-commit atomicity proof + it("commits a coupled decrement ∧ flip when both guards pass", async () => { + await inventory().insert("widget", { on_hand: 5 }); + await reservations().insert("r1", { state: "pending", sku: "widget", qty: 2 }); + + const result = await batch([ + { + op: "updateIf", + collection: "inventory", + id: "widget", + where: { on_hand: { gte: 2 } }, + delta: { on_hand: { dec: 2 } }, + }, + { + op: "updateIf", + collection: "reservations", + id: "r1", + where: { state: "pending" }, + set: { state: "held" }, + }, + ]); + + expect(result).toEqual({ + applied: true, + results: [ + { op: "updateIf", applied: true, data: { on_hand: 3 } }, + { op: "updateIf", applied: true, data: { state: "held", sku: "widget", qty: 2 } }, + ], + }); + expect((await inventory().get("widget"))?.on_hand).toBe(3); + expect((await reservations().get("r1"))?.state).toBe("held"); + }); + + // 2 — THE atomicity proof: guard-fail rolls BOTH back + it("rolls BOTH ops back when the decrement guard fails — neither applies", async () => { + await inventory().insert("widget", { on_hand: 1 }); + await reservations().insert("r1", { state: "pending", sku: "widget", qty: 2 }); + + const result = await batch([ + { + op: "updateIf", + collection: "inventory", + id: "widget", + where: { on_hand: { gte: 2 } }, + delta: { on_hand: { dec: 2 } }, + }, + { + op: "updateIf", + collection: "reservations", + id: "r1", + where: { state: "pending" }, + set: { state: "held" }, + }, + ]); + + expect(result).toEqual({ applied: false, failedIndex: 0, reason: "guard_failed" }); + expect((await inventory().get("widget"))?.on_hand).toBe(1); + expect((await reservations().get("r1"))?.state).toBe("pending"); + }); + + // 3 — second-op guard-fail rolls back the first + it("rolls back op 0 when the SECOND op's guard fails (failedIndex 1)", async () => { + await inventory().insert("widget", { on_hand: 5 }); + await reservations().insert("r1", { state: "held", sku: "widget", qty: 2 }); + + const result = await batch([ + { + op: "updateIf", + collection: "inventory", + id: "widget", + where: { on_hand: { gte: 2 } }, + delta: { on_hand: { dec: 2 } }, + }, + { + op: "updateIf", + collection: "reservations", + id: "r1", + where: { state: "pending" }, // already held → guard fails + set: { state: "held" }, + }, + ]); + + expect(result).toEqual({ applied: false, failedIndex: 1, reason: "guard_failed" }); + expect((await inventory().get("widget"))?.on_hand).toBe(5); + }); + + // 4 — full reserve: claim ∧ decrement ∧ flip + it("commits a claim insert ∧ decrement ∧ flip together (the reserve happy path)", async () => { + await inventory().insert("widget", { on_hand: 5 }); + + const result = await batch([ + { + op: "insert", + collection: "reservations", + id: "res-1", + data: { state: "held", sku: "widget", qty: 2, idempotency_key: "key-1" }, + }, + { + op: "updateIf", + collection: "inventory", + id: "widget", + where: { on_hand: { gte: 2 } }, + delta: { on_hand: { dec: 2 } }, + }, + ]); + + expect(result.applied).toBe(true); + if (!result.applied) throw new Error("expected applied"); + expect(result.results[0]).toEqual({ op: "insert", inserted: true }); + expect((await inventory().get("widget"))?.on_hand).toBe(3); + expect((await reservations().get("res-1"))?.state).toBe("held"); + }); + + // 5 — idempotent replay: duplicate claim (same id) → failedIndex 0 exists, no double-decrement + it("fails a duplicate claim insert at failedIndex 0 reason exists — no double-decrement", async () => { + await inventory().insert("widget", { on_hand: 5 }); + const reserve: BatchOp[] = [ + { + op: "insert", + collection: "reservations", + id: "res-1", + data: { state: "held", sku: "widget", qty: 2, idempotency_key: "key-1" }, + }, + { + op: "updateIf", + collection: "inventory", + id: "widget", + where: { on_hand: { gte: 2 } }, + delta: { on_hand: { dec: 2 } }, + }, + ]; + + expect((await batch(reserve)).applied).toBe(true); + expect((await inventory().get("widget"))?.on_hand).toBe(3); + + // Replay with the SAME reservation id → exists, decrement rolled back. + const replay = await batch(reserve); + expect(replay).toEqual({ applied: false, failedIndex: 0, reason: "exists" }); + expect((await inventory().get("widget"))?.on_hand).toBe(3); // NOT 1 + }); + + // 6 — unique_violation on a non-id unique field → failedIndex + conflictField + it("fails a claim with a duplicate idempotency_key (different id) → unique_violation + conflictField", async () => { + await inventory().insert("widget", { on_hand: 5 }); + await reservations().insert("res-1", { + state: "held", + sku: "widget", + qty: 2, + idempotency_key: "dup", + }); + + const result = await batch([ + { + op: "insert", + collection: "reservations", + id: "res-2", + data: { state: "held", sku: "widget", qty: 2, idempotency_key: "dup" }, + }, + { + op: "updateIf", + collection: "inventory", + id: "widget", + where: { on_hand: { gte: 2 } }, + delta: { on_hand: { dec: 2 } }, + }, + ]); + + expect(result).toEqual({ + applied: false, + failedIndex: 0, + reason: "unique_violation", + conflictField: "idempotency_key", + }); + expect((await inventory().get("widget"))?.on_hand).toBe(5); + expect(await reservations().get("res-2")).toBeNull(); + }); + + // 7 — ifNotExists insert treats an existing row as a satisfied no-op + it("ifNotExists insert on an existing row is a satisfied no-op — the batch still commits siblings", async () => { + await inventory().insert("widget", { on_hand: 5 }); + await reservations().insert("res-1", { + state: "held", + sku: "widget", + qty: 2, + idempotency_key: "k", + }); + + const result = await batch([ + { + op: "insert", + collection: "reservations", + id: "res-1", + ifNotExists: true, + data: { state: "held", sku: "widget", qty: 99, idempotency_key: "k" }, + }, + { + op: "updateIf", + collection: "inventory", + id: "widget", + where: { on_hand: { gte: 2 } }, + delta: { on_hand: { dec: 2 } }, + }, + ]); + + expect(result.applied).toBe(true); + if (!result.applied) throw new Error("expected applied"); + expect(result.results[0]).toEqual({ op: "insert", inserted: false, reason: "exists" }); + expect(result.results[1]).toEqual({ op: "updateIf", applied: true, data: { on_hand: 3 } }); + // The pre-existing row was NOT overwritten (qty stays 2, not 99). + expect((await reservations().get("res-1"))?.qty).toBe(2); + expect((await inventory().get("widget"))?.on_hand).toBe(3); + }); + + // 8 — cross-collection + cross-op-type + it("applies a cross-collection, cross-op-type batch (insert + updateIf), order-preserved", async () => { + await inventory().insert("widget", { on_hand: 4 }); + + const result = await batch([ + { + op: "insert", + collection: "reservations", + id: "res-1", + data: { state: "held", sku: "widget", qty: 1, idempotency_key: "a" }, + }, + { + op: "updateIf", + collection: "inventory", + id: "widget", + where: { on_hand: { gte: 1 } }, + delta: { on_hand: { dec: 1 } }, + }, + ]); + + expect(result.applied).toBe(true); + if (!result.applied) throw new Error("expected applied"); + expect(result.results[0]).toEqual({ op: "insert", inserted: true }); + expect(result.results[1]).toEqual({ op: "updateIf", applied: true, data: { on_hand: 3 } }); + }); + + // 9 — empty in:[] guard → guard_failed rolls back siblings + it("fails the batch when an updateIf op has an empty in:[] guard (rolls back siblings)", async () => { + await inventory().insert("widget", { on_hand: 5 }); + await reservations().insert("r1", { state: "pending", sku: "widget", qty: 2 }); + + const result = await batch([ + { + op: "updateIf", + collection: "inventory", + id: "widget", + where: { on_hand: { gte: 1 } }, + delta: { on_hand: { dec: 1 } }, + }, + { + op: "updateIf", + collection: "reservations", + id: "r1", + where: { state: { in: [] } }, // matches nothing + set: { state: "held" }, + }, + ]); + + expect(result).toEqual({ applied: false, failedIndex: 1, reason: "guard_failed" }); + expect((await inventory().get("widget"))?.on_hand).toBe(5); + }); + + // 10 — release: flip ∧ restore commit together / non-held rolls back the restore + it("release commits flip held→released ∧ stock restore together", async () => { + await inventory().insert("widget", { on_hand: 3 }); + await reservations().insert("r1", { state: "held", sku: "widget", qty: 2 }); + + const result = await batch([ + { + op: "updateIf", + collection: "reservations", + id: "r1", + where: { state: { in: ["held", "adopted"] } }, + set: { state: "released" }, + }, + { + op: "updateIf", + collection: "inventory", + id: "widget", + where: {}, + delta: { on_hand: { inc: 2 } }, + }, + ]); + + expect(result.applied).toBe(true); + expect((await reservations().get("r1"))?.state).toBe("released"); + expect((await inventory().get("widget"))?.on_hand).toBe(5); + }); + + it("release on a non-held reservation rolls back the stock restore (failedIndex 0)", async () => { + await inventory().insert("widget", { on_hand: 3 }); + await reservations().insert("r1", { state: "released", sku: "widget", qty: 2 }); + + const result = await batch([ + { + op: "updateIf", + collection: "reservations", + id: "r1", + where: { state: { in: ["held", "adopted"] } }, + set: { state: "released" }, + }, + { + op: "updateIf", + collection: "inventory", + id: "widget", + where: {}, + delta: { on_hand: { inc: 2 } }, + }, + ]); + + expect(result).toEqual({ applied: false, failedIndex: 0, reason: "guard_failed" }); + expect((await inventory().get("widget"))?.on_hand).toBe(3); // no double-restore + }); + + // 11 — malformed ops THROW (and land no partial write) + describe("malformed ops throw (programmer error), not reported failures", () => { + it("empty ops array throws", async () => { + await expect(batch([])).rejects.toThrow(/non-empty/i); + }); + + it("unknown op throws", async () => { + await expect( + batch([{ op: "frobnicate", collection: "inventory", id: "widget" } as unknown as BatchOp]), + ).rejects.toThrow(/unknown op/i); + }); + + it("float delta throws and lands NO partial write", async () => { + await inventory().insert("widget", { on_hand: 5 }); + await expect( + batch([ + { + op: "updateIf", + collection: "inventory", + id: "widget", + where: { on_hand: { gte: 1 } }, + delta: { on_hand: { dec: 1 } }, + }, + { + op: "updateIf", + collection: "inventory", + id: "widget", + where: {}, + delta: { on_hand: { dec: 1.5 } }, + }, + ]), + ).rejects.toThrow(TypeError); + // The valid op 0 never executed (validation precedes the transaction). + expect((await inventory().get("widget"))?.on_hand).toBe(5); + }); + + it("field in both set and delta throws", async () => { + await expect( + batch([ + { + op: "updateIf", + collection: "inventory", + id: "widget", + where: {}, + set: { on_hand: 1 }, + delta: { on_hand: { dec: 1 } }, + }, + ]), + ).rejects.toThrow(/both/i); + }); + + it("updateIf with neither set nor delta throws", async () => { + await expect( + batch([{ op: "updateIf", collection: "inventory", id: "widget", where: {} }]), + ).rejects.toThrow(/set.*delta/i); + }); + + it("insert without `data` throws up front and lands NO partial write", async () => { + await inventory().insert("widget", { on_hand: 5 }); + await expect( + batch([ + { + op: "updateIf", + collection: "inventory", + id: "widget", + where: { on_hand: { gte: 1 } }, + delta: { on_hand: { dec: 1 } }, + }, + { op: "insert", collection: "reservations", id: "x", data: undefined }, + ]), + ).rejects.toThrow(/insert.*requires.*data/i); + expect((await inventory().get("widget"))?.on_hand).toBe(5); + }); + + it("a batch exceeding the max op count throws", async () => { + const ops: BatchOp[] = Array.from({ length: 51 }, (_v, i) => ({ + op: "insert" as const, + collection: "reservations", + id: `r-${i}`, + data: { state: "held", sku: "widget", qty: 1, idempotency_key: `k-${i}` }, + })); + await expect(batch(ops)).rejects.toThrow(/maximum of 50 ops/i); + }); + }); + + // 12 — a non-abort executor error is re-thrown (only BatchAbort → {applied:false}) + it("re-throws a raw (non-guard, non-unique) executor error instead of swallowing it", async () => { + await inventory().insert("widget", { on_hand: 5 }); + // A BigInt in `data` throws in JSON.stringify inside the batch executor — + // not a BatchAbort and not a unique violation, so it must propagate (never + // become {applied:false}). The valid op 0 rolls back with it. + await expect( + batch([ + { + op: "updateIf", + collection: "inventory", + id: "widget", + where: { on_hand: { gte: 1 } }, + delta: { on_hand: { dec: 1 } }, + }, + { + op: "insert", + collection: "reservations", + id: "bad", + data: { idempotency_key: "x", nope: 10n as unknown as number }, + }, + ]), + ).rejects.toThrow(); + expect((await inventory().get("widget"))?.on_hand).toBe(5); + }); + + // 13 — results[] order matches ops[] order + it("results[] order matches ops[] order for a 3-op batch", async () => { + await inventory().insert("widget", { on_hand: 5 }); + + const result = await batch([ + { + op: "insert", + collection: "reservations", + id: "res-1", + data: { state: "held", sku: "widget", qty: 1, idempotency_key: "a" }, + }, + { + op: "updateIf", + collection: "inventory", + id: "widget", + where: { on_hand: { gte: 1 } }, + delta: { on_hand: { dec: 1 } }, + }, + { + op: "insert", + collection: "reservations", + id: "res-2", + ifNotExists: true, + data: { state: "held", sku: "widget", qty: 1, idempotency_key: "b" }, + }, + ]); + + expect(result.applied).toBe(true); + if (!result.applied) throw new Error("expected applied"); + expect(result.results.map((r) => r.op)).toEqual(["insert", "updateIf", "insert"]); + }); +}); diff --git a/packages/core/tests/unit/plugins/storage-batch-access.test.ts b/packages/core/tests/unit/plugins/storage-batch-access.test.ts new file mode 100644 index 0000000000..2519c351e0 --- /dev/null +++ b/packages/core/tests/unit/plugins/storage-batch-access.test.ts @@ -0,0 +1,79 @@ +/** + * Unit coverage for the storage ACCESS surface added by the batch primitive: + * - `ctx.storage.batch` is present AND every declared collection accessor still + * works (backwards compatibility of the `StorageAccess` intersection type). + * - a collection literally named `batch` is rejected at declaration time + * (it would be shadowed by the `ctx.storage.batch()` method). + * - a compile-time call-site typecheck of the intersection type (no `as`). + */ + +import Database from "better-sqlite3"; +import { Kysely, SqliteDialect } from "kysely"; +import { it, expect, describe } from "vitest"; + +import type { Database as DB } from "../../../src/database/types.js"; +import { createStorageAccess } from "../../../src/plugins/context.js"; +import { + assertCollectionNameAllowed, + createStorageIndexes, +} from "../../../src/plugins/storage-indexes.js"; +import type { BatchResult, StorageAccess, StorageCollection } from "../../../src/plugins/types.js"; + +function makeDb(): Kysely { + const sqlite = new Database(":memory:"); + // eslint-disable-next-line typescript-eslint/no-unsafe-type-assertion -- test-only in-memory db typed as the app schema + return new Kysely({ dialect: new SqliteDialect({ database: sqlite }) }); +} + +describe("ctx.storage access object (batch + per-collection)", () => { + it("exposes batch AND every declared collection accessor (backwards compat)", () => { + const db = makeDb(); + const storage = createStorageAccess(db, "shop", { + inventory: { indexes: ["sku"] }, + reservations: { indexes: [], uniqueIndexes: ["idempotency_key"] }, + }); + + // batch present and callable. + expect(typeof storage.batch).toBe("function"); + // per-collection accessors still present with the full StorageCollection API. + for (const name of ["inventory", "reservations"]) { + const coll = storage[name]; + expect(coll).toBeDefined(); + expect(typeof coll?.get).toBe("function"); + expect(typeof coll?.updateIf).toBe("function"); + expect(typeof coll?.insert).toBe("function"); + } + }); + + it("call-site typecheck: batch() returns Promise, collections stay StorageCollection (no `as`)", () => { + const db = makeDb(); + const storage: StorageAccess = createStorageAccess(db, "shop", { + inventory: { indexes: ["sku"] }, + }); + // These are compile-time assertions (the values are never awaited here). + const batchCall: Promise = storage.batch([]).catch( + (): BatchResult => ({ + applied: false, + failedIndex: 0, + reason: "guard_failed", + }), + ); + const coll: StorageCollection = storage.inventory!; + expect(batchCall).toBeInstanceOf(Promise); + expect(coll).toBeDefined(); + }); +}); + +describe("reserved collection name", () => { + it("assertCollectionNameAllowed rejects the reserved name `batch`", () => { + expect(() => assertCollectionNameAllowed("batch")).toThrow(/reserved/i); + // A normal name is fine. + expect(() => assertCollectionNameAllowed("inventory")).not.toThrow(); + }); + + it("createStorageIndexes rejects a collection named `batch` at declaration time", async () => { + const db = makeDb(); + // _plugin_storage / _plugin_indexes need not exist — validation happens first. + await expect(createStorageIndexes(db, "shop", "batch", ["x"])).rejects.toThrow(/reserved/i); + }); +}); diff --git a/packages/workerd/src/sandbox/bridge-handler.ts b/packages/workerd/src/sandbox/bridge-handler.ts index 86ba32631e..4350fb110b 100644 --- a/packages/workerd/src/sandbox/bridge-handler.ts +++ b/packages/workerd/src/sandbox/bridge-handler.ts @@ -14,8 +14,13 @@ * must produce same outputs, same return shapes, same error messages. */ -import { createHttpAccess, createUnrestrictedHttpAccess, PluginStorageRepository } from "emdash"; -import type { Database, SandboxEmailSendCallback } from "emdash"; +import { + createHttpAccess, + createUnrestrictedHttpAccess, + PluginStorageRepository, + applyPluginStorageBatch, +} from "emdash"; +import type { Database, SandboxEmailSendCallback, BatchOp } from "emdash"; import { sql, type Kysely, type RawBuilder } from "kysely"; /** @@ -324,6 +329,11 @@ async function dispatch( set: optionalRecord(body, "set"), delta: optionalRecord(body, "delta"), }); + case "storage/batch": + // requireBatchOps validates every op's declared collection BEFORE any + // write (the anti-smuggling defense), so an undeclared collection in + // ANY op rejects the whole batch without executing op 0. + return storageBatch(opts, requireBatchOps(opts, body, "ops")); // ── Logging ───────────────────────────────────────────────────── case "log": { @@ -535,6 +545,50 @@ function requireCapability(opts: BridgeHandlerOptions, capability: string): void } } +/** + * Validate the `storage/batch` ops array. Asserts an array of well-formed ops + * AND calls `validateStorageCollection` for EVERY op's collection so a batch + * cannot smuggle a write to an undeclared collection. This runs BEFORE any + * execution, so a rejected op never lets earlier ops commit. Guard/uniqueness + * outcomes are NOT validated here — those are reported by `applyPluginStorageBatch`. + */ +function requireBatchOps( + opts: BridgeHandlerOptions, + body: Record, + key: string, +): BatchOp[] { + const value = body[key]; + if (!Array.isArray(value)) { + throw new Error(`Parameter ${key} must be an array of batch ops`); + } + for (const op of value) { + if (!isRecord(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 (op.op === "updateIf") { + // Symmetric with the Cloudflare PluginBridge (same early errors) so both + // bridges reject malformed ops identically. + if (!isRecord(op.where)) { + throw new Error("storage/updateIf requires an object `where`"); + } + if (op.set !== undefined && !isRecord(op.set)) { + throw new Error("storage/updateIf `set` must be an object when provided"); + } + if (op.delta !== undefined && !isRecord(op.delta)) { + throw new Error("storage/updateIf `delta` must be an object when provided"); + } + } + validateStorageCollection(opts, op.collection); + } + // eslint-disable-next-line typescript-eslint/no-unsafe-type-assertion -- each entry validated above (op ∈ {insert,updateIf}, string collection/id); applyPluginStorageBatch re-validates set/delta shapes and reports guard outcomes. + return value as BatchOp[]; +} + function validateStorageCollection(opts: BridgeHandlerOptions, collection: string): void { if (!opts.storageCollections.includes(collection)) { // Error message matches Cloudflare PluginBridge format @@ -1660,3 +1714,13 @@ async function storageUpdateIf( delta: args.delta, }); } + +/** + * Apply an atomic multi-document batch. This workerd path is NEVER a D1 Kysely + * (`opts.db` is always the host Postgres / better-sqlite3 connection — verified), + * so it goes through the real-transaction `applyPluginStorageBatch`, producing + * the same `BatchResult` shape as the Cloudflare D1 bridge. + */ +async function storageBatch(opts: BridgeHandlerOptions, ops: BatchOp[]): Promise { + return applyPluginStorageBatch(opts.db, opts.pluginId, ops); +} diff --git a/packages/workerd/src/sandbox/wrapper.ts b/packages/workerd/src/sandbox/wrapper.ts index df6c0aa3e7..f208efdfb3 100644 --- a/packages/workerd/src/sandbox/wrapper.ts +++ b/packages/workerd/src/sandbox/wrapper.ts @@ -110,6 +110,9 @@ function createContext() { 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) => bridgeCall("storage/batch", { ops }); return createStorageCollection(collectionName); } }); diff --git a/packages/workerd/test/bridge-handler.test.ts b/packages/workerd/test/bridge-handler.test.ts index 3b8acf960a..7a4206466c 100644 --- a/packages/workerd/test/bridge-handler.test.ts +++ b/packages/workerd/test/bridge-handler.test.ts @@ -531,6 +531,167 @@ describe("Bridge Handler Conformance", () => { }); }); + // ── storage/batch (atomic multi-document) ───────────────────────────── + describe("storage/batch", () => { + function batchHandler() { + return makeHandler({ storageCollections: ["inventory", "reservations"] }); + } + + it("round-trips the ops array and applies a coupled decrement ∧ flip", async () => { + const handler = batchHandler(); + await call(handler, "storage/put", { + collection: "inventory", + id: "widget", + data: { on_hand: 5 }, + }); + await call(handler, "storage/put", { + collection: "reservations", + id: "r1", + data: { state: "pending" }, + }); + + const result = await call(handler, "storage/batch", { + ops: [ + { + op: "updateIf", + collection: "inventory", + id: "widget", + where: { on_hand: { gte: 2 } }, + delta: { on_hand: { dec: 2 } }, + }, + { + op: "updateIf", + collection: "reservations", + id: "r1", + where: { state: "pending" }, + set: { state: "held" }, + }, + ], + }); + + expect(result.result).toEqual({ + applied: true, + results: [ + { op: "updateIf", applied: true, data: { on_hand: 3 } }, + { op: "updateIf", applied: true, data: { state: "held" } }, + ], + }); + const inv = await call(handler, "storage/get", { collection: "inventory", id: "widget" }); + expect(inv.result).toEqual({ on_hand: 3 }); + }); + + it("guard-fail returns {applied:false, failedIndex, reason} and rolls back both ops", async () => { + const handler = batchHandler(); + await call(handler, "storage/put", { + collection: "inventory", + id: "widget", + data: { on_hand: 1 }, + }); + await call(handler, "storage/put", { + collection: "reservations", + id: "r1", + data: { state: "pending" }, + }); + + const result = await call(handler, "storage/batch", { + ops: [ + { + op: "updateIf", + collection: "inventory", + id: "widget", + where: { on_hand: { gte: 2 } }, + delta: { on_hand: { dec: 2 } }, + }, + { + op: "updateIf", + collection: "reservations", + id: "r1", + where: { state: "pending" }, + set: { state: "held" }, + }, + ], + }); + + expect(result.result).toEqual({ applied: false, failedIndex: 0, reason: "guard_failed" }); + const inv = await call(handler, "storage/get", { collection: "inventory", id: "widget" }); + expect(inv.result).toEqual({ on_hand: 1 }); + const res = await call(handler, "storage/get", { collection: "reservations", id: "r1" }); + expect(res.result).toEqual({ state: "pending" }); + }); + + it("rejects an undeclared collection in ANY op — op 0 must NOT commit", async () => { + const handler = makeHandler({ storageCollections: ["inventory"] }); + const result = await call(handler, "storage/batch", { + ops: [ + { op: "insert", collection: "inventory", id: "w1", data: { on_hand: 1 } }, + { op: "updateIf", collection: "secrets", id: "s1", where: {}, set: { a: 1 } }, + ], + }); + expect(result.error).toContain("Storage collection not declared: secrets"); + // Validation precedes execution → op 0 never inserted. + const inv = await call(handler, "storage/get", { collection: "inventory", id: "w1" }); + expect(inv.result).toBeNull(); + }); + + it("is scoped per plugin (a batch cannot touch another plugin's rows)", async () => { + const handlerA = createBridgeHandler({ + pluginId: "plugin-a", + version: "1.0.0", + capabilities: [], + allowedHosts: [], + storageCollections: ["inventory"], + db, + emailSend: () => null, + }); + const handlerB = createBridgeHandler({ + pluginId: "plugin-b", + version: "1.0.0", + capabilities: [], + allowedHosts: [], + storageCollections: ["inventory"], + db, + emailSend: () => null, + }); + + await call(handlerA, "storage/put", { + collection: "inventory", + id: "shared-id", + data: { on_hand: 5 }, + }); + // Plugin B's batch inserts its OWN row at the same id; A's row is untouched. + const result = await call(handlerB, "storage/batch", { + ops: [{ op: "insert", collection: "inventory", id: "shared-id", data: { on_hand: 99 } }], + }); + expect((result.result as { applied: boolean }).applied).toBe(true); + const a = await call(handlerA, "storage/get", { collection: "inventory", id: "shared-id" }); + expect(a.result).toEqual({ on_hand: 5 }); + }); + + it("throws a clean bridge error for malformed ops bodies", async () => { + const handler = batchHandler(); + const notArray = await call(handler, "storage/batch", { ops: { nope: true } }); + expect(notArray.error).toContain("must be an array"); + + const missingOp = await call(handler, "storage/batch", { + ops: [{ collection: "inventory", id: "x" }], + }); + expect(missingOp.error).toContain("unknown op"); + }); + + it("rejects a non-object set/delta up front (symmetric with the Cloudflare bridge)", async () => { + const handler = batchHandler(); + const badSet = await call(handler, "storage/batch", { + ops: [{ op: "updateIf", collection: "inventory", id: "widget", where: {}, set: "nope" }], + }); + expect(badSet.error).toContain("`set` must be an object"); + + const badDelta = await call(handler, "storage/batch", { + ops: [{ op: "updateIf", collection: "inventory", id: "widget", where: {}, delta: 5 }], + }); + expect(badDelta.error).toContain("`delta` must be an object"); + }); + }); + // ── Error Handling ──────────────────────────────────────────────────── describe("error handling", () => { diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index a2cd112ebc..ee6a5c4763 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1526,6 +1526,9 @@ importers: '@astrojs/cloudflare': specifier: 'catalog:' version: 14.0.0(@types/node@25.9.1)(astro@7.0.0(@emnapi/core@1.10.0)(@emnapi/runtime@1.10.0)(@types/node@25.9.1)(jiti@2.7.0)(rollup@4.55.2)(yaml@2.9.0))(esbuild@0.28.1)(jiti@2.7.0)(workerd@1.20260611.1)(wrangler@4.100.0(@cloudflare/workers-types@4.20260305.1))(yaml@2.9.0) + '@cloudflare/vitest-pool-workers': + specifier: 'catalog:' + version: 0.16.3(@cloudflare/workers-types@4.20260305.1)(@vitest/runner@4.1.5)(@vitest/snapshot@4.1.5)(vitest@4.1.5(@opentelemetry/api@1.9.0)(@types/node@25.9.1)(jsdom@26.1.0)(vite@8.0.16(@types/node@25.9.1)(esbuild@0.28.1)(jiti@2.7.0)(yaml@2.9.0))) '@cloudflare/workers-types': specifier: 'catalog:' version: 4.20260305.1 @@ -1544,6 +1547,9 @@ importers: vitest: specifier: 'catalog:' version: 4.1.5(@opentelemetry/api@1.9.0)(@types/node@25.9.1)(jsdom@26.1.0)(vite@8.0.16(@types/node@25.9.1)(esbuild@0.28.1)(jiti@2.7.0)(yaml@2.9.0)) + wrangler: + specifier: 'catalog:' + version: 4.100.0(@cloudflare/workers-types@4.20260305.1) packages/contentful-to-portable-text: dependencies: @@ -13973,6 +13979,21 @@ snapshots: - utf-8-validate - workerd + '@cloudflare/vitest-pool-workers@0.16.3(@cloudflare/workers-types@4.20260305.1)(@vitest/runner@4.1.5)(@vitest/snapshot@4.1.5)(vitest@4.1.5(@opentelemetry/api@1.9.0)(@types/node@25.9.1)(jsdom@26.1.0)(vite@8.0.16(@types/node@25.9.1)(esbuild@0.28.1)(jiti@2.7.0)(yaml@2.9.0)))': + dependencies: + '@vitest/runner': 4.1.5 + '@vitest/snapshot': 4.1.5 + cjs-module-lexer: 1.4.0 + esbuild: 0.27.3 + miniflare: 4.20260507.1 + vitest: 4.1.5(@opentelemetry/api@1.9.0)(@types/node@25.9.1)(jsdom@26.1.0)(vite@8.0.16(@types/node@25.9.1)(esbuild@0.28.1)(jiti@2.7.0)(yaml@2.9.0)) + wrangler: 4.90.0(@cloudflare/workers-types@4.20260305.1) + zod: 3.25.76 + transitivePeerDependencies: + - '@cloudflare/workers-types' + - bufferutil + - utf-8-validate + '@cloudflare/vitest-pool-workers@0.16.3(@vitest/runner@4.1.5)(@vitest/snapshot@4.1.5)(vitest@4.1.5(@opentelemetry/api@1.9.0)(@types/node@24.10.13)(jsdom@26.1.0)(vite@8.0.11(@types/node@24.10.13)(esbuild@0.28.1)(jiti@2.7.0)(yaml@2.9.0)))': dependencies: '@vitest/runner': 4.1.5 @@ -13981,7 +14002,7 @@ snapshots: esbuild: 0.27.3 miniflare: 4.20260507.1 vitest: 4.1.5(@opentelemetry/api@1.9.0)(@types/node@24.10.13)(jsdom@26.1.0)(vite@8.0.11(@types/node@24.10.13)(esbuild@0.28.1)(jiti@2.7.0)(yaml@2.9.0)) - wrangler: 4.90.0 + wrangler: 4.90.0(@cloudflare/workers-types@4.20260305.1) zod: 3.25.76 transitivePeerDependencies: - '@cloudflare/workers-types' @@ -23055,7 +23076,7 @@ snapshots: - bufferutil - utf-8-validate - wrangler@4.90.0: + wrangler@4.90.0(@cloudflare/workers-types@4.20260305.1): dependencies: '@cloudflare/kv-asset-handler': 0.5.0 '@cloudflare/unenv-preset': 2.16.1(unenv@2.0.0-rc.24)(workerd@1.20260507.1) @@ -23066,6 +23087,7 @@ snapshots: unenv: 2.0.0-rc.24 workerd: 1.20260507.1 optionalDependencies: + '@cloudflare/workers-types': 4.20260305.1 fsevents: 2.3.3 transitivePeerDependencies: - bufferutil