Add a streaming backfill that moves inline specs into the blob store

Walks integration rows whose config still inlines spec text (openapi
spec / graphql introspectionJson), writes each to the blob backend under
the runtime's exact object naming (o:<tenant>/<plugin>/spec/<sha256>),
verifies the round-trip, and rewrites the config to the hash pointer —
one row at a time, since the inline specs total ~500 MB and must never
be loaded at once.

The blob target must match the deployed worker: --target r2 --bucket
<name> writes through wrangler to the bucket the BLOBS binding reads;
--target db writes the Postgres blob table for setups without R2.
Idempotent (pointer rows plan zero work, blobs are content-addressed)
with --dry-run and --limit for a canary pass. Every rewrite is decoded
under the deployed runtime before anything is written.

Author: RhysSullivan

apps/cloud/package.json

@@ -17,6 +17,8 @@
     "db:migrate-auth:dev": "op run --env-file=.env.op -- bun run scripts/migrate-auth-configs.ts",
     "db:repartition-vault:prod": "op run --env-file=.env.production -- bun run scripts/repartition-vault-metadata.ts",
     "db:repartition-vault:dev": "op run --env-file=.env.op -- bun run scripts/repartition-vault-metadata.ts",
+    "db:migrate-specs:prod": "op run --env-file=.env.production -- bun run scripts/migrate-specs-to-blobs.ts",
+    "db:migrate-specs:dev": "op run --env-file=.env.op -- bun run scripts/migrate-specs-to-blobs.ts",
     "db:migrate:dev": "op run --env-file=.env.op -- bun --bun ../../node_modules/.bun/node_modules/drizzle-kit/bin.cjs migrate",
     "build": "node scripts/build.mjs",
     "preview": "vite preview",

apps/cloud/scripts/migrate-specs-to-blobs.ts

@@ -0,0 +1,220 @@
+/* oxlint-disable executor/no-try-catch-or-throw, executor/no-json-parse, executor/no-error-constructor -- boundary: out-of-band migration script over a raw postgres connection + wrangler subprocess */
+// ---------------------------------------------------------------------------
+// One-off data migration: move inline spec text out of `integration.config`
+// into the blob store the deployed worker reads.
+//
+//   openapi: config.spec             -> blob spec/<sha256>,          config.specHash
+//   graphql: config.introspectionJson -> blob introspection/<sha256>, config.introspectionHash
+//
+// Blob object names follow the runtime seam exactly
+// (`pluginBlobStore` + `makeR2BlobStore`): `o:<tenant>/<plugin>/<key>`.
+//
+// The blob TARGET must match what the worker is deployed with:
+//   --target r2 --bucket executor-cloud-blobs   # wrangler-authed R2 writes
+//   --target db                                  # the Postgres `blob` table
+// Run AFTER the worker that reads `specHash` is deployable, BEFORE relying
+// on it for latency (legacy inline rows keep working either way).
+//
+//   bun run db:migrate-specs:prod -- --target r2 --bucket executor-cloud-blobs
+//   bun run db:migrate-specs:dev -- --target db
+//
+// Rows are processed ONE AT A TIME (the inline specs total ~500 MB in prod;
+// never load them all). Idempotent: pointer-shaped rows plan zero work, blob
+// writes are content-addressed, and a re-run resumes where it left off.
+// Pass --dry-run to print the plan without writing, --limit N for a canary.
+// ---------------------------------------------------------------------------
+
+import { createHash } from "node:crypto";
+import { execFileSync } from "node:child_process";
+import { mkdtempSync, rmSync, writeFileSync, readFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+
+import postgres from "postgres";
+import { Option } from "effect";
+import { decodeOpenApiIntegrationConfig } from "@executor-js/plugin-openapi";
+import { decodeGraphqlIntegrationConfigOption } from "@executor-js/plugin-graphql";
+
+// --- per-plugin shape: which inline field moves, to which blob key prefix ---
+
+interface SpecField {
+  readonly inlineField: string;
+  readonly hashField: string;
+  readonly blobKeyPrefix: string;
+  readonly decodes: (config: unknown) => boolean;
+}
+
+const SPEC_FIELDS: Record<string, SpecField> = {
+  openapi: {
+    inlineField: "spec",
+    hashField: "specHash",
+    blobKeyPrefix: "spec",
+    decodes: (config) => decodeOpenApiIntegrationConfig(config) !== null,
+  },
+  graphql: {
+    inlineField: "introspectionJson",
+    hashField: "introspectionHash",
+    blobKeyPrefix: "introspection",
+    decodes: (config) => Option.isSome(decodeGraphqlIntegrationConfigOption(config)),
+  },
+};
+
+// --- args ---
+
+const argValue = (name: string): string | undefined => {
+  const index = process.argv.indexOf(name);
+  return index >= 0 ? process.argv[index + 1] : undefined;
+};
+
+const dryRun = process.argv.includes("--dry-run");
+const target = argValue("--target") ?? "r2";
+const bucket = argValue("--bucket");
+const limit = Number(argValue("--limit") ?? Infinity);
+
+if (target !== "r2" && target !== "db") {
+  console.error(`--target must be "r2" or "db", got "${target}"`);
+  process.exit(1);
+}
+if (target === "r2" && !bucket && !dryRun) {
+  console.error("--target r2 requires --bucket <name> (e.g. executor-cloud-blobs)");
+  process.exit(1);
+}
+
+const connectionString = process.env.DATABASE_URL;
+if (!connectionString) {
+  console.error("DATABASE_URL is not set");
+  process.exit(1);
+}
+
+// Direct (non-Hyperdrive) connection — PlanetScale requires TLS.
+const sql = postgres(connectionString, { max: 1, prepare: false, ssl: "require" });
+
+const sha256Hex = (text: string): string => createHash("sha256").update(text, "utf8").digest("hex");
+
+// Runtime object naming, byte-for-byte: `pluginBlobStore` namespaces by
+// `o:<tenant>/<pluginId>` and `makeR2BlobStore` joins `namespace/key`; the
+// FumaDB store keys rows by `JSON.stringify([namespace, key])`.
+const blobNamespace = (tenant: string, pluginId: string) => `o:${tenant}/${pluginId}`;
+const fumaBlobId = (namespace: string, key: string) => JSON.stringify([namespace, key]);
+
+// --- blob writers (verify-on-write: a pointer must never replace a spec the
+// --- store can't serve back) ---
+
+const tempDir = mkdtempSync(join(tmpdir(), "spec-blobs-"));
+
+const putBlobR2 = (objectName: string, value: string): void => {
+  const file = join(tempDir, "blob-upload");
+  writeFileSync(file, value, "utf8");
+  execFileSync(
+    "wrangler",
+    ["r2", "object", "put", `${bucket}/${objectName}`, "--file", file, "--remote"],
+    { stdio: ["ignore", "ignore", "inherit"] },
+  );
+  const verifyFile = join(tempDir, "blob-verify");
+  execFileSync(
+    "wrangler",
+    ["r2", "object", "get", `${bucket}/${objectName}`, "--file", verifyFile, "--remote"],
+    { stdio: ["ignore", "ignore", "inherit"] },
+  );
+  const roundTripped = readFileSync(verifyFile, "utf8");
+  if (sha256Hex(roundTripped) !== sha256Hex(value)) {
+    throw new Error(`R2 round-trip mismatch for ${objectName}`);
+  }
+};
+
+const putBlobDb = async (namespace: string, key: string, value: string): Promise<void> => {
+  const id = fumaBlobId(namespace, key);
+  await sql`
+    INSERT INTO blob (namespace, key, value, id)
+    VALUES (${namespace}, ${key}, ${value}, ${id})
+    ON CONFLICT (id) DO UPDATE SET value = EXCLUDED.value
+  `;
+};
+
+// --- main ---
+
+try {
+  // Plan pass: metadata only — never load every config at once.
+  const candidates = await sql<
+    { row_id: string; tenant: string; slug: string; plugin_id: string; config_bytes: number }[]
+  >`
+    SELECT row_id, tenant, slug, plugin_id, length(config::text) AS config_bytes
+    FROM integration
+    WHERE (plugin_id = 'openapi' AND config::jsonb ? 'spec')
+       OR (plugin_id = 'graphql' AND config::jsonb ? 'introspectionJson')
+    ORDER BY row_id
+  `;
+
+  const totalBytes = candidates.reduce((sum, row) => sum + Number(row.config_bytes), 0);
+  console.log(
+    `${candidates.length} row(s) carry inline spec text (${(totalBytes / 1024 / 1024).toFixed(1)} MB total config)`,
+  );
+
+  const work = candidates.slice(0, Number.isFinite(limit) ? limit : candidates.length);
+  if (work.length < candidates.length) console.log(`--limit: processing first ${work.length}`);
+
+  if (dryRun) {
+    for (const row of work) {
+      console.log(
+        `  would move ${row.plugin_id} ${row.tenant}/${row.slug} (${(Number(row.config_bytes) / 1024).toFixed(0)} kB) -> blob`,
+      );
+    }
+    process.exit(0);
+  }
+
+  let moved = 0;
+  const writtenObjects = new Set<string>();
+
+  for (const candidate of work) {
+    const field = SPEC_FIELDS[candidate.plugin_id];
+    if (!field) continue;
+
+    // Fetch this one row's config — one multi-MB value in memory at a time.
+    const [row] = await sql<{ config: unknown }[]>`
+      SELECT config FROM integration WHERE row_id = ${candidate.row_id}
+    `;
+    if (!row) continue;
+    const config = typeof row.config === "string" ? JSON.parse(row.config) : row.config;
+    if (typeof config !== "object" || config === null) continue;
+
+    const inline = (config as Record<string, unknown>)[field.inlineField];
+    if (typeof inline !== "string") continue; // already migrated (or foreign shape)
+
+    const hash = sha256Hex(inline);
+    const namespace = blobNamespace(candidate.tenant, candidate.plugin_id);
+    const blobKey = `${field.blobKeyPrefix}/${hash}`;
+    const objectName = `${namespace}/${blobKey}`;
+
+    // Audit BEFORE writing: the rewritten config must decode under the
+    // deployed runtime, or this row would read as "no usable config".
+    const { [field.inlineField]: _removed, ...rest } = config as Record<string, unknown>;
+    const nextConfig = { ...rest, [field.hashField]: hash };
+    if (!field.decodes(nextConfig)) {
+      console.error(`SKIP ${candidate.tenant}/${candidate.slug}: rewrite would not decode`);
+      continue;
+    }
+
+    // Content-addressed: identical specs (same tenant) write once.
+    if (!writtenObjects.has(objectName)) {
+      if (target === "r2") putBlobR2(objectName, inline);
+      else await putBlobDb(namespace, blobKey, inline);
+      writtenObjects.add(objectName);
+    }
+
+    await sql`
+      UPDATE integration
+      SET config = ${sql.json(nextConfig as never)}
+      WHERE row_id = ${candidate.row_id}
+    `;
+
+    moved += 1;
+    console.log(
+      `moved ${candidate.plugin_id} ${candidate.tenant}/${candidate.slug} -> ${field.blobKeyPrefix}/${hash.slice(0, 12)}… (${moved}/${work.length})`,
+    );
+  }
+
+  console.log(`done: ${moved} row(s) rewritten, ${writtenObjects.size} blob object(s) written`);
+} finally {
+  rmSync(tempDir, { recursive: true, force: true });
+  await sql.end();
+}