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