feat(plugins): predicate-guarded atomic updateIf for plugin storage (no-oversell)

.changeset/plugin-storage-range-filter-guard.md

@@ -0,0 +1,12 @@
+---
+"emdash": patch
+---
+
+Fixes a plugin storage range filter whose every bound is `undefined` matching every row instead of failing. Building a bound from an optional value — `where: { stock: { gte: minStock } }` where `minStock` is `undefined` — type-checks, but contributed no SQL, so `query()` and `count()` returned the whole collection and `updateIf()` applied its write with no guard at all. A guarded decrement could then drive a counter past the bound the caller asked for.
+
+Such a filter now throws `StorageQueryError` naming the field. Pass a defined bound, or omit the field when you mean to match unconditionally:
+
+```typescript
+const where = minStock === undefined ? {} : { stock: { gte: minStock } };
+await ctx.storage.products.query({ where });
+```

.changeset/plugin-storage-updateif.md

@@ -0,0 +1,11 @@
+---
+"emdash": minor
+"@emdash-cms/cloudflare": patch
+"@emdash-cms/sandbox-workerd": patch
+---
+
+Adds `ctx.storage.<collection>.updateIf(id, { where, set?, delta? })` for atomic conditional updates to existing plugin documents. Use `where` to check stored fields, `set` to replace field values, and `delta` to increment or decrement integer counters. The method returns `{ applied: true, data }` with the updated document, or `{ applied: false }` when the document is absent or the condition fails. It never inserts a document.
+
+Malformed update arguments reject without writing. Deltas require safe integer operands and results; missing or `null` counters start at `0`. Invalid stored counters, overflow, and non-object documents return `{ applied: false }` without changing any fields.
+
+Available to native plugins and sandboxed plugins on Cloudflare and Workerd, with SQLite, D1, and PostgreSQL support. PostgreSQL serialization failures and deadlocks expose `code: "STORAGE_SERIALIZATION_FAILURE"` and `retryable: true`, including across sandbox transports. Retry standalone calls with bounded backoff, or restart the entire explicit transaction.

docs/src/content/docs/plugins/creating-plugins/storage.mdx

@@ -85,6 +85,7 @@ interface StorageCollection<T = unknown> {
 	// Basic CRUD
 	get(id: string): Promise<T | null>;
 	put(id: string, data: T): Promise<void>;
+	updateIf(id: string, args: UpdateIfArgs<T>): Promise<UpdateIfResult<T>>;
 	delete(id: string): Promise<boolean>;
 	exists(id: string): Promise<boolean>;
 
@@ -99,6 +100,52 @@ interface StorageCollection<T = unknown> {
 }
 ```
 
+## Conditional updates
+
+Use `updateIf()` to change an existing document only when its stored fields match a condition. The database checks the condition and applies the changes in one atomic operation. This method is available to native plugins and sandboxed plugins on Cloudflare and Workerd.
+
+Import its `NumericDelta`, `UpdateIfArgs`, and `UpdateIfResult` types with `import type` from `emdash` or `emdash/plugin`.
+
+The following call approves a pending submission and increments its review count in the same operation:
+
+```typescript
+const result = await ctx.storage.submissions.updateIf("sub_123", {
+	where: { status: "pending" },
+	set: { status: "approved" },
+	delta: { reviewCount: { inc: 1 } },
+});
+
+if (result.applied) {
+	ctx.log.info("Submission approved", { submission: result.data });
+}
+```
+
+A successful call returns `{ applied: true, data }` with the complete updated document. It returns `{ applied: false }` if the document is missing or the condition does not match. It never inserts a document.
+
+The arguments have the following behavior:
+
+- `where` is required and uses the same operators as [query filters](#where-clause-operators). An explicit `where: {}` adds no field conditions. Guard fields do not need declared query indexes because the update targets one document by ID.
+- A range filter needs at least one defined bound. Undefined bounds are ignored when another bound is defined. Numeric operands used by a guard must be finite.
+- `set` replaces each supplied top-level field value and leaves other fields unchanged. Values must be JSON-serializable.
+- `delta` applies exactly one `{ inc: number }` or `{ dec: number }` per field. Each operand must be a safe integer; negative operands are allowed.
+- A field cannot appear in both `set` and `delta`. Top-level `undefined` entries in either object are ignored. At least one defined field must remain.
+
+Malformed update arguments reject the promise without changing the document. The arguments object, `set`, `delta`, and each delta operation must be plain objects.
+
+### Integer counters
+
+A delta starts a missing or `null` counter at `0`. Existing counters and their results must be integers between `Number.MIN_SAFE_INTEGER` and `Number.MAX_SAFE_INTEGER`. A string, boolean, object, array, fractional number, unsafe integer, or out-of-range result causes the entire update to return `{ applied: false }`. A stored document that is not a JSON object also returns `{ applied: false }`. No fields change in either case.
+
+Deltas can produce negative values. To keep a counter nonnegative, pair a decrement of `n` with a `where` condition requiring that counter to be at least `n`.
+
+### Retry serialization failures
+
+PostgreSQL can reject concurrent writes with a serialization failure or deadlock. A deadlock can occur at any isolation level, including READ COMMITTED. In native plugins, these failures throw `StorageSerializationError` with `code: "STORAGE_SERIALIZATION_FAILURE"`, `retryable: true`, and an optional `sqlState` (`40001` or `40P01`). Import the error class from `emdash`.
+
+Use bounded retries with backoff for a standalone call. If the call is inside an explicit transaction, restart the entire transaction, including its reads; retrying the write inside the aborted transaction cannot succeed. Handle `{ applied: false }` as an unapplied update rather than a serialization error.
+
+Sandbox transports preserve the error name and retry metadata, but do not guarantee `instanceof StorageSerializationError`. Check `code` and `retryable` when handling errors across a sandbox boundary.
+
 ## Querying
 
 `query()` returns paginated results filtered by indexed fields:

packages/cloudflare/src/sandbox/bridge.ts

@@ -16,12 +16,14 @@ import {
 	getSandboxRouteErrorDetails,
 	ulid,
 	PluginStorageRepository,
+	StorageSerializationError,
 	resolveContentCreateLocale,
 } from "emdash";
 import { Kysely } from "kysely";
 import { D1Dialect } from "kysely-d1";
 
 import { sandboxHttpFetch } from "./bridge-http.js";
+import type { StorageUpdateIfResponse } from "./types.js";
 
 /** Regex to validate collection names (prevent SQL injection) */
 const COLLECTION_NAME_REGEX = /^[a-z][a-z0-9_]*$/;
@@ -349,6 +351,33 @@ export class PluginBridge extends WorkerEntrypoint<PluginBridgeEnv, PluginBridge
 			.run();
 	}
 
+	async storageUpdateIf(
+		collection: string,
+		id: string,
+		args: unknown,
+	): Promise<StorageUpdateIfResponse> {
+		if (!this.ctx.props.storageCollections.includes(collection)) {
+			throw new Error(`Storage collection not declared: ${collection}`);
+		}
+		try {
+			return await this.getStorageRepo(collection).updateIf(id, args);
+		} catch (error) {
+			if (!(error instanceof StorageSerializationError)) throw error;
+			return {
+				__emdashStorageError: {
+					name: "StorageSerializationError",
+					code: "STORAGE_SERIALIZATION_FAILURE",
+					retryable: true,
+					...(error.sqlState === "40001" || error.sqlState === "40P01"
+						? { sqlState: error.sqlState }
+						: {}),
+					message:
+						"Storage write must be retried. Restart the transaction before retrying when using an explicit transaction.",
+				},
+			};
+		}
+	}
+
 	async storageDelete(collection: string, id: string): Promise<boolean> {
 		const { pluginId, storageCollections } = this.ctx.props;
 		if (!storageCollections.includes(collection)) {

packages/cloudflare/src/sandbox/types.ts

@@ -3,7 +3,7 @@
  */
 
 import type { D1Database, R2Bucket } from "@cloudflare/workers-types";
-import type { ContentCreateOptions } from "emdash";
+import type { ContentCreateOptions, UpdateIfArgs, UpdateIfResult } from "emdash";
 
 /**
  * Environment bindings required for sandbox runner.
@@ -151,6 +151,18 @@ interface BridgeMediaItem {
 	createdAt: string;
 }
 
+export interface StorageSerializationFailureDetails {
+	name: "StorageSerializationError";
+	code: "STORAGE_SERIALIZATION_FAILURE";
+	retryable: true;
+	sqlState?: "40001" | "40P01";
+	message: string;
+}
+
+export type StorageUpdateIfResponse =
+	| UpdateIfResult<unknown>
+	| { __emdashStorageError: StorageSerializationFailureDetails };
+
 /**
  * Type for the PluginBridge binding passed to sandboxed workers.
  * This is the RPC interface exposed by PluginBridge WorkerEntrypoint.
@@ -164,6 +176,11 @@ export interface PluginBridgeBinding {
 	// Storage
 	storageGet(collection: string, id: string): Promise<unknown>;
 	storagePut(collection: string, id: string, data: unknown): Promise<void>;
+	storageUpdateIf(
+		collection: string,
+		id: string,
+		args: UpdateIfArgs<unknown>,
+	): Promise<StorageUpdateIfResponse>;
 	storageDelete(collection: string, id: string): Promise<boolean>;
 	storageQuery(
 		collection: string,

packages/cloudflare/src/sandbox/wrapper.ts

@@ -60,6 +60,19 @@ import pluginModule from "sandbox-plugin.js";
 const hooks = pluginModule?.hooks || pluginModule?.default?.hooks || {};
 const routes = pluginModule?.routes || pluginModule?.default?.routes || {};
 
+function storageSerializationErrorDetails(value) {
+	if (!value || typeof value !== "object" ||
+		value.code !== "STORAGE_SERIALIZATION_FAILURE" || value.retryable !== true ||
+		(value.sqlState !== undefined && value.sqlState !== "40001" && value.sqlState !== "40P01")) return null;
+	return {
+		name: "StorageSerializationError",
+		code: "STORAGE_SERIALIZATION_FAILURE",
+		retryable: true,
+		...(value.sqlState === undefined ? {} : { sqlState: value.sqlState }),
+		message: "Storage write must be retried. Restart the transaction before retrying when using an explicit transaction.",
+	};
+}
+
 function sandboxRouteErrorDetails(value) {
 	if (!value || typeof value !== "object") return null;
 	const code =
@@ -102,6 +115,15 @@ function createContext(env) {
 		return {
 			get: (id) => bridge.storageGet(collectionName, id),
 			put: (id, data) => bridge.storagePut(collectionName, id, data),
+			updateIf: async (id, args) => {
+				const result = await bridge.storageUpdateIf(collectionName, id, args);
+				if (result && typeof result === "object" && "__emdashStorageError" in result) {
+					const details = storageSerializationErrorDetails(result.__emdashStorageError);
+					if (!details) throw new Error("Invalid storage error response");
+					throw Object.assign(new Error(details.message), details);
+				}
+				return result;
+			},
 			delete: (id) => bridge.storageDelete(collectionName, id),
 			exists: async (id) => (await bridge.storageGet(collectionName, id)) !== null,
 			query: (opts) => bridge.storageQuery(collectionName, opts),

packages/cloudflare/tests/sandbox/wrapper-storage.test.ts

@@ -0,0 +1,133 @@
+import { describe, expect, it } from "vitest";
+
+import { generatePluginWrapper } from "../../src/sandbox/wrapper.js";
+
+type StorageUpdate = (collection: string, id: string, args: unknown) => Promise<unknown>;
+
+type TestEnv = { BRIDGE: { storageUpdateIf: StorageUpdate } };
+
+interface TestEntrypoint {
+	invokeHook(name: string, args: unknown): Promise<unknown>;
+}
+
+function createWrapper(storageUpdateIf: StorageUpdate): TestEntrypoint {
+	const source = generatePluginWrapper({
+		id: "storage-wrapper",
+		version: "1.0.0",
+		capabilities: [],
+		allowedHosts: [],
+		storage: { records: { indexes: ["state"] } },
+		hooks: ["content:beforeSave"],
+		routes: [],
+		admin: {},
+	})
+		.replace('import { WorkerEntrypoint } from "cloudflare:workers";', "")
+		.replace('import pluginModule from "sandbox-plugin.js";', "")
+		.replace("export default class PluginEntrypoint", "return class PluginEntrypoint");
+	class WorkerEntrypoint {
+		constructor(readonly env: TestEnv) {}
+	}
+	const pluginModule = {
+		hooks: {
+			"content:beforeSave": (
+				args: unknown,
+				ctx: {
+					storage: { records: { updateIf(id: string, input: unknown): Promise<unknown> } };
+				},
+			) => ctx.storage.records.updateIf("constructor", args),
+		},
+	};
+	// eslint-disable-next-line no-implied-eval -- The generated worker module is exercised in an isolated function scope.
+	const factory = new Function("WorkerEntrypoint", "pluginModule", source);
+	const Entrypoint = factory(WorkerEntrypoint, pluginModule) as new (
+		env: TestEnv,
+	) => TestEntrypoint;
+	return new Entrypoint({ BRIDGE: { storageUpdateIf } });
+}
+
+describe("Cloudflare generated storage wrapper", () => {
+	it("forwards literal IDs and guard arguments without changing the collection namespace", async () => {
+		const args = {
+			where: { state: "ready" },
+			set: { state: "running" },
+			collection: "other",
+			id: "other-id",
+		};
+		const calls: Array<{ collection: string; id: string; args: unknown }> = [];
+		const wrapper = createWrapper(async (collection, id, input) => {
+			calls.push({ collection, id, args: input });
+			return { applied: true, data: { state: "running" } };
+		});
+
+		expect(await wrapper.invokeHook("content:beforeSave", args)).toEqual({
+			applied: true,
+			data: { state: "running" },
+		});
+		expect(calls).toEqual([{ collection: "records", id: "constructor", args }]);
+	});
+
+	it("preserves malformed arguments for host validation instead of normalizing them", async () => {
+		const inputs: unknown[] = [];
+		const wrapper = createWrapper(async (_collection, _id, args) => {
+			inputs.push(args);
+			throw new TypeError("Invalid update arguments");
+		});
+		for (const args of [undefined, null, ["invalid"]]) {
+			await expect(wrapper.invokeHook("content:beforeSave", args)).rejects.toThrow(TypeError);
+		}
+		expect(inputs).toEqual([undefined, null, ["invalid"]]);
+	});
+
+	it.each(["40001", "40P01", undefined])(
+		"reconstructs a safe retry error with SQLSTATE %s",
+		async (sqlState) => {
+			const wrapper = createWrapper(async () => ({
+				__emdashStorageError: {
+					name: "PrivateDatabaseError",
+					code: "STORAGE_SERIALIZATION_FAILURE",
+					retryable: true,
+					...(sqlState === undefined ? {} : { sqlState }),
+					message: "private SQL and parameters",
+					cause: { query: "private SQL" },
+					query: "private SQL",
+				},
+			}));
+			const outcome = await wrapper
+				.invokeHook("content:beforeSave", { where: {}, set: { state: "ready" } })
+				.then(
+					(value) => ({ value }),
+					(error: unknown) => ({ error }),
+				);
+			expect(outcome).toHaveProperty("error");
+			if (!("error" in outcome)) throw new Error("Expected a rejected storage update");
+			expect(outcome.error).toBeInstanceOf(Error);
+			expect(outcome.error).toMatchObject({
+				name: "StorageSerializationError",
+				code: "STORAGE_SERIALIZATION_FAILURE",
+				retryable: true,
+				...(sqlState === undefined ? {} : { sqlState }),
+				message:
+					"Storage write must be retried. Restart the transaction before retrying when using an explicit transaction.",
+			});
+			expect(outcome.error).not.toHaveProperty("cause");
+			expect(outcome.error).not.toHaveProperty("query");
+			if (sqlState === undefined) expect(outcome.error).not.toHaveProperty("sqlState");
+			expect(String(outcome.error)).not.toContain("private SQL");
+		},
+	);
+
+	it("rejects retry envelopes carrying an unrecognized SQLSTATE", async () => {
+		const wrapper = createWrapper(async () => ({
+			__emdashStorageError: {
+				code: "STORAGE_SERIALIZATION_FAILURE",
+				retryable: true,
+				sqlState: "23505",
+				message: "private SQL",
+				cause: "private parameters",
+			},
+		}));
+		await expect(
+			wrapper.invokeHook("content:beforeSave", { where: {}, set: { state: "ready" } }),
+		).rejects.toThrow("Invalid storage error response");
+	});
+});