feat(plugins): add revision based conditional writes to storage and KV (#2980)

* feat(plugins): add revision-based conditional storage writes

* test(database): include all trailing migrations in recovery fixture

* Initialize plugin revisions without a backfill

* test(plugins): verify guarded updates invalidate revisions

Cover updateIf and revision based storage writes together on SQLite,
PostgreSQL and D1, including equal-value updates, failed guards and
existing records that start at revision 0. Update the D1 guarded-write
fixture for the revision column.

Co-authored-by: Vedanshu <7571618+vedanshujain@users.noreply.github.com>

---------

Co-authored-by: logelog <194732487+logelog@users.noreply.github.com>
Co-authored-by: Vedanshu <7571618+vedanshujain@users.noreply.github.com>

Author: logelog

.changeset/wild-schools-roll.md

@@ -0,0 +1,11 @@
+---
+"emdash": minor
+"@emdash-cms/cloudflare": minor
+"@emdash-cms/sandbox-workerd": minor
+---
+
+Adds `getVersioned`, `compareAndSet` and `compareAndDelete` to plugin storage collections and `ctx.kv`. Native and sandboxed plugins can create an absent key or condition a replacement or deletion on the revision they read, preventing concurrent requests from silently overwriting each other.
+
+Pass an explicit `null` revision to create only when absent. A successful replacement returns its new revision; a conflict returns `{ applied: false }`. Invalid input, permission failures and database failures reject the promise. Atomicity applies to one key, so changes spanning multiple records still require an application-level protocol.
+
+Update core and the sandbox adapter together and apply the host database migrations before using the methods. The migration initializes existing records without a backfill. Stored values are preserved, and existing unconditional writes continue to work while invalidating old revisions. Conditional keys are limited to 1,024 JavaScript string characters and values to 1 MiB of UTF-8 JSON.

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

@@ -14,6 +14,10 @@ Every hook and route receives this KV interface on `ctx`:
 ```typescript
 interface KVAccess {
 	get<T>(key: string): Promise<T | null>;
+	getVersioned<T>(key: string): Promise<{ value: T; revision: string } | null>;
+	compareAndSet(key: string, expectedRevision: string | null, value: unknown):
+		Promise<{ applied: true; revision: string } | { applied: false }>;
+	compareAndDelete(key: string, expectedRevision: string): Promise<{ applied: boolean }>;
 	set(key: string, value: unknown): Promise<void>;
 	delete(key: string): Promise<boolean>;
 	list(prefix?: string): Promise<Array<{ key: string; value: unknown }>>;
@@ -22,6 +26,8 @@ interface KVAccess {
 
 KV is namespaced by plugin. Two plugins can use the same key without reading or overwriting each other's values.
 
+When concurrent requests can change the same key, use [conditional writes](/plugins/creating-plugins/storage/#conditional-writes) to reject updates based on a stale revision. The same methods work in native plugins and sandboxed plugins.
+
 Use prefixes to separate user settings from internal state and cached values:
 
 | Prefix | Purpose | Example |

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

@@ -87,10 +87,16 @@ 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>;
 
+	// Conditional writes
+	getVersioned(id: string): Promise<{ value: T; revision: string } | null>;
+	compareAndSet(id: string, expectedRevision: string | null, data: T):
+		Promise<{ applied: true; revision: string } | { applied: false }>;
+	compareAndDelete(id: string, expectedRevision: string): Promise<{ applied: boolean }>;
+	updateIf(id: string, args: UpdateIfArgs<T>): Promise<UpdateIfResult<T>>;
+
 	// Batch operations
 	getMany(ids: string[]): Promise<Map<string, T>>;
 	putMany(items: Array<{ id: string; data: T }>): Promise<void>;
@@ -102,6 +108,48 @@ interface StorageCollection<T = unknown> {
 }
 ```
 
+## Conditional writes
+
+Use `getVersioned()` and `compareAndSet()` when concurrent requests can update the same record. These methods are available on declared `ctx.storage` collections and on `ctx.kv`, for native and sandboxed plugins. Each operation accesses one key in the calling plugin's namespace.
+
+The methods have the following behavior:
+
+| Method | Result |
+| --- | --- |
+| `getVersioned(key)` | The stored JSON value and an opaque revision, or `null` when the row is absent. A stored JSON `null` returns `{ value: null, revision }`. |
+| `compareAndSet(key, null, value)` | Creates the row only when it is absent. |
+| `compareAndSet(key, revision, value)` | Replaces the entire value only when the stored revision matches. |
+| `compareAndDelete(key, revision)` | Deletes the row only when the stored revision matches. |
+
+A successful `compareAndSet()` returns `{ applied: true, revision }`. A failed precondition returns `{ applied: false }`; invalid arguments, missing permissions and database failures reject the promise. `compareAndDelete()` returns `{ applied: boolean }`. An unrelated unique-index violation is an error, even when the requested key is absent.
+
+Pass revisions back unchanged and only for the key they came from. Every write changes the revision, including equal-value `set()`, `put()` and batch writes. Deleting and recreating a key invalidates its previous revision.
+
+The following helper adds a completed job to a plugin's counter, retrying up to three times when another request writes first.
+
+```typescript title="src/completed-jobs.ts"
+import type { PluginContext } from "emdash/plugin";
+
+export async function recordCompletedJob(ctx: PluginContext): Promise<number> {
+	const key = "state:completedJobs";
+	for (let attempt = 0; attempt < 3; attempt++) {
+		const current = await ctx.kv.getVersioned<number>(key);
+		const count = (current?.value ?? 0) + 1;
+		const result = await ctx.kv.compareAndSet(key, current?.revision ?? null, count);
+		if (result.applied) return count;
+	}
+	throw new Error("Job counter changed repeatedly; try again later");
+}
+```
+
+On a conflict, read the value again and recompute the proposed change. Keep retries bounded. A lost response can leave the outcome of a write unknown; these methods do not make external actions or retried job executions happen exactly once.
+
+Atomicity covers a single key. Reading a content item and writing a plugin record, or writing two plugin records, are separate operations. Put fields that must change together in one value. Enforce business rules such as job ownership or quantity limits when constructing that value.
+
+Conditional methods require a nonempty key of at most 1,024 JavaScript string characters and a JSON value of at most 1 MiB after UTF-8 encoding. A revision must be a nonempty string of at most 128 characters. Omitted revisions are invalid; only an explicit `null` requests creation. Existing unconditional methods retain their behavior.
+
+Deploy the matching core and sandbox adapter versions and apply the host database migrations before using these methods. The migration preserves stored values and makes writes from older host processes invalidate revisions during a rolling deployment.
+
 ## 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.

packages/cloudflare/src/sandbox/bridge.ts

@@ -9,7 +9,15 @@
 
 import type { D1Database } from "@cloudflare/workers-types";
 import { WorkerEntrypoint } from "cloudflare:workers";
-import type { ContentCreateOptions, Database, I18nConfig, SandboxEmailSendCallback } from "emdash";
+import type {
+	ConditionalDeleteResult,
+	ConditionalWriteResult,
+	ContentCreateOptions,
+	Database,
+	I18nConfig,
+	SandboxEmailSendCallback,
+	VersionedValue,
+} from "emdash";
 import {
 	ContentRepository,
 	createSandboxRouteError,
@@ -291,12 +299,31 @@ export class PluginBridge extends WorkerEntrypoint<PluginBridgeEnv, PluginBridge
 	async kvSet(key: string, value: unknown): Promise<void> {
 		const { pluginId } = this.ctx.props;
 		await this.env.DB.prepare(
-			"INSERT OR REPLACE INTO _plugin_storage (plugin_id, collection, id, data, updated_at) VALUES (?, '__kv', ?, ?, datetime('now'))",
+			"INSERT OR REPLACE INTO _plugin_storage (plugin_id, collection, id, data, revision, updated_at) VALUES (?, '__kv', ?, ?, ?, datetime('now'))",
 		)
-			.bind(pluginId, key, JSON.stringify(value))
+			.bind(pluginId, key, JSON.stringify(value), crypto.randomUUID())
 			.run();
 	}
 
+	async kvGetVersioned(key: string): Promise<VersionedValue | null> {
+		return this.getStorageRepo("__kv").getVersioned(key);
+	}
+
+	async kvCompareAndSet(
+		key: string,
+		expectedRevision: string | null,
+		value: unknown,
+	): Promise<ConditionalWriteResult> {
+		return this.getStorageRepo("__kv").compareAndSet(key, expectedRevision, value);
+	}
+
+	async kvCompareAndDelete(
+		key: string,
+		expectedRevision: string,
+	): Promise<ConditionalDeleteResult> {
+		return this.getStorageRepo("__kv").compareAndDelete(key, expectedRevision);
+	}
+
 	async kvDelete(key: string): Promise<boolean> {
 		const { pluginId } = this.ctx.props;
 		const result = await this.env.DB.prepare(
@@ -345,12 +372,42 @@ export class PluginBridge extends WorkerEntrypoint<PluginBridgeEnv, PluginBridge
 			throw new Error(`Storage collection not declared: ${collection}`);
 		}
 		await this.env.DB.prepare(
-			"INSERT OR REPLACE INTO _plugin_storage (plugin_id, collection, id, data, updated_at) VALUES (?, ?, ?, ?, datetime('now'))",
+			"INSERT OR REPLACE INTO _plugin_storage (plugin_id, collection, id, data, revision, updated_at) VALUES (?, ?, ?, ?, ?, datetime('now'))",
 		)
-			.bind(pluginId, collection, id, JSON.stringify(data))
+			.bind(pluginId, collection, id, JSON.stringify(data), crypto.randomUUID())
 			.run();
 	}
 
+	async storageGetVersioned(collection: string, id: string): Promise<VersionedValue | null> {
+		if (!this.ctx.props.storageCollections.includes(collection)) {
+			throw new Error(`Storage collection not declared: ${collection}`);
+		}
+		return this.getStorageRepo(collection).getVersioned(id);
+	}
+
+	async storageCompareAndSet(
+		collection: string,
+		id: string,
+		expectedRevision: string | null,
+		data: unknown,
+	): Promise<ConditionalWriteResult> {
+		if (!this.ctx.props.storageCollections.includes(collection)) {
+			throw new Error(`Storage collection not declared: ${collection}`);
+		}
+		return this.getStorageRepo(collection).compareAndSet(id, expectedRevision, data);
+	}
+
+	async storageCompareAndDelete(
+		collection: string,
+		id: string,
+		expectedRevision: string,
+	): Promise<ConditionalDeleteResult> {
+		if (!this.ctx.props.storageCollections.includes(collection)) {
+			throw new Error(`Storage collection not declared: ${collection}`);
+		}
+		return this.getStorageRepo(collection).compareAndDelete(id, expectedRevision);
+	}
+
 	async storageUpdateIf(
 		collection: string,
 		id: string,
@@ -465,13 +522,11 @@ export class PluginBridge extends WorkerEntrypoint<PluginBridgeEnv, PluginBridge
 		}
 		if (items.length === 0) return;
 
-		// D1 doesn't support batch in prepare, so we do individual inserts
-		// In future, we could use batch API
 		for (const item of items) {
 			await this.env.DB.prepare(
-				"INSERT OR REPLACE INTO _plugin_storage (plugin_id, collection, id, data, updated_at) VALUES (?, ?, ?, ?, datetime('now'))",
+				"INSERT OR REPLACE INTO _plugin_storage (plugin_id, collection, id, data, revision, updated_at) VALUES (?, ?, ?, ?, ?, datetime('now'))",
 			)
-				.bind(pluginId, collection, item.id, JSON.stringify(item.data))
+				.bind(pluginId, collection, item.id, JSON.stringify(item.data), crypto.randomUUID())
 				.run();
 		}
 	}

packages/cloudflare/src/sandbox/types.ts

@@ -3,7 +3,14 @@
  */
 
 import type { D1Database, R2Bucket } from "@cloudflare/workers-types";
-import type { ContentCreateOptions, UpdateIfArgs, UpdateIfResult } from "emdash";
+import type {
+	ConditionalDeleteResult,
+	ConditionalWriteResult,
+	ContentCreateOptions,
+	UpdateIfArgs,
+	UpdateIfResult,
+	VersionedValue,
+} from "emdash";
 
 /**
  * Environment bindings required for sandbox runner.
@@ -171,11 +178,30 @@ export interface PluginBridgeBinding {
 	// KV
 	kvGet(key: string): Promise<unknown>;
 	kvSet(key: string, value: unknown): Promise<void>;
+	kvGetVersioned(key: string): Promise<VersionedValue | null>;
+	kvCompareAndSet(
+		key: string,
+		expectedRevision: string | null,
+		value: unknown,
+	): Promise<ConditionalWriteResult>;
+	kvCompareAndDelete(key: string, expectedRevision: string): Promise<ConditionalDeleteResult>;
 	kvDelete(key: string): Promise<boolean>;
 	kvList(prefix?: string): Promise<Array<{ key: string; value: unknown }>>;
 	// Storage
 	storageGet(collection: string, id: string): Promise<unknown>;
 	storagePut(collection: string, id: string, data: unknown): Promise<void>;
+	storageGetVersioned(collection: string, id: string): Promise<VersionedValue | null>;
+	storageCompareAndSet(
+		collection: string,
+		id: string,
+		expectedRevision: string | null,
+		data: unknown,
+	): Promise<ConditionalWriteResult>;
+	storageCompareAndDelete(
+		collection: string,
+		id: string,
+		expectedRevision: string,
+	): Promise<ConditionalDeleteResult>;
 	storageUpdateIf(
 		collection: string,
 		id: string,

packages/cloudflare/src/sandbox/wrapper.ts

@@ -106,6 +106,9 @@ function createContext(env) {
 	const kv = {
 		get: (key) => bridge.kvGet(key),
 		set: (key, value) => bridge.kvSet(key, value),
+		getVersioned: (key) => bridge.kvGetVersioned(key),
+		compareAndSet: (key, expectedRevision, value) => bridge.kvCompareAndSet(key, expectedRevision, value),
+		compareAndDelete: (key, expectedRevision) => bridge.kvCompareAndDelete(key, expectedRevision),
 		delete: (key) => bridge.kvDelete(key),
 		list: (prefix) => bridge.kvList(prefix)
 	};
@@ -115,6 +118,9 @@ function createContext(env) {
 		return {
 			get: (id) => bridge.storageGet(collectionName, id),
 			put: (id, data) => bridge.storagePut(collectionName, id, data),
+			getVersioned: (id) => bridge.storageGetVersioned(collectionName, id),
+			compareAndSet: (id, expectedRevision, data) => bridge.storageCompareAndSet(collectionName, id, expectedRevision, data),
+			compareAndDelete: (id, expectedRevision) => bridge.storageCompareAndDelete(collectionName, id, expectedRevision),
 			updateIf: async (id, args) => {
 				const result = await bridge.storageUpdateIf(collectionName, id, args);
 				if (result && typeof result === "object" && "__emdashStorageError" in result) {

packages/core/src/database/migrations/077_plugin_storage_revisions.ts

@@ -0,0 +1,89 @@
+import { type Kysely, sql } from "kysely";
+
+import { columnExists, isPostgres } from "../dialect-helpers.js";
+
+const TABLES = [
+	{ name: "options", keys: ["name"] },
+	{ name: "_plugin_storage", keys: ["plugin_id", "collection", "id"] },
+] as const;
+
+const DUPLICATE_COLUMN_REGEX =
+	/(?:duplicate column|column .* already exists|already exists.*column)/i;
+
+export async function up(db: Kysely<unknown>): Promise<void> {
+	for (const table of TABLES) {
+		if (await columnExists(db, table.name, "revision")) continue;
+		try {
+			await db.schema
+				.alterTable(table.name)
+				.addColumn("revision", "text", (column) => column.notNull().defaultTo("0"))
+				.execute();
+		} catch (error) {
+			if (isDuplicateColumnError(error) && (await columnExists(db, table.name, "revision"))) {
+				continue;
+			}
+			throw error;
+		}
+	}
+
+	if (isPostgres(db)) {
+		await sql`
+			CREATE OR REPLACE FUNCTION emdash_plugin_storage_assign_revision()
+			RETURNS trigger
+			LANGUAGE plpgsql
+			AS $$
+			BEGIN
+				IF TG_OP = 'INSERT' THEN
+					IF NEW.revision = '0' THEN
+						NEW.revision := gen_random_uuid()::text;
+					END IF;
+				ELSIF NEW.revision = '0' OR NEW.revision = OLD.revision THEN
+					NEW.revision := gen_random_uuid()::text;
+				END IF;
+				RETURN NEW;
+			END;
+			$$
+		`.execute(db);
+		for (const table of TABLES) {
+			await sql`
+				CREATE OR REPLACE TRIGGER ${sql.ref(`emdash_${table.name}_revision`)}
+				BEFORE INSERT OR UPDATE ON ${sql.ref(table.name)}
+				FOR EACH ROW EXECUTE FUNCTION emdash_plugin_storage_assign_revision()
+			`.execute(db);
+		}
+	} else {
+		for (const table of TABLES) {
+			const rowKey = sql.join(
+				table.keys.map((key) => sql`${sql.ref(key)} = ${sql.ref(`NEW.${key}`)}`),
+				sql` AND `,
+			);
+			await sql`
+				CREATE TRIGGER IF NOT EXISTS ${sql.ref(`emdash_${table.name}_revision_insert`)}
+				AFTER INSERT ON ${sql.ref(table.name)}
+				WHEN NEW.revision = '0'
+				BEGIN
+					UPDATE ${sql.ref(table.name)} SET revision = lower(hex(randomblob(16)))
+					WHERE ${rowKey} AND revision = NEW.revision;
+				END
+			`.execute(db);
+			await sql`
+				CREATE TRIGGER IF NOT EXISTS ${sql.ref(`emdash_${table.name}_revision_update`)}
+				AFTER UPDATE ON ${sql.ref(table.name)}
+				WHEN NEW.revision = '0' OR NEW.revision = OLD.revision
+				BEGIN
+					UPDATE ${sql.ref(table.name)} SET revision = lower(hex(randomblob(16)))
+					WHERE ${rowKey} AND revision = NEW.revision;
+				END
+			`.execute(db);
+		}
+	}
+}
+
+function isDuplicateColumnError(error: unknown): boolean {
+	if (!(error instanceof Error)) return false;
+	return DUPLICATE_COLUMN_REGEX.test(error.message) || isDuplicateColumnError(error.cause);
+}
+
+export async function down(_db: Kysely<unknown>): Promise<void> {
+	// Revisions must survive a host rollback while other writers still use them.
+}

packages/core/src/database/migrations/runner.ts

@@ -79,6 +79,7 @@ import * as m073 from "./073_media_focal_point.js";
 import * as m074 from "./074_content_deleted_scheduled_index.js";
 import * as m075 from "./075_entry_edit_locks.js";
 import * as m076 from "./076_collection_nav_group.js";
+import * as m077 from "./077_plugin_storage_revisions.js";
 
 const MIGRATIONS: Readonly<Record<string, Migration>> = Object.freeze({
 	"001_initial": m001,
@@ -156,6 +157,7 @@ const MIGRATIONS: Readonly<Record<string, Migration>> = Object.freeze({
 	"074_content_deleted_scheduled_index": m074,
 	"075_entry_edit_locks": m075,
 	"076_collection_nav_group": m076,
+	"077_plugin_storage_revisions": m077,
 });
 
 /** Ordered names from the statically registered migration set. */