Move inline specs to the blob table at boot on the libSQL hosts (#968)

Cloud's databases were rewritten by the out-of-band migrate-specs-to-blobs script, but the libSQL hosts (local, selfhost) have no operator — their integration.config rows still inline multi-MB OpenAPI spec text and GraphQL introspection snapshots from before the blob seam existed. This adds the ledger counterpart: a shared migration body in the SDK (runSqliteConfigBlobMigration) that moves a named inline config field into the blob table under the runtime's exact naming (o:<tenant>/<plugin> namespace, <prefix>/<sha256> key) and rewrites the config to the content-hash pointer, all inside one transaction. The openapi and graphql plugins bind their field names and export ledger entries, registered in both apps' data-migration registries. Idempotent (pointer rows plan zero work, blob writes are content-addressed upserts), tolerant of a fresh database with no integration table, and stamped once by the ledger. With every host now converging on the pointer shape, the legacy inline-read paths in the plugins become removable.

Rhys Sullivan committed 5 months ago

Commit 8a74ca01 parent 6d0ad939 files · +411 −2Browse files

Unified · Split

9 files

apps/host-selfhost/src/db/data-migrations.ts +10 −1
apps/local/src/db/data-migrations.ts +10 −1
packages/core/sdk/src/index.ts +6 −0
packages/core/sdk/src/sqlite-config-blob-migration.test.ts +192 −0
packages/core/sdk/src/sqlite-config-blob-migration.ts +132 −0
packages/plugins/graphql/src/sdk/index.ts +2 −0
packages/plugins/graphql/src/sdk/introspection-blob-migration.ts +29 −0
packages/plugins/openapi/src/sdk/index.ts +2 −0
packages/plugins/openapi/src/sdk/spec-blob-migration.ts +28 −0

apps/host-selfhost/src/db/data-migrations.ts+10−1

 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";
   // 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+10−1

 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";
   // 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+6−0

   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+192−0

@@ -0,0 +1,192 @@
	1	import { describe, expect, it } from "@effect/vitest";
	2	import { Effect, Schema } from "effect";
	3
	4	import { collectTables } from "./executor";
	5	import { createSqliteTestFumaDb } from "./sqlite-test-db";
	6	import { runSqliteConfigBlobMigration } from "./sqlite-config-blob-migration";
	7	import { runSqliteDataMigrations, type SqliteDataMigrationClient } from "./sqlite-data-migrations";
	8
	9	const OPENAPI_OPTIONS = {
	10	migrationName: "test-spec-to-blob",
	11	pluginId: "openapi",
	12	inlineField: "spec",
	13	hashField: "specHash",
	14	blobKeyPrefix: "spec",
	15	} as const;
	16
	17	const decodeConfigJson = Schema.decodeUnknownSync(
	18	Schema.fromJsonString(Schema.Record(Schema.String, Schema.Unknown)),
	19	);
	20
	21	const insertIntegration = async (
	22	client: SqliteDataMigrationClient,
	23	row: { rowId: string; tenant: string; slug: string; pluginId: string; config: unknown },
	24	) => {
	25	await client.execute({
	26	sql: `INSERT INTO integration (row_id, tenant, slug, plugin_id, description, config, can_remove, can_refresh, created_at, updated_at)
	27	VALUES (?, ?, ?, ?, ?, ?, 1, 0, ?, ?)`,
	28	args: [
	29	row.rowId,
	30	row.tenant,
	31	row.slug,
	32	row.pluginId,
	33	row.slug,
	34	JSON.stringify(row.config),
	35	Date.now(),
	36	Date.now(),
	37	],
	38	});
	39	};
	40
	41	describe("runSqliteConfigBlobMigration", () => {
	42	it.effect("moves inline spec text to a blob row and rewrites the config pointer", () =>
	43	Effect.gen(function* () {
	44	const db = yield* Effect.promise(() => createSqliteTestFumaDb({ tables: collectTables() }));
	45	const spec = JSON.stringify({ openapi: "3.0.0", info: { title: "T", version: "1" } });
	46	yield* Effect.promise(() =>
	47	insertIntegration(db.client, {
	48	rowId: "r1",
	49	tenant: "t1",
	50	slug: "legacy_api",
	51	pluginId: "openapi",
	52	config: { spec, sourceUrl: "https://example.com/spec.json" },
	53	}),
	54	);
	55
	56	const moved = yield* runSqliteConfigBlobMigration(db.client, OPENAPI_OPTIONS);
	57	expect(moved).toBe(1);
	58
	59	const integration = yield* Effect.promise(() =>
	60	db.client.execute("SELECT config FROM integration WHERE row_id = 'r1'"),
	61	);
	62	const config = decodeConfigJson(String(integration.rows[0]!.config));
	63	expect(config.spec).toBeUndefined();
	64	expect(typeof config.specHash).toBe("string");
	65	// Untouched fields survive the rewrite.
	66	expect(config.sourceUrl).toBe("https://example.com/spec.json");
	67
	68	// The blob row uses the runtime's exact naming: namespace
	69	// `o:<tenant>/<pluginId>`, key `spec/<hash>`, id JSON.stringify pair.
	70	const blob = yield* Effect.promise(() =>
	71	db.client.execute({
	72	sql: "SELECT namespace, key, value FROM blob WHERE id = ?",
	73	args: [JSON.stringify([`o:t1/openapi`, `spec/${config.specHash}`])],
	74	}),
	75	);
	76	expect(blob.rows).toHaveLength(1);
	77	expect(blob.rows[0]!.value).toBe(spec);
	78
	79	yield* Effect.promise(() => db.close());
	80	}),
	81	);
	82
	83	it.effect("is idempotent and leaves pointer-shaped and foreign rows alone", () =>
	84	Effect.gen(function* () {
	85	const db = yield* Effect.promise(() => createSqliteTestFumaDb({ tables: collectTables() }));
	86	yield* Effect.promise(() =>
	87	insertIntegration(db.client, {
	88	rowId: "r1",
	89	tenant: "t1",
	90	slug: "already_migrated",
	91	pluginId: "openapi",
	92	config: { specHash: "abc123" },
	93	}),
	94	);
	95	yield* Effect.promise(() =>
	96	insertIntegration(db.client, {
	97	rowId: "r2",
	98	tenant: "t1",
	99	slug: "mcp_thing",
	100	pluginId: "mcp",
	101	config: { endpoint: "https://mcp.example.com" },
	102	}),
	103	);
	104	yield* Effect.promise(() =>
	105	insertIntegration(db.client, {
	106	rowId: "r3",
	107	tenant: "t1",
	108	slug: "legacy_api",
	109	pluginId: "openapi",
	110	config: { spec: "{}" },
	111	}),
	112	);
	113
	114	expect(yield* runSqliteConfigBlobMigration(db.client, OPENAPI_OPTIONS)).toBe(1);
	115	// Second run: everything is pointer-shaped now.
	116	expect(yield* runSqliteConfigBlobMigration(db.client, OPENAPI_OPTIONS)).toBe(0);
	117
	118	const mcpRow = yield* Effect.promise(() =>
	119	db.client.execute("SELECT config FROM integration WHERE row_id = 'r2'"),
	120	);
	121	expect(decodeConfigJson(String(mcpRow.rows[0]!.config))).toEqual({
	122	endpoint: "https://mcp.example.com",
	123	});
	124
	125	yield* Effect.promise(() => db.close());
	126	}),
	127	);
	128
	129	it.effect("identical specs across integrations share one blob per tenant", () =>
	130	Effect.gen(function* () {
	131	const db = yield* Effect.promise(() => createSqliteTestFumaDb({ tables: collectTables() }));
	132	const spec = JSON.stringify({ openapi: "3.0.0", info: { title: "Shared", version: "1" } });
	133	for (const [rowId, slug] of [
	134	["r1", "api_a"],
	135	["r2", "api_b"],
	136	] as const) {
	137	yield* Effect.promise(() =>
	138	insertIntegration(db.client, {
	139	rowId,
	140	tenant: "t1",
	141	slug,
	142	pluginId: "openapi",
	143	config: { spec },
	144	}),
	145	);
	146	}
	147
	148	expect(yield* runSqliteConfigBlobMigration(db.client, OPENAPI_OPTIONS)).toBe(2);
	149	const blobs = yield* Effect.promise(() => db.client.execute("SELECT id FROM blob"));
	150	expect(blobs.rows).toHaveLength(1);
	151
	152	yield* Effect.promise(() => db.close());
	153	}),
	154	);
	155
	156	it.effect("returns 0 when the integration table does not exist yet", () =>
	157	Effect.gen(function* () {
	158	// The ledger may run against a brand-new database before any schema
	159	// bring-up; an absent table is "nothing to migrate", not an error.
	160	const db = yield* Effect.promise(() => createSqliteTestFumaDb({ tables: collectTables() }));
	161	yield* Effect.promise(() => db.client.execute("DROP TABLE integration"));
	162	expect(yield* runSqliteConfigBlobMigration(db.client, OPENAPI_OPTIONS)).toBe(0);
	163	yield* Effect.promise(() => db.close());
	164	}),
	165	);
	166
	167	it.effect("runs once under the ledger and is stamped", () =>
	168	Effect.gen(function* () {
	169	const db = yield* Effect.promise(() => createSqliteTestFumaDb({ tables: collectTables() }));
	170	yield* Effect.promise(() =>
	171	insertIntegration(db.client, {
	172	rowId: "r1",
	173	tenant: "t1",
	174	slug: "legacy_api",
	175	pluginId: "openapi",
	176	config: { spec: "{}" },
	177	}),
	178	);
	179	const entry = {
	180	name: "test-spec-to-blob",
	181	run: (client: SqliteDataMigrationClient) =>
	182	runSqliteConfigBlobMigration(client, OPENAPI_OPTIONS).pipe(Effect.asVoid),
	183	};
	184
	185	expect(yield* runSqliteDataMigrations(db.client, [entry])).toEqual(["test-spec-to-blob"]);
	186	// Stamped: the second boot doesn't re-run it.
	187	expect(yield* runSqliteDataMigrations(db.client, [entry])).toEqual([]);
	188
	189	yield* Effect.promise(() => db.close());
	190	}),
	191	);
	192	});

packages/core/sdk/src/sqlite-config-blob-migration.ts+132−0

@@ -0,0 +1,132 @@
	1	// ---------------------------------------------------------------------------
	2	// Data migration: move an oversized inline `integration.config` field into
	3	// the blob table. The shape both protocol plugins need is identical — only
	4	// the field names differ — so the body lives here and each plugin exports a
	5	// ledger entry that binds its constants (openapi: spec → specHash under
	6	// `spec/<hash>`; graphql: introspectionJson → introspectionHash under
	7	// `introspection/<hash>`).
	8	//
	9	// Blob rows are written with the EXACT naming `makeFumaBlobStore` +
	10	// `pluginBlobStore` read back at runtime: namespace `o:<tenant>/<pluginId>`
	11	// (the org partition — integration configs are catalog-level), key
	12	// `<prefix>/<sha256>`, id `JSON.stringify([namespace, key])`. That makes it
	13	// correct ONLY for hosts whose runtime blob backend is the FumaDB store
	14	// (local, selfhost) — a host that reads blobs elsewhere (the D1 host reads
	15	// R2) must not register it, or the rewritten pointers would dangle.
	16	//
	17	// Idempotent: pointer-shaped configs (no inline field) plan zero updates,
	18	// and blob writes are content-addressed upserts.
	19	// ---------------------------------------------------------------------------
	20
	21	import { Effect, Option, Schema } from "effect";
	22
	23	import { sha256Hex } from "./blob";
	24	import { DataMigrationError, type SqliteDataMigrationClient } from "./sqlite-data-migrations";
	25
	26	export interface SqliteConfigBlobMigrationOptions {
	27	/** The ledger entry name, for error attribution. */
	28	readonly migrationName: string;
	29	/** Rows whose `plugin_id` equals this are candidates. */
	30	readonly pluginId: string;
	31	/** The config field holding the inline text to move (e.g. `spec`). */
	32	readonly inlineField: string;
	33	/** The config field that will carry the content hash (e.g. `specHash`). */
	34	readonly hashField: string;
	35	/** Blob key prefix; the key is `<prefix>/<sha256>` (e.g. `spec`). */
	36	readonly blobKeyPrefix: string;
	37	}
	38
	39	const decodeJsonOption = Schema.decodeUnknownOption(Schema.UnknownFromJsonString);
	40
	41	const isRecord = (value: unknown): value is Record<string, unknown> =>
	42	typeof value === "object" && value !== null && !Array.isArray(value);
	43
	44	/**
	45	* Move every inline `options.inlineField` in this plugin's integration
	46	* configs into the blob table and rewrite the config to carry
	47	* `options.hashField`. Returns the number of rows rewritten. The
	48	* `integration` table may not exist yet on a fresh database — that counts
	49	* as nothing to migrate.
	50	*/
	51	export const runSqliteConfigBlobMigration = (
	52	client: SqliteDataMigrationClient,
	53	options: SqliteConfigBlobMigrationOptions,
	54	): Effect.Effect<number, DataMigrationError> =>
	55	Effect.gen(function* () {
	56	const execute = (stmt: string \| { readonly sql: string; readonly args: readonly unknown[] }) =>
	57	Effect.tryPromise({
	58	try: () => client.execute(stmt),
	59	catch: (cause) => new DataMigrationError({ migration: options.migrationName, cause }),
	60	});
	61
	62	const exists = yield* execute(
	63	"SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'integration'",
	64	);
	65	if (exists.rows.length === 0) return 0;
	66
	67	const result = yield* execute({
	68	sql: "SELECT row_id, tenant, config FROM integration WHERE plugin_id = ? AND config IS NOT NULL",
	69	args: [options.pluginId],
	70	});
	71
	72	interface PlannedMove {
	73	readonly rowId: string;
	74	readonly namespace: string;
	75	readonly key: string;
	76	readonly blobId: string;
	77	readonly inlineText: string;
	78	readonly nextConfig: string;
	79	}
	80	const moves: PlannedMove[] = [];
	81	for (const row of result.rows) {
	82	if (typeof row.row_id !== "string" \|\| typeof row.tenant !== "string") continue;
	83	if (typeof row.config !== "string") continue;
	84	const decoded = decodeJsonOption(row.config);
	85	if (Option.isNone(decoded) \|\| !isRecord(decoded.value)) continue;
	86	const inline = decoded.value[options.inlineField];
	87	if (typeof inline !== "string") continue;
	88
	89	const hash = yield* sha256Hex(inline);
	90	const namespace = `o:${row.tenant}/${options.pluginId}`;
	91	const key = `${options.blobKeyPrefix}/${hash}`;
	92	const { [options.inlineField]: _removed, ...rest } = decoded.value;
	93	moves.push({
	94	rowId: row.row_id,
	95	namespace,
	96	key,
	97	blobId: JSON.stringify([namespace, key]),
	98	inlineText: inline,
	99	nextConfig: JSON.stringify({ ...rest, [options.hashField]: hash }),
	100	});
	101	}
	102	if (moves.length === 0) return 0;
	103
	104	const applyAll = Effect.gen(function* () {
	105	for (const move of moves) {
	106	const existing = yield* execute({
	107	sql: "SELECT row_id FROM blob WHERE id = ?",
	108	args: [move.blobId],
	109	});
	110	if (existing.rows.length === 0) {
	111	yield* execute({
	112	sql: "INSERT INTO blob (namespace, key, value, row_id, id) VALUES (?, ?, ?, ?, ?)",
	113	args: [move.namespace, move.key, move.inlineText, crypto.randomUUID(), move.blobId],
	114	});
	115	} else {
	116	yield* execute({
	117	sql: "UPDATE blob SET value = ? WHERE id = ?",
	118	args: [move.inlineText, move.blobId],
	119	});
	120	}
	121	yield* execute({
	122	sql: "UPDATE integration SET config = ? WHERE row_id = ?",
	123	args: [move.nextConfig, move.rowId],
	124	});
	125	}
	126	yield* execute("COMMIT");
	127	});
	128
	129	yield* execute("BEGIN");
	130	yield* applyAll.pipe(Effect.tapError(() => execute("ROLLBACK").pipe(Effect.ignore)));
	131	return moves.length;
	132	});

packages/plugins/graphql/src/sdk/index.ts+2−0

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

packages/plugins/graphql/src/sdk/introspection-blob-migration.ts+29−0

@@ -0,0 +1,29 @@
	1	// ---------------------------------------------------------------------------
	2	// Data migration: move inline GraphQL introspection snapshots out of
	3	// `integration.config` into the blob table (`introspection/<sha256>`, config
	4	// keeps `introspectionHash`) — the libSQL-ledger counterpart of cloud's
	5	// out-of-band migrate-specs-to-blobs script. Runs once per database through
	6	// the data-migration ledger; the shared body lives in @executor-js/sdk
	7	// (`runSqliteConfigBlobMigration`).
	8	// ---------------------------------------------------------------------------
	9
	10	import { Effect } from "effect";
	11	import {
	12	runSqliteConfigBlobMigration,
	13	type SqliteDataMigrationClient,
	14	} from "@executor-js/sdk/core";
	15
	16	const MIGRATION_NAME = "2026-06-12-graphql-introspection-to-blob";
	17
	18	/** Registry entry for the boot-time data-migration ledger. */
	19	export const graphqlIntrospectionBlobDataMigration = {
	20	name: MIGRATION_NAME,
	21	run: (client: SqliteDataMigrationClient) =>
	22	runSqliteConfigBlobMigration(client, {
	23	migrationName: MIGRATION_NAME,
	24	pluginId: "graphql",
	25	inlineField: "introspectionJson",
	26	hashField: "introspectionHash",
	27	blobKeyPrefix: "introspection",
	28	}).pipe(Effect.asVoid),
	29	};

packages/plugins/openapi/src/sdk/index.ts+2−0

@@ -83,3 +83,5 @@ export { · 82 unmodified lines
83	runSqliteOpenApiOutputSchemaMigration,	83	runSqliteOpenApiOutputSchemaMigration,
84	unwrapOpenApiTransportEnvelope,	84	unwrapOpenApiTransportEnvelope,
85	} from "./output-schema-migration";	85	} from "./output-schema-migration";
		86
		87	export { openApiSpecBlobDataMigration } from "./spec-blob-migration";

packages/plugins/openapi/src/sdk/spec-blob-migration.ts+28−0

@@ -0,0 +1,28 @@
	1	// ---------------------------------------------------------------------------
	2	// Data migration: move inline OpenAPI spec text out of `integration.config`
	3	// into the blob table (`spec/<sha256>`, config keeps `specHash`) — the
	4	// libSQL-ledger counterpart of cloud's out-of-band migrate-specs-to-blobs
	5	// script. Runs once per database through the data-migration ledger; the
	6	// shared body lives in @executor-js/sdk (`runSqliteConfigBlobMigration`).
	7	// ---------------------------------------------------------------------------
	8
	9	import { Effect } from "effect";
	10	import {
	11	runSqliteConfigBlobMigration,
	12	type SqliteDataMigrationClient,
	13	} from "@executor-js/sdk/core";
	14
	15	const MIGRATION_NAME = "2026-06-12-openapi-spec-to-blob";
	16
	17	/** Registry entry for the boot-time data-migration ledger. */
	18	export const openApiSpecBlobDataMigration = {
	19	name: MIGRATION_NAME,
	20	run: (client: SqliteDataMigrationClient) =>
	21	runSqliteConfigBlobMigration(client, {
	22	migrationName: MIGRATION_NAME,
	23	pluginId: "openapi",
	24	inlineField: "spec",
	25	hashField: "specHash",
	26	blobKeyPrefix: "spec",
	27	}).pipe(Effect.asVoid),
	28	};