Move inline specs to the blob table at boot on the libSQL hosts

apps/host-selfhost/src/db/data-migrations.ts

@@ -7,7 +7,11 @@
 
 import { sqliteDataMigration, type SqliteDataMigration } from "@executor-js/sdk";
 import { runSqliteAuthConfigMigration } from "@executor-js/sdk/http-auth";
-import { openApiOutputSchemaDataMigration } from "@executor-js/plugin-openapi";
+import {
+  openApiOutputSchemaDataMigration,
+  openApiSpecBlobDataMigration,
+} from "@executor-js/plugin-openapi";
+import { graphqlIntrospectionBlobDataMigration } from "@executor-js/plugin-graphql";
 
 import { authConfigTransforms } from "./auth-config-migration";
 
@@ -20,4 +24,9 @@ export const selfHostDataMigrations: readonly SqliteDataMigration[] = [
   // Unwrap the retired {status, headers, data} transport envelope from
   // persisted openapi tool output schemas (mirrors cloud's drizzle 0002).
   openApiOutputSchemaDataMigration,
+  // Move inline spec / introspection text out of integration.config into the
+  // blob table (config keeps the content hash). Mirrors cloud's
+  // migrate-specs-to-blobs script.
+  openApiSpecBlobDataMigration,
+  graphqlIntrospectionBlobDataMigration,
 ];

apps/local/src/db/data-migrations.ts

@@ -7,7 +7,11 @@
 
 import { sqliteDataMigration, type SqliteDataMigration } from "@executor-js/sdk";
 import { runSqliteAuthConfigMigration } from "@executor-js/sdk/http-auth";
-import { openApiOutputSchemaDataMigration } from "@executor-js/plugin-openapi";
+import {
+  openApiOutputSchemaDataMigration,
+  openApiSpecBlobDataMigration,
+} from "@executor-js/plugin-openapi";
+import { graphqlIntrospectionBlobDataMigration } from "@executor-js/plugin-graphql";
 
 import { authConfigTransforms } from "./auth-config-migration";
 
@@ -20,4 +24,9 @@ export const localDataMigrations: readonly SqliteDataMigration[] = [
   // Unwrap the retired {status, headers, data} transport envelope from
   // persisted openapi tool output schemas (mirrors cloud's drizzle 0002).
   openApiOutputSchemaDataMigration,
+  // Move inline spec / introspection text out of integration.config into the
+  // blob table (config keeps the content hash). Mirrors cloud's
+  // migrate-specs-to-blobs script.
+  openApiSpecBlobDataMigration,
+  graphqlIntrospectionBlobDataMigration,
 ];

packages/core/sdk/src/index.ts

@@ -334,6 +334,12 @@ export {
   type SqliteDataMigration,
   type SqliteDataMigrationClient,
 } from "./sqlite-data-migrations";
+// Shared inline-config-field → blob-table migration body; the protocol
+// plugins bind their field names and export the ledger entries.
+export {
+  runSqliteConfigBlobMigration,
+  type SqliteConfigBlobMigrationOptions,
+} from "./sqlite-config-blob-migration";
 export {
   authToolFailure,
   type AuthToolFailureCode,

packages/core/sdk/src/sqlite-config-blob-migration.test.ts

@@ -0,0 +1,192 @@
+import { describe, expect, it } from "@effect/vitest";
+import { Effect, Schema } from "effect";
+
+import { collectTables } from "./executor";
+import { createSqliteTestFumaDb } from "./sqlite-test-db";
+import { runSqliteConfigBlobMigration } from "./sqlite-config-blob-migration";
+import { runSqliteDataMigrations, type SqliteDataMigrationClient } from "./sqlite-data-migrations";
+
+const OPENAPI_OPTIONS = {
+  migrationName: "test-spec-to-blob",
+  pluginId: "openapi",
+  inlineField: "spec",
+  hashField: "specHash",
+  blobKeyPrefix: "spec",
+} as const;
+
+const decodeConfigJson = Schema.decodeUnknownSync(
+  Schema.fromJsonString(Schema.Record(Schema.String, Schema.Unknown)),
+);
+
+const insertIntegration = async (
+  client: SqliteDataMigrationClient,
+  row: { rowId: string; tenant: string; slug: string; pluginId: string; config: unknown },
+) => {
+  await client.execute({
+    sql: `INSERT INTO integration (row_id, tenant, slug, plugin_id, description, config, can_remove, can_refresh, created_at, updated_at)
+          VALUES (?, ?, ?, ?, ?, ?, 1, 0, ?, ?)`,
+    args: [
+      row.rowId,
+      row.tenant,
+      row.slug,
+      row.pluginId,
+      row.slug,
+      JSON.stringify(row.config),
+      Date.now(),
+      Date.now(),
+    ],
+  });
+};
+
+describe("runSqliteConfigBlobMigration", () => {
+  it.effect("moves inline spec text to a blob row and rewrites the config pointer", () =>
+    Effect.gen(function* () {
+      const db = yield* Effect.promise(() => createSqliteTestFumaDb({ tables: collectTables() }));
+      const spec = JSON.stringify({ openapi: "3.0.0", info: { title: "T", version: "1" } });
+      yield* Effect.promise(() =>
+        insertIntegration(db.client, {
+          rowId: "r1",
+          tenant: "t1",
+          slug: "legacy_api",
+          pluginId: "openapi",
+          config: { spec, sourceUrl: "https://example.com/spec.json" },
+        }),
+      );
+
+      const moved = yield* runSqliteConfigBlobMigration(db.client, OPENAPI_OPTIONS);
+      expect(moved).toBe(1);
+
+      const integration = yield* Effect.promise(() =>
+        db.client.execute("SELECT config FROM integration WHERE row_id = 'r1'"),
+      );
+      const config = decodeConfigJson(String(integration.rows[0]!.config));
+      expect(config.spec).toBeUndefined();
+      expect(typeof config.specHash).toBe("string");
+      // Untouched fields survive the rewrite.
+      expect(config.sourceUrl).toBe("https://example.com/spec.json");
+
+      // The blob row uses the runtime's exact naming: namespace
+      // `o:<tenant>/<pluginId>`, key `spec/<hash>`, id JSON.stringify pair.
+      const blob = yield* Effect.promise(() =>
+        db.client.execute({
+          sql: "SELECT namespace, key, value FROM blob WHERE id = ?",
+          args: [JSON.stringify([`o:t1/openapi`, `spec/${config.specHash}`])],
+        }),
+      );
+      expect(blob.rows).toHaveLength(1);
+      expect(blob.rows[0]!.value).toBe(spec);
+
+      yield* Effect.promise(() => db.close());
+    }),
+  );
+
+  it.effect("is idempotent and leaves pointer-shaped and foreign rows alone", () =>
+    Effect.gen(function* () {
+      const db = yield* Effect.promise(() => createSqliteTestFumaDb({ tables: collectTables() }));
+      yield* Effect.promise(() =>
+        insertIntegration(db.client, {
+          rowId: "r1",
+          tenant: "t1",
+          slug: "already_migrated",
+          pluginId: "openapi",
+          config: { specHash: "abc123" },
+        }),
+      );
+      yield* Effect.promise(() =>
+        insertIntegration(db.client, {
+          rowId: "r2",
+          tenant: "t1",
+          slug: "mcp_thing",
+          pluginId: "mcp",
+          config: { endpoint: "https://mcp.example.com" },
+        }),
+      );
+      yield* Effect.promise(() =>
+        insertIntegration(db.client, {
+          rowId: "r3",
+          tenant: "t1",
+          slug: "legacy_api",
+          pluginId: "openapi",
+          config: { spec: "{}" },
+        }),
+      );
+
+      expect(yield* runSqliteConfigBlobMigration(db.client, OPENAPI_OPTIONS)).toBe(1);
+      // Second run: everything is pointer-shaped now.
+      expect(yield* runSqliteConfigBlobMigration(db.client, OPENAPI_OPTIONS)).toBe(0);
+
+      const mcpRow = yield* Effect.promise(() =>
+        db.client.execute("SELECT config FROM integration WHERE row_id = 'r2'"),
+      );
+      expect(decodeConfigJson(String(mcpRow.rows[0]!.config))).toEqual({
+        endpoint: "https://mcp.example.com",
+      });
+
+      yield* Effect.promise(() => db.close());
+    }),
+  );
+
+  it.effect("identical specs across integrations share one blob per tenant", () =>
+    Effect.gen(function* () {
+      const db = yield* Effect.promise(() => createSqliteTestFumaDb({ tables: collectTables() }));
+      const spec = JSON.stringify({ openapi: "3.0.0", info: { title: "Shared", version: "1" } });
+      for (const [rowId, slug] of [
+        ["r1", "api_a"],
+        ["r2", "api_b"],
+      ] as const) {
+        yield* Effect.promise(() =>
+          insertIntegration(db.client, {
+            rowId,
+            tenant: "t1",
+            slug,
+            pluginId: "openapi",
+            config: { spec },
+          }),
+        );
+      }
+
+      expect(yield* runSqliteConfigBlobMigration(db.client, OPENAPI_OPTIONS)).toBe(2);
+      const blobs = yield* Effect.promise(() => db.client.execute("SELECT id FROM blob"));
+      expect(blobs.rows).toHaveLength(1);
+
+      yield* Effect.promise(() => db.close());
+    }),
+  );
+
+  it.effect("returns 0 when the integration table does not exist yet", () =>
+    Effect.gen(function* () {
+      // The ledger may run against a brand-new database before any schema
+      // bring-up; an absent table is "nothing to migrate", not an error.
+      const db = yield* Effect.promise(() => createSqliteTestFumaDb({ tables: collectTables() }));
+      yield* Effect.promise(() => db.client.execute("DROP TABLE integration"));
+      expect(yield* runSqliteConfigBlobMigration(db.client, OPENAPI_OPTIONS)).toBe(0);
+      yield* Effect.promise(() => db.close());
+    }),
+  );
+
+  it.effect("runs once under the ledger and is stamped", () =>
+    Effect.gen(function* () {
+      const db = yield* Effect.promise(() => createSqliteTestFumaDb({ tables: collectTables() }));
+      yield* Effect.promise(() =>
+        insertIntegration(db.client, {
+          rowId: "r1",
+          tenant: "t1",
+          slug: "legacy_api",
+          pluginId: "openapi",
+          config: { spec: "{}" },
+        }),
+      );
+      const entry = {
+        name: "test-spec-to-blob",
+        run: (client: SqliteDataMigrationClient) =>
+          runSqliteConfigBlobMigration(client, OPENAPI_OPTIONS).pipe(Effect.asVoid),
+      };
+
+      expect(yield* runSqliteDataMigrations(db.client, [entry])).toEqual(["test-spec-to-blob"]);
+      // Stamped: the second boot doesn't re-run it.
+      expect(yield* runSqliteDataMigrations(db.client, [entry])).toEqual([]);
+
+      yield* Effect.promise(() => db.close());
+    }),
+  );
+});

packages/core/sdk/src/sqlite-config-blob-migration.ts

@@ -0,0 +1,132 @@
+// ---------------------------------------------------------------------------
+// Data migration: move an oversized inline `integration.config` field into
+// the blob table. The shape both protocol plugins need is identical — only
+// the field names differ — so the body lives here and each plugin exports a
+// ledger entry that binds its constants (openapi: spec → specHash under
+// `spec/<hash>`; graphql: introspectionJson → introspectionHash under
+// `introspection/<hash>`).
+//
+// Blob rows are written with the EXACT naming `makeFumaBlobStore` +
+// `pluginBlobStore` read back at runtime: namespace `o:<tenant>/<pluginId>`
+// (the org partition — integration configs are catalog-level), key
+// `<prefix>/<sha256>`, id `JSON.stringify([namespace, key])`. That makes it
+// correct ONLY for hosts whose runtime blob backend is the FumaDB store
+// (local, selfhost) — a host that reads blobs elsewhere (the D1 host reads
+// R2) must not register it, or the rewritten pointers would dangle.
+//
+// Idempotent: pointer-shaped configs (no inline field) plan zero updates,
+// and blob writes are content-addressed upserts.
+// ---------------------------------------------------------------------------
+
+import { Effect, Option, Schema } from "effect";
+
+import { sha256Hex } from "./blob";
+import { DataMigrationError, type SqliteDataMigrationClient } from "./sqlite-data-migrations";
+
+export interface SqliteConfigBlobMigrationOptions {
+  /** The ledger entry name, for error attribution. */
+  readonly migrationName: string;
+  /** Rows whose `plugin_id` equals this are candidates. */
+  readonly pluginId: string;
+  /** The config field holding the inline text to move (e.g. `spec`). */
+  readonly inlineField: string;
+  /** The config field that will carry the content hash (e.g. `specHash`). */
+  readonly hashField: string;
+  /** Blob key prefix; the key is `<prefix>/<sha256>` (e.g. `spec`). */
+  readonly blobKeyPrefix: string;
+}
+
+const decodeJsonOption = Schema.decodeUnknownOption(Schema.UnknownFromJsonString);
+
+const isRecord = (value: unknown): value is Record<string, unknown> =>
+  typeof value === "object" && value !== null && !Array.isArray(value);
+
+/**
+ * Move every inline `options.inlineField` in this plugin's integration
+ * configs into the blob table and rewrite the config to carry
+ * `options.hashField`. Returns the number of rows rewritten. The
+ * `integration` table may not exist yet on a fresh database — that counts
+ * as nothing to migrate.
+ */
+export const runSqliteConfigBlobMigration = (
+  client: SqliteDataMigrationClient,
+  options: SqliteConfigBlobMigrationOptions,
+): Effect.Effect<number, DataMigrationError> =>
+  Effect.gen(function* () {
+    const execute = (stmt: string | { readonly sql: string; readonly args: readonly unknown[] }) =>
+      Effect.tryPromise({
+        try: () => client.execute(stmt),
+        catch: (cause) => new DataMigrationError({ migration: options.migrationName, cause }),
+      });
+
+    const exists = yield* execute(
+      "SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'integration'",
+    );
+    if (exists.rows.length === 0) return 0;
+
+    const result = yield* execute({
+      sql: "SELECT row_id, tenant, config FROM integration WHERE plugin_id = ? AND config IS NOT NULL",
+      args: [options.pluginId],
+    });
+
+    interface PlannedMove {
+      readonly rowId: string;
+      readonly namespace: string;
+      readonly key: string;
+      readonly blobId: string;
+      readonly inlineText: string;
+      readonly nextConfig: string;
+    }
+    const moves: PlannedMove[] = [];
+    for (const row of result.rows) {
+      if (typeof row.row_id !== "string" || typeof row.tenant !== "string") continue;
+      if (typeof row.config !== "string") continue;
+      const decoded = decodeJsonOption(row.config);
+      if (Option.isNone(decoded) || !isRecord(decoded.value)) continue;
+      const inline = decoded.value[options.inlineField];
+      if (typeof inline !== "string") continue;
+
+      const hash = yield* sha256Hex(inline);
+      const namespace = `o:${row.tenant}/${options.pluginId}`;
+      const key = `${options.blobKeyPrefix}/${hash}`;
+      const { [options.inlineField]: _removed, ...rest } = decoded.value;
+      moves.push({
+        rowId: row.row_id,
+        namespace,
+        key,
+        blobId: JSON.stringify([namespace, key]),
+        inlineText: inline,
+        nextConfig: JSON.stringify({ ...rest, [options.hashField]: hash }),
+      });
+    }
+    if (moves.length === 0) return 0;
+
+    const applyAll = Effect.gen(function* () {
+      for (const move of moves) {
+        const existing = yield* execute({
+          sql: "SELECT row_id FROM blob WHERE id = ?",
+          args: [move.blobId],
+        });
+        if (existing.rows.length === 0) {
+          yield* execute({
+            sql: "INSERT INTO blob (namespace, key, value, row_id, id) VALUES (?, ?, ?, ?, ?)",
+            args: [move.namespace, move.key, move.inlineText, crypto.randomUUID(), move.blobId],
+          });
+        } else {
+          yield* execute({
+            sql: "UPDATE blob SET value = ? WHERE id = ?",
+            args: [move.inlineText, move.blobId],
+          });
+        }
+        yield* execute({
+          sql: "UPDATE integration SET config = ? WHERE row_id = ?",
+          args: [move.nextConfig, move.rowId],
+        });
+      }
+      yield* execute("COMMIT");
+    });
+
+    yield* execute("BEGIN");
+    yield* applyAll.pipe(Effect.tapError(() => execute("ROLLBACK").pipe(Effect.ignore)));
+    return moves.length;
+  });

packages/plugins/graphql/src/sdk/index.ts

@@ -37,5 +37,7 @@ export {
 
 export { migrateGraphqlAuthConfig } from "./migrate-config";
 
+export { graphqlIntrospectionBlobDataMigration } from "./introspection-blob-migration";
+
 // Request-shaped authoring: `headers: { Authorization: ["Bearer ", variable("token")] }`.
 export { variable, type ApiKeyAuthTemplate } from "@executor-js/sdk/http-auth";