Add an injectable blob backend with an R2 store for the Cloudflare hosts

RhysSullivan · RhysSullivan · commit fa6a3ee5e54a · 2026-06-11T19:09:26.000-07:00
The plugin blob seam (StorageDeps.blobs) was hardcoded to the FumaDB blob
table, so every blob write landed in the relational tier. createExecutor
now accepts a BlobStore, the host DB handle can carry one, and
makeScopedExecutor threads it through.

The SDK keeps only the platform-neutral contract and backends; the R2
implementation lives with the Cloudflare host package
(@executor-js/cloudflare/blob-store), typed against the real R2Bucket.

cloud gains a BLOBS R2 binding (bucket must exist before deploy);
host-cloudflare reuses its existing BLOBS bucket directly for blob-seam
writes, which never enter D1 and so never need the oversized-value
offload wrap. Hosts without a bucket keep the FumaDB fallback. The prod
blob table is empty, so swapping the backend strands no data.
diff --git a/apps/cloud/src/db/fuma.ts b/apps/cloud/src/db/fuma.ts
@@ -1,3 +1,4 @@
+import { env } from "cloudflare:workers";
 import { Effect, Layer } from "effect";
 import { type FumaDB } from "@executor-js/fumadb";
 import { type DrizzleConfig } from "@executor-js/fumadb/adapters/drizzle";
@@ -9,6 +10,7 @@ import {
   type ExecutorDbHandle,
   type ExecutorDbProvider,
 } from "@executor-js/api/server";
+import { makeR2BlobStore } from "@executor-js/cloudflare/blob-store";
 import type { FumaDb, FumaTables } from "@executor-js/sdk";
 
 import { DbService } from "./db";
@@ -66,6 +68,10 @@ export const cloudDbProviderLayer = (
         db: fuma.db,
         fuma: fuma.fuma,
         close: async () => {},
+        // Plugin blobs (multi-MB resolved specs) live in R2, not Postgres.
+        // Guarded because test workers / local dev may run without the
+        // binding — the executor then falls back to the FumaDB `blob` table.
+        blobs: env.BLOBS ? makeR2BlobStore(env.BLOBS) : undefined,
       };
     }),
   );
diff --git a/apps/cloud/src/env-augment.d.ts b/apps/cloud/src/env-augment.d.ts
@@ -20,6 +20,14 @@ declare global {
       DATABASE_URL?: string;
       EXECUTOR_DIRECT_DATABASE_URL?: string;
 
+      // Plugin blob seam backend (wrangler.jsonc `r2_buckets`). Declared here
+      // (optional) rather than regenerating worker-configuration.d.ts: test
+      // workers and older local setups run without the binding, and the db
+      // layer falls back to the Postgres `blob` table when absent. Typed via
+      // @cloudflare/workers-types (not the wrangler-generated global) to match
+      // what `@executor-js/cloudflare/blob-store` accepts.
+      BLOBS?: import("@cloudflare/workers-types").R2Bucket;
+
       // SSRF / private-network egress guard. Unset in production -> the guard is
       // ON; the test workers set "true" so fixtures can reach localhost.
       ALLOW_LOCAL_NETWORK?: string;
diff --git a/apps/cloud/wrangler.jsonc b/apps/cloud/wrangler.jsonc
@@ -43,6 +43,14 @@
       "localConnectionString": "postgresql://postgres:postgres@127.0.0.1:5433/postgres",
     },
   ],
+  // Plugin blob seam backend (multi-MB resolved specs). Bucket must exist
+  // before deploy: `wrangler r2 bucket create executor-cloud-blobs`.
+  "r2_buckets": [
+    {
+      "binding": "BLOBS",
+      "bucket_name": "executor-cloud-blobs",
+    },
+  ],
   "placement": {
     "region": "aws:us-east-1",
   },
diff --git a/apps/host-cloudflare/src/db/d1.ts b/apps/host-cloudflare/src/db/d1.ts
@@ -12,6 +12,7 @@ import {
   createExecutorFumaDb,
   type ExecutorDbHandle,
 } from "@executor-js/api/server";
+import { makeR2BlobStore } from "@executor-js/cloudflare/blob-store";
 
 import { CLOUDFLARE_NAMESPACE, CLOUDFLARE_SCHEMA_VERSION } from "../config";
 
@@ -69,5 +70,9 @@ export const createD1ExecutorDb = async (
     fuma,
     // The D1 binding owns its own lifecycle; nothing to release.
     close: async () => {},
+    // Blob-seam writes go straight to R2 — they never enter D1, so they never
+    // need the offload wrap above (which remains only for legacy oversized
+    // values already inlined in D1 rows).
+    blobs: blobs ? makeR2BlobStore(blobs) : undefined,
   };
 };
diff --git a/bun.lock b/bun.lock
diff --git a/packages/core/api/src/server/scoped-executor.ts b/packages/core/api/src/server/scoped-executor.ts
@@ -178,7 +178,7 @@ export const makeScopedExecutor = <
   _organizationName: string,
 ): Effect.Effect<Executor<TPlugins>, StorageFailure, DbProvider | PluginsProvider | HostConfig> =>
   Effect.gen(function* () {
-    const { db } = yield* DbProvider;
+    const { db, blobs } = yield* DbProvider;
     const { plugins: pluginsFactory } = yield* PluginsProvider;
     const config = yield* HostConfig;
     // Explicit config wins; otherwise fall back to the request origin if a host
@@ -219,6 +219,7 @@ export const makeScopedExecutor = <
       tenant: Tenant.make(organizationId),
       subject: Subject.make(accountId),
       db,
+      blobs,
       plugins,
       httpClientLayer,
       onElicitation: "accept-all",
diff --git a/packages/core/sdk/src/executor-fuma-db.ts b/packages/core/sdk/src/executor-fuma-db.ts
@@ -23,6 +23,7 @@ import { type DrizzleRuntimeProvider } from "@executor-js/fumadb/adapters/drizzl
 import { drizzleAdapter } from "@executor-js/fumadb/adapters/drizzle";
 import { schema as fumaSchema, type RelationsMap } from "@executor-js/fumadb/schema";
 
+import type { BlobStore } from "./blob";
 import type { FumaDb, FumaTables } from "./fuma-runtime";
 
 // The FumaDB provider both the runtime-schema generator and the drizzle adapter
@@ -106,4 +107,11 @@ export interface ExecutorDbHandle<
   TTables extends FumaTables = FumaTables,
 > extends ExecutorFumaDb<TTables> {
   readonly close: () => Promise<void>;
+  /**
+   * Optional blob backend assembled alongside the driver (an R2 bucket on the
+   * Cloudflare hosts). `makeScopedExecutor` threads it into
+   * `createExecutor({ blobs })`; absent, the executor defaults to the FumaDB
+   * `blob` table over `db`.
+   */
+  readonly blobs?: BlobStore;
 }
diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts
@@ -16,7 +16,7 @@ import {
   type FumaTables,
   type StorageFailure,
 } from "./fuma-runtime";
-import { makeFumaBlobStore, pluginBlobStore, type OwnerPartitions } from "./blob";
+import { makeFumaBlobStore, pluginBlobStore, type BlobStore, type OwnerPartitions } from "./blob";
 import { coreToolsPlugin } from "./core-tools";
 import type {
   Connection,
@@ -328,6 +328,13 @@ export interface ExecutorConfig<TPlugins extends readonly AnyPlugin[] = readonly
   /** The acting member, or omit for a pure-org executor (no `owner:"user"`). */
   readonly subject?: Subject;
   readonly db?: ExecutorDbInput | ExecutorDbFactory;
+  /**
+   * Backend for the plugin blob seam (`StorageDeps.blobs`). Defaults to the
+   * FumaDB `blob` table over `db`. Hosts with an object store hand one in
+   * (e.g. the R2 store in `@executor-js/cloudflare/blob-store`) so multi-MB
+   * values stay out of the relational tier.
+   */
+  readonly blobs?: BlobStore;
   readonly plugins?: TPlugins;
   /** Config-level credential providers, merged with every
    *  `plugin.credentialProviders`. Config providers register first, so the
@@ -1217,7 +1224,7 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
     const rootDb = withQueryContext(rootDbUntyped, ownerContext);
     const fuma = makeFumaClient(rootDb);
     const core = makeCoreDb(fuma);
-    const blobs = makeFumaBlobStore(fuma);
+    const blobs = config.blobs ?? makeFumaBlobStore(fuma);
     const transaction = <A, E>(effect: Effect.Effect<A, E>) => fuma.transaction(effect);
 
     // Populated once, never mutated after startup.
diff --git a/packages/core/sdk/src/index.ts b/packages/core/sdk/src/index.ts
@@ -166,8 +166,10 @@ export {
   type InvokeOptions,
 } from "./elicitation";
 
-// Blob store — the plugin-facing CONTRACT only. The concrete makers
-// (`makeFumaBlobStore`/`makeInMemoryBlobStore`) are SDK-internal.
+// Blob store — the plugin-facing contract (`BlobStore`/`PluginBlobStore`)
+// plus the platform-neutral backends (`makeFumaBlobStore` default,
+// `makeInMemoryBlobStore` for tests). Platform-specific backends live with
+// their host (R2: `@executor-js/cloudflare/blob-store`).
 export {
   pluginBlobStore,
   makeInMemoryBlobStore,
diff --git a/packages/hosts/cloudflare/package.json b/packages/hosts/cloudflare/package.json
@@ -4,6 +4,10 @@
   "private": true,
   "type": "module",
   "exports": {
+    "./blob-store": {
+      "types": "./src/blob-store.ts",
+      "default": "./src/blob-store.ts"
+    },
     "./mcp/worker-transport": {
       "types": "./src/mcp/worker-transport.ts",
       "default": "./src/mcp/worker-transport.ts"
@@ -38,6 +42,7 @@
     "@executor-js/api": "workspace:*",
     "@executor-js/execution": "workspace:*",
     "@executor-js/host-mcp": "workspace:*",
+    "@executor-js/sdk": "workspace:*",
     "@modelcontextprotocol/sdk": "^1.29.0",
     "agents": "^0.10.0",
     "effect": "catalog:"
diff --git a/packages/hosts/cloudflare/src/blob-store.test.ts b/packages/hosts/cloudflare/src/blob-store.test.ts
@@ -0,0 +1,105 @@
+import { describe, expect, it } from "@effect/vitest";
+import { Effect } from "effect";
+import type { R2Bucket } from "@cloudflare/workers-types";
+
+import { StorageError, pluginBlobStore } from "@executor-js/sdk";
+
+import { makeR2BlobStore } from "./blob-store";
+
+// Minimal in-memory R2 bucket double — covers the get/put/delete/head slice
+// the store uses. Typed via the boundary cast because a full R2Bucket carries
+// a dozen unrelated methods.
+const makeFakeBucket = (): { bucket: R2Bucket; objects: Map<string, string> } => {
+  const objects = new Map<string, string>();
+  // oxlint-disable-next-line executor/no-double-cast -- test double: only the slice the store calls is implemented
+  const bucket = {
+    get: async (key: string) => {
+      const value = objects.get(key);
+      return value === undefined ? null : { text: async () => value };
+    },
+    put: async (key: string, value: string) => {
+      objects.set(key, value);
+    },
+    delete: async (key: string) => {
+      objects.delete(key);
+    },
+    head: async (key: string) => (objects.has(key) ? {} : null),
+  } as unknown as R2Bucket;
+  return { bucket, objects };
+};
+
+describe("makeR2BlobStore", () => {
+  it.effect("round-trips a value and names objects `namespace/key`", () =>
+    Effect.gen(function* () {
+      const { bucket, objects } = makeFakeBucket();
+      const store = makeR2BlobStore(bucket);
+
+      yield* store.put("o:t1/openapi", "spec/abc123", "spec-text");
+      expect(yield* store.get("o:t1/openapi", "spec/abc123")).toBe("spec-text");
+      expect(objects.has("o:t1/openapi/spec/abc123")).toBe(true);
+    }),
+  );
+
+  it.effect("get returns null for a missing object", () =>
+    Effect.gen(function* () {
+      const { bucket } = makeFakeBucket();
+      const store = makeR2BlobStore(bucket);
+      expect(yield* store.get("o:t1/openapi", "missing")).toBeNull();
+    }),
+  );
+
+  it.effect("getMany returns hits keyed by namespace", () =>
+    Effect.gen(function* () {
+      const { bucket } = makeFakeBucket();
+      const store = makeR2BlobStore(bucket);
+      yield* store.put("u:t1:s1/openapi", "k", "user-value");
+      yield* store.put("o:t1/openapi", "k", "org-value");
+
+      const hits = yield* store.getMany(["u:t1:s1/openapi", "o:t1/openapi", "o:t2/openapi"], "k");
+      expect(hits.size).toBe(2);
+      expect(hits.get("u:t1:s1/openapi")).toBe("user-value");
+      expect(hits.get("o:t1/openapi")).toBe("org-value");
+    }),
+  );
+
+  it.effect("delete and has agree", () =>
+    Effect.gen(function* () {
+      const { bucket } = makeFakeBucket();
+      const store = makeR2BlobStore(bucket);
+      yield* store.put("ns/p", "k", "v");
+      expect(yield* store.has("ns/p", "k")).toBe(true);
+      yield* store.delete("ns/p", "k");
+      expect(yield* store.has("ns/p", "k")).toBe(false);
+      expect(yield* store.get("ns/p", "k")).toBeNull();
+    }),
+  );
+
+  it.effect("bucket failures surface as StorageError", () =>
+    Effect.gen(function* () {
+      // oxlint-disable-next-line executor/no-double-cast -- test double: only the slice the store calls is implemented
+      const failing = {
+        get: async () => {
+          // oxlint-disable-next-line executor/no-try-catch-or-throw -- test double simulating a bucket outage
+          throw { name: "R2Error", message: "bucket unavailable" };
+        },
+      } as unknown as R2Bucket;
+      const store = makeR2BlobStore(failing);
+      const err = yield* store.get("ns", "k").pipe(Effect.flip);
+      expect(err).toBeInstanceOf(StorageError);
+    }),
+  );
+
+  it.effect("works behind pluginBlobStore owner partitioning", () =>
+    Effect.gen(function* () {
+      const { bucket } = makeFakeBucket();
+      const store = makeR2BlobStore(bucket);
+      const plugin = pluginBlobStore(store, { org: "o:t1", user: "u:t1:s1" }, "openapi");
+
+      yield* plugin.put("spec/h1", "org-spec", { owner: "org" });
+      expect(yield* plugin.get("spec/h1")).toBe("org-spec");
+
+      yield* plugin.put("spec/h1", "user-spec", { owner: "user" });
+      expect(yield* plugin.get("spec/h1")).toBe("user-spec");
+    }),
+  );
+});
diff --git a/packages/hosts/cloudflare/src/blob-store.ts b/packages/hosts/cloudflare/src/blob-store.ts
@@ -0,0 +1,70 @@
+// ---------------------------------------------------------------------------
+// BlobStore over a Cloudflare R2 bucket — the object-store backend for the
+// SDK's blob seam on the Cloudflare hosts. The SDK defines only the
+// `BlobStore` contract (plus its FumaDB/in-memory defaults); this R2 binding
+// lives here so the SDK stays platform-agnostic.
+//
+// Object name: `${namespace}/${key}`. Unambiguous because a namespace is
+// always `partition/pluginId` (exactly one slash; partitions use `:`
+// separators, plugin ids contain no slash), so the first two segments always
+// recover the namespace and the rest is the key.
+//
+// Unlike `makeFumaBlobStore`, writes do NOT participate in FumaDB
+// transactions — a rolled-back transaction leaves the blob behind. Callers
+// should use idempotent (content-derived) keys so orphaned writes are
+// harmless and re-puts are no-ops in effect.
+// ---------------------------------------------------------------------------
+
+import { Effect } from "effect";
+import type { R2Bucket } from "@cloudflare/workers-types";
+
+import { StorageError, type BlobStore } from "@executor-js/sdk";
+
+const objectName = (namespace: string, key: string): string => `${namespace}/${key}`;
+
+const storeError = (op: string) => (cause: unknown) =>
+  new StorageError({ message: `R2 blob ${op} failed`, cause });
+
+export const makeR2BlobStore = (bucket: R2Bucket): BlobStore => ({
+  get: (namespace, key) =>
+    Effect.tryPromise({
+      try: async () => {
+        const object = await bucket.get(objectName(namespace, key));
+        return object == null ? null : await object.text();
+      },
+      catch: storeError("get"),
+    }),
+  // R2 has no multi-get; fetch the (at most two — user + org partition)
+  // namespaces concurrently.
+  getMany: (namespaces, key) =>
+    Effect.tryPromise({
+      try: async () => {
+        const hits = new Map<string, string>();
+        await Promise.all(
+          namespaces.map(async (namespace) => {
+            const object = await bucket.get(objectName(namespace, key));
+            if (object != null) hits.set(namespace, await object.text());
+          }),
+        );
+        return hits;
+      },
+      catch: storeError("getMany"),
+    }),
+  put: (namespace, key, value) =>
+    Effect.tryPromise({
+      try: async () => {
+        await bucket.put(objectName(namespace, key), value);
+      },
+      catch: storeError("put"),
+    }),
+  delete: (namespace, key) =>
+    Effect.tryPromise({
+      try: () => bucket.delete(objectName(namespace, key)),
+      catch: storeError("delete"),
+    }),
+  has: (namespace, key) =>
+    Effect.tryPromise({
+      try: async () => (await bucket.head(objectName(namespace, key))) != null,
+      catch: storeError("has"),
+    }),
+});
