feat(plugins): atomic conditional writes for plugin storage (insert / updateIf)

Add two additive, single-statement primitives to the plugin
StorageCollection<T> API:

- insert(id, data): insert-once via INSERT … ON CONFLICT (pk) DO NOTHING.
  A same-id collision returns { inserted:false, reason:"exists" }; a
  declared unique-index collision on a non-id field is classified as
  { inserted:false, reason:"unique_violation", conflictField? } and any
  other DB error is re-thrown.
- updateIf(id, { where, set?, delta? }): predicate-guarded atomic update
  via one UPDATE … SET json_set/jsonb_set(…) WHERE <pk> AND <guard>
  RETURNING data. The guard reuses the numeric-correct where translation
  from the prior fix, and integer deltas are applied in-SQL with
  COALESCE(base,0) ± n, so N concurrent guarded decrements cannot
  oversell. set and delta are separate args; floats are rejected at
  runtime; an empty in:[] guard short-circuits to applied:false.

Declared uniqueIndexes are now materialized as real unique indexes on
plugin install (fail-loud on error) and dropped on uninstall. The new
methods are delegated through the in-process context, the workerd bridge,
and the Cloudflare D1 bridge so all runners behave identically.

Proven no-oversell on real Postgres (and the SQLite/D1 dialect); both
dialects covered by the contract suite.

Author: Ubuntu

.changeset/plugin-storage-conditional-writes-workerd.md

@@ -0,0 +1,5 @@
+---
+"@emdash-cms/sandbox-workerd": minor
+---
+
+Sandboxed plugins can now call `insert` and `updateIf` on their storage collections through the workerd bridge, matching the in-process storage API.

.changeset/plugin-storage-conditional-writes.md

@@ -0,0 +1,5 @@
+---
+"emdash": minor
+---
+
+Adds `insert` (insert-once) and `updateIf` (predicate-guarded atomic update) to plugin storage collections. `insert` creates a document only if its id is free and no declared unique index is violated; `updateIf` applies a wholesale `set` and/or integer `delta` in a single guarded statement, so concurrent guarded decrements (e.g. inventory) can no longer oversell. A collection's declared `uniqueIndexes` are now materialized as real unique indexes when the plugin installs — installation fails loudly if a unique index cannot be created (for instance because existing data already violates it).

packages/cloudflare/src/sandbox/bridge.ts

@@ -438,6 +438,55 @@ export class PluginBridge extends WorkerEntrypoint<PluginBridgeEnv, PluginBridge
 		return deleted;
 	}
 
+	/**
+	 * Insert-once. Delegates to PluginStorageRepository (same single-statement
+	 * `INSERT … ON CONFLICT DO NOTHING` + unique-violation classification as the
+	 * in-process and workerd paths) so the D1 production sandbox behaves
+	 * identically.
+	 */
+	async storageInsert(collection: string, id: string, data: unknown): Promise<unknown> {
+		const { storageCollections } = this.ctx.props;
+		if (!storageCollections.includes(collection)) {
+			throw new Error(`Storage collection not declared: ${collection}`);
+		}
+		return this.getStorageRepo(collection).insert(id, data);
+	}
+
+	/**
+	 * Predicate-guarded atomic update. Delegates to PluginStorageRepository so
+	 * the guarded single-statement `UPDATE … RETURNING` (the no-oversell
+	 * primitive) runs identically on the D1 production path. D1 serializes
+	 * writes, so concurrent guarded decrements can never oversell.
+	 */
+	async storageUpdateIf(
+		collection: string,
+		id: string,
+		args: { where?: unknown; set?: unknown; delta?: unknown },
+	): Promise<unknown> {
+		const { storageCollections } = this.ctx.props;
+		if (!storageCollections.includes(collection)) {
+			throw new Error(`Storage collection not declared: ${collection}`);
+		}
+		// Validate the guard/patch shapes up front (symmetric with the workerd
+		// bridge's requireRecord/optionalRecord) so a malformed call fails with a
+		// clean error rather than an incidental repo throw.
+		if (!isJsonObject(args.where)) {
+			throw new Error("storage/updateIf requires an object `where`");
+		}
+		if (args.set !== undefined && !isJsonObject(args.set)) {
+			throw new Error("storage/updateIf `set` must be an object when provided");
+		}
+		if (args.delta !== undefined && !isJsonObject(args.delta)) {
+			throw new Error("storage/updateIf `delta` must be an object when provided");
+		}
+		return this.getStorageRepo(collection).updateIf(id, {
+			// eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion -- WhereClause is structurally Record<string, WhereValue>; validated as an object above and re-validated by the repo.
+			where: args.where as never,
+			set: args.set,
+			delta: args.delta,
+		});
+	}
+
 	// =========================================================================
 	// Content Operations - capability-gated
 	// =========================================================================

packages/cloudflare/src/sandbox/wrapper.ts

@@ -82,7 +82,9 @@ function createContext(env) {
 			count: (where) => bridge.storageCount(collectionName, where),
 			getMany: (ids) => bridge.storageGetMany(collectionName, ids),
 			putMany: (items) => bridge.storagePutMany(collectionName, items),
-			deleteMany: (ids) => bridge.storageDeleteMany(collectionName, ids)
+			deleteMany: (ids) => bridge.storageDeleteMany(collectionName, ids),
+			insert: (id, data) => bridge.storageInsert(collectionName, id, data),
+			updateIf: (id, args) => bridge.storageUpdateIf(collectionName, id, args)
 		};
 	}
 	

packages/core/src/database/dialect-helpers.ts

@@ -277,6 +277,71 @@ export function pluginDataExtractExpr(
 	return `CASE WHEN json_type(data, '$.${field}') IN ('integer', 'real') THEN ${extract} END`;
 }
 
+/**
+ * Build the new value of the `_plugin_storage.data` (text-JSON) column for a
+ * guarded `updateIf`, composing wholesale `set` fields and integer `delta`
+ * fields into a SINGLE dialect-correct expression.
+ *
+ * Both branches go through `json_set` / `jsonb_set` so the write never rewrites
+ * the whole column from JS (which would require a read-then-write and break the
+ * single-statement atomicity that makes no-oversell hold):
+ *
+ * - **set** field → the value is stored via `json(?)` (SQLite) / `?::jsonb`
+ *   (Postgres) with `JSON.stringify(value)`, uniformly handling scalars,
+ *   objects, arrays, and `null` (stored as JSON `null`, never SQL `NULL` — a
+ *   SQL `NULL` in `jsonb_set` would null the entire `data` column and hit the
+ *   `NOT NULL` constraint).
+ * - **delta** field → `COALESCE(<numeric extract>, 0) + n`, where the extract
+ *   is the type-guarded numeric form from {@link pluginDataExtractExpr} so a
+ *   missing / null / non-number stored value coalesces to `0` on BOTH dialects
+ *   instead of throwing (Postgres) or coercing oddly. Integer arithmetic stays
+ *   integer (Postgres `to_jsonb(numeric)` and SQLite integer `+` both round-trip
+ *   without a spurious `.0`).
+ *
+ * Field names are validated (`validateJsonFieldName` / `pluginDataExtractExpr`)
+ * before interpolation, so the JSON path is a safe identifier and values are
+ * bound parameters — no injection surface.
+ *
+ * SQLite:   json_set(json_set(data, '$.f1', json(?)), '$.f2', COALESCE(json_extract(...), 0) + ?)
+ * Postgres: (jsonb_set(jsonb_set(data::jsonb, '{f1}', ?::jsonb), '{f2}', to_jsonb(COALESCE(..., 0) + ?)))::text
+ */
+// eslint-disable-next-line @typescript-eslint/no-explicit-any -- accepts any Kysely instance
+export function pluginDataWriteExpr(
+	db: Kysely<any>,
+	setEntries: Array<[string, unknown]>,
+	deltaEntries: Array<[string, number]>,
+): RawBuilder<string> {
+	const pg = isPostgres(db);
+	let expr: RawBuilder<unknown> = pg ? sql`data::jsonb` : sql`data`;
+
+	for (const [field, value] of setEntries) {
+		validateJsonFieldName(field, "plugin storage set field name");
+		const json = JSON.stringify(value ?? null);
+		if (pg) {
+			expr = sql`jsonb_set(${expr}, ${sql.lit(`{${field}}`)}, ${json}::jsonb)`;
+		} else {
+			expr = sql`json_set(${expr}, ${sql.lit(`$.${field}`)}, json(${json}))`;
+		}
+	}
+
+	for (const [field, n] of deltaEntries) {
+		// Type-guarded numeric extract over the ORIGINAL `data` column so the
+		// arithmetic is total (non-number → NULL → COALESCE 0), matching PR A's
+		// numeric-correctness posture on both dialects.
+		const numericExtract = pluginDataExtractExpr(db, field, { numeric: true });
+		if (pg) {
+			expr = sql`jsonb_set(${expr}, ${sql.lit(`{${field}}`)}, to_jsonb(COALESCE(${sql.raw(numericExtract)}, 0) + ${n}))`;
+		} else {
+			expr = sql`json_set(${expr}, ${sql.lit(`$.${field}`)}, COALESCE(${sql.raw(numericExtract)}, 0) + ${n})`;
+		}
+	}
+
+	if (pg) {
+		return sql<string>`(${expr})::text`;
+	}
+	return sql<string>`${expr}`;
+}
+
 /**
  * SQL expression for ordering plugin-storage rows by a `data` field.
  *

packages/core/src/database/repositories/plugin-storage.ts

@@ -16,17 +16,58 @@ import {
 	validateOrderByClause,
 	getIndexedFields,
 	jsonOrderExtract,
+	isInFilter,
 } from "../../plugins/storage-query.js";
 import type {
 	StorageCollection,
 	QueryOptions,
 	PaginatedResult,
 	WhereClause,
+	InsertResult,
+	UpdateIfArgs,
+	UpdateIfResult,
 } from "../../plugins/types.js";
+import { pluginDataWriteExpr } from "../dialect-helpers.js";
 import { withTransaction } from "../transaction.js";
 import type { Database } from "../types.js";
 import { encodeCursor, decodeCursor } from "./types.js";
 
+/**
+ * Classify a thrown DB error as a UNIQUE-constraint violation, returning the
+ * offending index name when the driver exposes it.
+ *
+ * - **Postgres** (`pg`): SQLSTATE `23505`; `error.constraint` carries the index
+ *   name.
+ * - **SQLite / D1** (better-sqlite3): `error.code === "SQLITE_CONSTRAINT_UNIQUE"`
+ *   or a message `UNIQUE constraint failed: index 'uidx_…'`. Our unique indexes
+ *   are partial EXPRESSION indexes, so the message names the index, not a column.
+ *
+ * Returns `null` for anything that is not a unique violation (the caller
+ * re-throws those — raw DB errors are never swallowed).
+ */
+const UNIQUE_CONSTRAINT_MESSAGE_RE = /UNIQUE constraint failed/i;
+const SQLITE_INDEX_NAME_RE = /index ['"]([^'"]+)['"]/;
+const SAFE_FIELD_NAME_RE = /^[a-zA-Z][a-zA-Z0-9_]*$/;
+
+function classifyUniqueViolation(error: unknown): { indexName?: string } | null {
+	if (typeof error !== "object" || error === null) return null;
+	const e = error as { code?: unknown; constraint?: unknown; message?: unknown };
+	if (e.code === "23505") {
+		return { indexName: typeof e.constraint === "string" ? e.constraint : undefined };
+	}
+	const message = typeof e.message === "string" ? e.message : "";
+	if (e.code === "SQLITE_CONSTRAINT_UNIQUE" || UNIQUE_CONSTRAINT_MESSAGE_RE.test(message)) {
+		const match = SQLITE_INDEX_NAME_RE.exec(message);
+		return { indexName: match?.[1] };
+	}
+	return null;
+}
+
+/** True for any non-null object that may carry `inc`/`dec` delta keys. */
+function isDeltaLike(value: unknown): value is { inc?: unknown; dec?: unknown } {
+	return typeof value === "object" && value !== null;
+}
+
 /**
  * Turn a `buildWhereClause` result (`?`-placeholder SQL + ordered params) into a
  * single boolean expression suitable for Kysely's `.where()`.
@@ -319,6 +360,176 @@ export class PluginStorageRepository<T = unknown> implements StorageCollection<T
 		// so this always satisfies its Promise<number> contract on both dialects.
 		return Number(result?.count ?? 0);
 	}
+
+	/**
+	 * Insert-once (see {@link StorageCollection.insert}).
+	 *
+	 * Single `INSERT … ON CONFLICT (plugin_id, collection, id) DO NOTHING`. The
+	 * conflict target is the primary key, so a same-`id` collision is swallowed
+	 * (0 rows affected → `{ inserted: false, reason: "exists" }`). A collision on
+	 * a declared partial UNIQUE expression index is NOT the conflict target, so
+	 * the statement throws — we classify that as `unique_violation` and re-throw
+	 * anything else.
+	 */
+	async insert(id: string, data: T): Promise<InsertResult> {
+		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 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);
+			return conflictField
+				? { inserted: false, reason: "unique_violation", conflictField }
+				: { inserted: false, reason: "unique_violation" };
+		}
+	}
+
+	/**
+	 * Predicate-guarded atomic update (see {@link StorageCollection.updateIf}).
+	 *
+	 * One guarded `UPDATE _plugin_storage SET data = <json_set/jsonb_set expr>,
+	 * updated_at = ? WHERE <pk> AND <guard> RETURNING data`. The guard reuses
+	 * PR A's numeric-correct `buildWhereClause` translation verbatim, and the
+	 * `set`/`delta` arithmetic is computed in-SQL — no read-then-write — which is
+	 * what makes N concurrent guarded decrements correct (no oversell).
+	 *
+	 * `applied` is derived from whether a `RETURNING` row came back (equivalently
+	 * rows-affected > 0). A missing row and a failed guard both yield 0 rows →
+	 * `{ applied: false }`; the two are intentionally indistinguishable. Never
+	 * inserts.
+	 */
+	async updateIf(id: string, args: UpdateIfArgs<T>): Promise<UpdateIfResult<T>> {
+		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`.");
+		}
+
+		// 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;
+				} 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]);
+			}
+		}
+
+		// 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 };
+			}
+		}
+
+		const now = new Date().toISOString();
+		const dataExpr = pluginDataWriteExpr(this.db, setEntries, deltaEntries);
+
+		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);
+
+		const whereResult = buildWhereClause(this.db, where);
+		if (whereResult.sql) {
+			query = query.where(buildRawWhereExpression(whereResult));
+		}
+
+		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 };
+	}
+
+	/**
+	 * 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<string | undefined> {
+		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();
+
+		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_<pluginId>_<collection>_<field>
+		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;
+		}
+		return undefined;
+	}
 }
 
 /**

packages/core/src/plugins/context.ts

@@ -123,6 +123,8 @@ function createStorageCollection<T>(
 		putMany: (items) => repo.putMany(items),
 		deleteMany: (ids) => repo.deleteMany(ids),
 		count: (where) => repo.count(where),
+		insert: (id, data) => repo.insert(id, data),
+		updateIf: (id, updateArgs) => repo.updateIf(id, updateArgs),
 
 		// Query returns PaginatedResult instead of the old format
 		async query(options?: QueryOptions): Promise<PaginatedResult<{ id: string; data: T }>> {