ストレージ

マニフェストでストレージを宣言する

サンドボックス型プラグインでは、storage は emdash-plugin.jsonc に書きます。プラグインがどのコレクションに触れてよいかをサンドボックスのブリッジが把握できるように、宣言はビルド時に見える必要があります。

{
	"slug": "forms",
	// ...identity + profile...
	"capabilities": ["content:read"],

	"storage": {
		"submissions": {
			"indexes": [
				"formId",
				"status",
				"createdAt",
				["formId", "createdAt"],
				["status", "createdAt"]
			]
		},
		"forms": {
			"indexes": ["slug"]
		}
	}
}

storage の各キーは、コレクション名です。indexes の配列には、効率よく検索できるフィールドを並べます。1つのフィールドのインデックスは文字列で、複合インデックスは文字列の配列で書きます。ルールの全体は、マニフェストのリファレンスを参照します。

コレクション名は英小文字で始まり、英小文字、数字、アンダースコアを使います。インデックスのフィールド名は英字で始まり、英字、数字、アンダースコアを使います。値が一意になるフィールドやフィールドの組み合わせは、uniqueIndexes に書きます。一意のインデックスはそれだけで検索に使えるため、indexes に重ねて書かないでください。

実行時にストレージを使う

src/plugin.ts では、ctx.storage を通じてコレクションにアクセスします。形は、マニフェストで宣言した内容と対応しています。

import type { SandboxedPlugin } from "emdash/plugin";

const plugin: SandboxedPlugin = {
	hooks: {
		"content:afterSave": {
			handler: async (event, ctx) => {
				const { submissions } = ctx.storage;

				await submissions.put("sub_123", {
					formId: "contact",
					email: "user@example.com",
					status: "pending",
					createdAt: new Date().toISOString(),
				});

				const item = await submissions.get("sub_123");
				ctx.log.info("Stored submission", { id: item?.formId });
			},
		},
	},
};

export default plugin;

マニフェストで宣言していないコレクションにアクセスすると、例外が発生します。ブリッジが実行時にこれを強制します。

コレクションのAPI

宣言した各コレクションには、次の読み取り、書き込み、一括処理、検索、件数の取得のメソッドがあります。

interface StorageCollection<T = unknown> {
	// Basic CRUD
	get(id: string): Promise<T | null>;
	put(id: string, data: T): Promise<void>;
	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>;
	deleteMany(ids: string[]): Promise<number>;

	// Query (indexed fields only)
	query(options?: QueryOptions): Promise<PaginatedResult<{ id: string; data: T }>>;
	count(where?: WhereClause): Promise<number>;
}

条件付き書き込み

同じレコードを複数のリクエストが同時に更新しうる場合は、getVersioned() と compareAndSet() を使います。これらのメソッドは、ネイティブ型とサンドボックス型のどちらのプラグインでも、宣言した ctx.storage のコレクションと ctx.kv で使えます。各操作は、呼び出したプラグインの名前空間の中の1つのキーにアクセスします。

各メソッドの動きは次のとおりです。

compareAndSet() が成功すると、{ applied: true, revision } を返します。前提条件を満たさなかった場合は { applied: false } を返します。不正な引数、権限の不足、データベースの障害の場合は、Promiseが拒否されます。compareAndDelete() は { applied: boolean } を返します。指定したキーがない場合でも、関係のない一意のインデックスの違反はエラーになります。

リビジョンは変更せずにそのまま渡し、そのリビジョンを取得したキーにだけ使います。値が同じ set()、put()、一括の書き込みを含め、すべての書き込みでリビジョンが変わります。キーを削除して作り直すと、以前のリビジョンは無効になります。

次の補助関数は、プラグインのカウンターに完了したジョブを1件加えます。別のリクエストが先に書き込んだ場合は、3回まで再試行します。

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");
}

競合した場合は、値を読み直してから、変更内容を計算し直します。再試行の回数には上限を設けます。レスポンスが失われると、書き込みの結果が分からなくなることがあります。これらのメソッドは、外部への操作や、再試行されたジョブの実行を、ちょうど1回だけにするものではありません。

原子性（atomicity）は1つのキーの範囲です。コンテンツのアイテムを読んでプラグインのレコードを書くことや、プラグインのレコードを2件書くことは、別々の操作です。一緒に変える必要があるフィールドは、1つの値にまとめます。ジョブの担当者や数量の上限などの業務上のルールは、その値を組み立てるときに守らせます。

条件付きのメソッドには、JavaScriptの文字列で1,024文字以内の空でないキーと、UTF-8でエンコードして1 MiB以内のJSONの値が必要です。リビジョンは、128文字以内の空でない文字列である必要があります。リビジョンを省略すると無効です。作成を求めるのは、明示的な null だけです。既存の条件なしのメソッドの動きは変わりません。

これらのメソッドを使う前に、対応するバージョンのコアとサンドボックスのアダプターをデプロイし、ホストのデータベースのマイグレーションを適用します。このマイグレーションは保存済みの値を保ちます。また、段階的なデプロイの途中で、古いホストのプロセスからの書き込みがリビジョンを無効にするようにします。

条件付き更新

保存されているフィールドが条件に一致する場合にだけ既存のドキュメントを変更するには、updateIf() を使います。データベースは、条件の確認と変更の適用を、1つの原子的な操作で実行します。このメソッドは、ネイティブ型プラグインと、CloudflareとWorkerdで動くサンドボックス型プラグインで使えます。

NumericDelta、UpdateIfArgs、UpdateIfResult の型は、emdash または emdash/plugin から import type でインポートします。

次の呼び出しは、保留中の送信データを承認し、同じ操作の中でレビュー回数を1増やします。

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 });
}

呼び出しが成功すると、更新後のドキュメント全体を含む { applied: true, data } を返します。ドキュメントがない場合や、条件に一致しない場合は { applied: false } を返します。ドキュメントを新しく挿入することはありません。

引数の動きは次のとおりです。

• where は必須で、検索の絞り込みと同じ演算子を使います。明示的に where: {} と書くと、フィールドの条件は追加されません。更新はIDで1つのドキュメントを対象にするため、条件に使うフィールドに、検索用のインデックスを宣言する必要はありません。
• 範囲の絞り込みには、少なくとも1つの境界を指定する必要があります。別の境界が指定されていれば、未定義の境界は無視されます。条件で使う数値のオペランドは有限の値である必要があります。
• set は、指定したトップレベルのフィールドの値をそれぞれ置き換え、ほかのフィールドは変えません。値はJSONにシリアライズできる必要があります。
• delta は、フィールドごとに { inc: number } か { dec: number } のどちらか1つだけを適用します。各オペランドは安全な整数（safe integer）である必要があります。負のオペランドも使えます。
• 同じフィールドを set と delta の両方に書くことはできません。どちらのオブジェクトでも、トップレベルの undefined の項目は無視されます。定義されたフィールドが少なくとも1つ残る必要があります。

更新の引数の形式が正しくない場合は、ドキュメントを変更せずにPromiseが拒否されます。引数のオブジェクト、set、delta、各deltaの操作は、プレーンなオブジェクトである必要があります。

整数のカウンター

deltaは、存在しないカウンターや null のカウンターを 0 から始めます。既存のカウンターとその計算結果は、Number.MIN_SAFE_INTEGER から Number.MAX_SAFE_INTEGER までの整数である必要があります。文字列、真偽値、オブジェクト、配列、小数、安全でない整数、範囲外の結果の場合は、更新全体が { applied: false } を返します。保存されているドキュメントがJSONのオブジェクトでない場合も { applied: false } を返します。どちらの場合も、フィールドは変更されません。

deltaの結果は負の値になることがあります。カウンターを0以上に保つには、n を減らす操作に、そのカウンターが n 以上であることを求める where の条件を組み合わせます。

シリアライゼーションの失敗を再試行する

PostgreSQLは、同時に実行された書き込みを、シリアライゼーションの失敗やデッドロックとして拒否することがあります。デッドロックは、READ COMMITTEDを含むどの分離レベルでも起こりえます。ネイティブ型プラグインでは、これらの失敗は StorageSerializationError としてスローされます。このエラーは、code: "STORAGE_SERIALIZATION_FAILURE"、retryable: true、省略可能な sqlState（40001 または 40P01）を持ちます。エラーのクラスは emdash からインポートします。

単独の呼び出しには、待ち時間を空けながら（backoff）、回数に上限を設けて再試行します。呼び出しが明示的なトランザクションの中にある場合は、読み取りも含めてトランザクション全体をやり直します。中断されたトランザクションの中で書き込みだけを再試行しても成功しません。{ applied: false } は、シリアライゼーションのエラーではなく、適用されなかった更新として扱います。

サンドボックスのトランスポートは、エラーの名前と再試行のメタデータを保ちますが、instanceof StorageSerializationError が成り立つことは保証しません。サンドボックスの境界をまたいでエラーを扱う場合は、code と retryable を確認します。

検索

query() は、インデックス付きのフィールドで絞り込んだ結果を、ページ分割して返します。

const result = await ctx.storage.submissions.query({
	where: {
		formId: "contact",
		status: "pending",
	},
	orderBy: { createdAt: "desc" },
	limit: 20,
});

// result.items   — Array<{ id, data }>
// result.cursor  — pagination cursor (if more results exist)
// result.hasMore — boolean

検索のオプション

結果の絞り込み、並べ替え、ページ分割には、次のオプションを query() に渡します。

interface QueryOptions {
	where?: WhereClause;
	orderBy?: Record<string, "asc" | "desc">;
	limit?: number;     // default 50, max 100
	cursor?: string;    // for pagination
}

where句の演算子

インデックス付きのフィールドを、次の演算子で絞り込みます。

where: {
	status: "pending",     // exact string match
	count: 5,              // exact number match
	archived: false,       // exact boolean match
}

where: {
	createdAt: { gte: "2024-01-01" },
	score: { gt: 50, lte: 100 },
}
// Available: gt, gte, lt, lte

where: {
	status: { in: ["pending", "approved"] },
}

where: {
	slug: { startsWith: "blog-" },
}

並べ替え

1つ以上のインデックス付きのフィールドに、昇順または降順を指定します。

orderBy: { createdAt: "desc" }   // newest first
orderBy: { score: "asc" }        // lowest first

ページ分割

カーソルを最後まで読み進めると、一致するすべてのアイテムを順に取得できます。

async function getAllSubmissions(ctx: PluginContext) {
	const all: Array<{ id: string; data: unknown }> = [];
	let cursor: string | undefined;

	do {
		const result = await ctx.storage.submissions.query({
			orderBy: { createdAt: "desc" },
			limit: 100,
			cursor,
		});
		all.push(...result.items);
		cursor = result.cursor;
	} while (cursor);

	return all;
}

件数の取得

コレクションのすべてのレコード、またはインデックス付きのフィールドに一致するレコードだけを数えます。

const total = await ctx.storage.submissions.count();

const pending = await ctx.storage.submissions.count({
	status: "pending",
});

一括処理

1つの処理で、IDが分かっている複数のレコードを読み取り、書き込み、削除する場合は、一括処理のメソッドを使います。

const items = await ctx.storage.submissions.getMany(["sub_1", "sub_2", "sub_3"]);
// Returns Map<string, T>

await ctx.storage.submissions.putMany([
	{ id: "sub_1", data: { formId: "contact", status: "new" } },
	{ id: "sub_2", data: { formId: "contact", status: "new" } },
]);

const deletedCount = await ctx.storage.submissions.deleteMany(["sub_1", "sub_2"]);

インデックスの設計

インデックスは、実際の検索のパターンに合わせて選びます。

複合インデックスは、1つ目のフィールドで絞り込み、必要に応じて2つ目のフィールドで並べ替える検索に使われます。

// With index ["formId", "createdAt"]:
query({ where: { formId: "contact" }, orderBy: { createdAt: "desc" } });  // uses index
query({ where: { formId: "contact" } });                                  // uses index (filter only)
query({ where: { createdAt: { gte: "2024-01-01" } } });                   // does NOT use this composite — filter starts at the wrong field

indexes または uniqueIndexes のどこかに書いたフィールドは、すべて検索のAPIのインデックス付きフィールドの確認を通ります。ただし、データベースがどの形の検索を効率よく実行できるかは、複合インデックスのフィールドの順序で決まります。プラグインが formId なしで createdAt による絞り込みや並べ替えをよく使う場合は、別に "createdAt" のインデックスを追加します。

型の安全性

アイテムの形に対してIntelliSenseを効かせるには、コレクションへのアクセスをキャストします。

import type { SandboxedPlugin } from "emdash/plugin";
import type { StorageCollection } from "emdash";

interface Submission {
	formId: string;
	email: string;
	data: Record<string, unknown>;
	status: "pending" | "approved" | "spam";
	createdAt: string;
}

const plugin: SandboxedPlugin = {
	hooks: {
		"content:afterSave": {
			handler: async (event, ctx) => {
				const submissions = ctx.storage.submissions as StorageCollection<Submission>;

				await submissions.put(`sub_${Date.now()}`, {
					formId: "contact",
					email: "user@example.com",
					data: { message: "Hello" },
					status: "pending",
					createdAt: new Date().toISOString(),
				});
			},
		},
	},
};

export default plugin;

どちらのインポートも型だけのものなので、サンドボックス型プラグインは実行時に emdash に依存しません。

ストレージ・コンテンツ・KVの使い分け

データの種類ごとに、適した仕組みを選びます。

サイトの編集者が、通常のコンテンツエディターを通じて管理画面でデータを表示・編集する必要がある場合は、代わりにサイトのコレクションを作成します。

コレクションの分離の仕組み

EmDashは、プラグインのドキュメントを、プラグインID、コレクション名、レコードID、JSONのデータ、タイムスタンプとともに保存します。これらの名前空間の列は、すべてのキーとインデックスに含まれます。プラグインは、マニフェストにあるコレクションのアクセサーだけを受け取り、サンドボックスのブリッジはそれ以外のコレクションへのアクセスを拒否します。

宣言したフィールドは、プラグインとコレクションの名前空間とあわせて、式インデックス（expression index）になります。EmDashは、SQLite、D1、PostgreSQLそれぞれの方言のSQLを生成します。プラグインのコードは、どのデータベースでも同じコレクションのAPIを使います。

インデックスの追加

プラグインの更新でインデックスを追加すると、EmDashは次にプラグインを読み込むときにインデックスを作成します。既存のレコードに重複した値がある間は、一意のインデックスを作成できません。そのため、その変更をリリースする前に、重複を確認して解消します。

更新でインデックスを削除すると、EmDashはそのインデックスを削除します。その後、そのフィールドを使う検索や並べ替えが残っていると、検証が失敗します。コードとマニフェストは一緒に更新します。

インデックスは、マニフェストのストレージに関する信頼の取り決めの一部です。インデックスを追加、削除、変更するときは、必ずプラグインのバージョンを上げます。変更が既存の検索や一意性の前提を壊す場合は、メジャーバージョンを上げます。