UsefulSoftwareCo · jadch · Jul 23, 2026
diff --git a/apps/host-cloudflare/src/db/d1.test.ts b/apps/host-cloudflare/src/db/d1.test.ts
@@ -0,0 +1,207 @@
+import { describe, expect, it } from "@effect/vitest";
+import type { D1Database } from "@cloudflare/workers-types";
+
+import { collectTables, type SqliteDataMigrationClient } from "@executor-js/sdk";
+import { createSqliteTestFumaDb } from "@executor-js/sdk/testing";
+
+import { createD1ExecutorDb } from "./d1";
+
+const makeRecordingD1 = (
+  client: SqliteDataMigrationClient,
+): {
+  readonly db: D1Database;
+  readonly statements: string[];
+  readonly failWhen: (predicate: ((sql: string) => boolean) | null) => void;
+} => {
+  const statements: string[] = [];
+  let failurePredicate: ((sql: string) => boolean) | null = null;
+  const record = (sql: string): void => {
+    statements.push(sql);
+    if (failurePredicate?.(sql)) {
+      // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- test boundary: the fake D1 adapter must reject exactly where the real D1 query would reject
+      throw new Error("forced D1 failure");
+    }
+  };
+  const prepare = (sql: string) => {
+    const statement = (args: readonly unknown[]): Record<string, unknown> => ({
+      bind: (...values: readonly unknown[]) => statement([...args, ...values]),
+      all: async () => {
+        record(sql);
+        const result = await client.execute({ sql, args });
+        return { success: true, meta: {}, results: result.rows };
+      },
+      run: async () => {
+        record(sql);
+        await client.execute({ sql, args });
+        return { success: true, meta: {}, results: [] };
+      },
+    });
+    return statement([]);
+  };
+
+  // oxlint-disable-next-line executor/no-double-cast -- test double: only the D1 methods used by schema preparation and migrations are implemented
+  const db = {
+    prepare,
+    withSession: () => ({ prepare }),
+  } as unknown as D1Database;
+  return {
+    db,
+    statements,
+    failWhen: (predicate) => {
+      failurePredicate = predicate;
+    },
+  };
+};
+
+const isRuntimeSchemaStatement = (sql: string): boolean => {
+  const normalized = sql.trim().toUpperCase();
+  if (normalized.startsWith("CREATE UNIQUE INDEX IF NOT EXISTS")) return true;
+  if (normalized.startsWith("ALTER TABLE") && normalized.includes(" ADD COLUMN ")) return true;
+  if (!normalized.startsWith("CREATE TABLE IF NOT EXISTS")) return false;
+  return !normalized.includes("DATA_MIGRATION");
+};
+
+describe("createD1ExecutorDb", () => {
+  it("does not repeat runtime schema DDL after the current schema was prepared", async () => {
+    const sqlite = await createSqliteTestFumaDb({ tables: collectTables() });
+    const { db, statements } = makeRecordingD1(sqlite.client);
+
+    const first = await createD1ExecutorDb(db, undefined);
+    await first.close();
+    const firstSchemaStatements = statements.filter(isRuntimeSchemaStatement);
+    expect(firstSchemaStatements.length).toBeGreaterThan(0);
+
+    const secondStart = statements.length;
+    const second = await createD1ExecutorDb(db, undefined);
+    await second.close();
+    const secondStatements = statements.slice(secondStart);
+
+    expect(secondStatements.filter(isRuntimeSchemaStatement)).toEqual([]);
+    expect(secondStatements.some((sql) => sql.includes("SELECT name FROM data_migration"))).toBe(
+      true,
+    );
+
+    await sqlite.close();
+  });
+
+  it("runs the schema ensure again when the generated fingerprint is stale", async () => {
+    const sqlite = await createSqliteTestFumaDb({ tables: collectTables() });
+    const { db, statements } = makeRecordingD1(sqlite.client);
+
+    const first = await createD1ExecutorDb(db, undefined);
+    await first.close();
+    await sqlite.client.execute(
+      `UPDATE private_executor_cloudflare_schema_fingerprint
+       SET fingerprint = 'stale'
+       WHERE id = 'runtime-schema'`,
+    );
+
+    const secondStart = statements.length;
+    const second = await createD1ExecutorDb(db, undefined);
+    await second.close();
+    const secondStatements = statements.slice(secondStart);
+
+    expect(
+      secondStatements.some((sql) => sql.startsWith('CREATE TABLE IF NOT EXISTS "integration"')),
+    ).toBe(true);
+
+    await sqlite.close();
+  });
+
+  it("converges safely when concurrent opens see the same stale fingerprint", async () => {
+    const sqlite = await createSqliteTestFumaDb({ tables: collectTables() });
+    const { db } = makeRecordingD1(sqlite.client);
+    const prepared = await createD1ExecutorDb(db, undefined);
+    await prepared.close();
+    await sqlite.client.execute(
+      `UPDATE private_executor_cloudflare_schema_fingerprint
+       SET fingerprint = 'stale'
+       WHERE id = 'runtime-schema'`,
+    );
+
+    const handles = await Promise.all([
+      createD1ExecutorDb(db, undefined),
+      createD1ExecutorDb(db, undefined),
+    ]);
+    await Promise.all(handles.map((handle) => handle.close()));
+
+    const fingerprints = await sqlite.client.execute(
+      `SELECT fingerprint
+       FROM private_executor_cloudflare_schema_fingerprint
+       WHERE id = 'runtime-schema'`,
+    );
+    expect(fingerprints.rows).toHaveLength(1);
+    expect(fingerprints.rows[0]?.fingerprint).not.toBe("stale");
+
+    await sqlite.close();
+  });
+
+  it("stamps only after compatibility migrations leave the final schema current", async () => {
+    const sqlite = await createSqliteTestFumaDb({ tables: collectTables() });
+    await sqlite.client.execute("DROP TABLE connection");
+    await sqlite.client.execute(`
+      CREATE TABLE connection (
+        integration text NOT NULL,
+        name text NOT NULL,
+        template text NOT NULL,
+        provider text NOT NULL,
+        item_id text NOT NULL,
+        identity_label text,
+        description text,
+        tools_synced_at integer,
+        oauth_client text,
+        oauth_client_owner text,
+        refresh_item_id text,
+        expires_at integer,
+        oauth_scope text,
+        oauth_token_url text,
+        provider_state text,
+        created_at integer NOT NULL,
+        updated_at integer NOT NULL,
+        row_id text PRIMARY KEY NOT NULL,
+        tenant text NOT NULL,
+        owner text NOT NULL,
+        subject text NOT NULL
+      )
+    `);
+    const { db } = makeRecordingD1(sqlite.client);
+
+    const handle = await createD1ExecutorDb(db, undefined);
+    await handle.close();
+
+    const columns = await sqlite.client.execute("PRAGMA table_info('connection')");
+    const columnNames = columns.rows.map((row) => row.name);
+    expect(columnNames).toContain("item_ids");
+    expect(columnNames).toContain("last_health");
+    expect(columnNames).not.toContain("item_id");
+    const indexes = await sqlite.client.execute(
+      `SELECT name FROM sqlite_master
+       WHERE type = 'index'
+         AND name = 'connection_uidx'`,
+    );
+    expect(indexes.rows).toEqual([{ name: "connection_uidx" }]);
+
+    await sqlite.close();
+  });
+
+  it("does not stamp a fingerprint when schema preparation fails", async () => {
+    const sqlite = await createSqliteTestFumaDb({ tables: collectTables() });
+    const { db, failWhen } = makeRecordingD1(sqlite.client);
+    failWhen((sql) => sql.startsWith('CREATE TABLE IF NOT EXISTS "integration"'));
+
+    await expect(createD1ExecutorDb(db, undefined)).rejects.toBeDefined();
+
+    const marker = await sqlite.client.execute(
+      `SELECT name FROM sqlite_master
+       WHERE type = 'table'
+         AND name = 'private_executor_cloudflare_schema_fingerprint'`,
+    );
+    expect(marker.rows).toEqual([]);
+
+    failWhen(null);
+    const recovered = await createD1ExecutorDb(db, undefined);
+    await recovered.close();
+
+    await sqlite.close();
+  });
+});
diff --git a/apps/host-cloudflare/src/db/d1.ts b/apps/host-cloudflare/src/db/d1.ts
@@ -1,6 +1,7 @@
 import { drizzle } from "drizzle-orm/d1";
 import {
   createDrizzleRuntimeSchemaFromTables,
+  createDrizzleRuntimeSchemaSqlFromTables,
   ensureDrizzleRuntimeSchemaFromTables,
 } from "@executor-js/fumadb/adapters/drizzle";
 import type { D1Database, R2Bucket } from "@cloudflare/workers-types";
@@ -13,7 +14,64 @@ import {
 import { makeR2BlobStore } from "@executor-js/cloudflare/blob-store";
 
 import { CLOUDFLARE_NAMESPACE, CLOUDFLARE_SCHEMA_VERSION } from "../config";
-import { runCloudflareDataMigrations } from "./data-migrations";
+import { prepareCloudflareD1Data } from "./data-migrations";
+
+const SCHEMA_FINGERPRINT_TABLE = `private_${CLOUDFLARE_NAMESPACE}_schema_fingerprint`;
+const SCHEMA_FINGERPRINT_ID = "runtime-schema";
+
+const schemaFingerprint = async (statements: readonly string[]): Promise<string> => {
+  const bytes = new TextEncoder().encode(statements.join("\0"));
+  const digest = await crypto.subtle.digest("SHA-256", bytes);
+  return Array.from(new Uint8Array(digest), (byte) => byte.toString(16).padStart(2, "0")).join("");
+};
+
+const isMissingFingerprintTable = (error: unknown): boolean => {
+  for (let current = error, depth = 0; current != null && depth < 8; depth += 1) {
+    const record =
+      typeof current === "object" ? (current as { message?: unknown; cause?: unknown }) : null;
+    const message = typeof record?.message === "string" ? record.message : String(current);
+    if (/no such table/i.test(message) && message.includes(SCHEMA_FINGERPRINT_TABLE)) return true;
+    current = record?.cause ?? null;
+  }
+  return false;
+};
+
+const readPreparedSchemaFingerprint = async (db: D1Database): Promise<string | null> => {
+  const session = db.withSession("first-primary");
+  // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: a missing marker table is the expected first-boot signal; every other D1 failure must still reject startup
+  try {
+    const result = await session
+      .prepare(`SELECT fingerprint FROM "${SCHEMA_FINGERPRINT_TABLE}" WHERE id = ?`)
+      .bind(SCHEMA_FINGERPRINT_ID)
+      .all<{ readonly fingerprint?: unknown }>();
+    const value = result.results[0]?.fingerprint;
+    return typeof value === "string" ? value : null;
+  } catch (error) {
+    if (isMissingFingerprintTable(error)) return null;
+    // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: preserve the original D1 adapter rejection for every failure except the expected absent first-boot marker
+    throw error;
+  }
+};
+
+const storePreparedSchemaFingerprint = async (
+  db: D1Database,
+  fingerprint: string,
+): Promise<void> => {
+  const session = db.withSession("first-primary");
+  await session
+    .prepare(
+      `CREATE TABLE IF NOT EXISTS "${SCHEMA_FINGERPRINT_TABLE}" (id text PRIMARY KEY NOT NULL, fingerprint text NOT NULL)`,
+    )
+    .run();
+  await session
+    .prepare(
+      `INSERT INTO "${SCHEMA_FINGERPRINT_TABLE}" (id, fingerprint)
+       VALUES (?, ?)
+       ON CONFLICT(id) DO UPDATE SET fingerprint = excluded.fingerprint`,
+    )
+    .bind(SCHEMA_FINGERPRINT_ID, fingerprint)
+    .run();
+};
 
 // ---------------------------------------------------------------------------
 // D1 DbProvider handle — the CF-native swap for self-host's libSQL handle.
@@ -42,10 +100,31 @@ export const createD1ExecutorDb = async (
 
   // D1 rejects SQL `BEGIN TRANSACTION` / `SAVEPOINT` (it requires the JS batch
   // API), and the shared ensure wraps its DDL in a transaction when the handle
-  // exposes one. The bring-up is idempotent `CREATE TABLE IF NOT EXISTS`, so run
-  // it WITHOUT a transaction by handing the ensure a run-only view of the handle.
-  await ensureDrizzleRuntimeSchemaFromTables({ run: (query) => drizzleDb.run(query) }, options);
-  await runCloudflareDataMigrations(db, blobs);
+  // exposes one. A generated fingerprint lets every Worker/DO isolate prove the
+  // expected generated schema was already prepared with one primary read instead of
+  // replaying dozens of idempotent CREATE/ALTER statements on every database
+  // open. The marker is stored only after the ensure, data migrations, and any
+  // required post-compatibility ensure succeed. Data migrations retain their
+  // own ledger and still run below on every open.
+  const expectedFingerprint = await schemaFingerprint(
+    createDrizzleRuntimeSchemaSqlFromTables(options),
+  );
+  const preparedFingerprint = await readPreparedSchemaFingerprint(db);
+  const requiresSchemaPreparation = preparedFingerprint !== expectedFingerprint;
+  if (requiresSchemaPreparation) {
+    await ensureDrizzleRuntimeSchemaFromTables({ run: (query) => drizzleDb.run(query) }, options);
+  }
+  const migrationResult = await prepareCloudflareD1Data(db, blobs);
+  if (requiresSchemaPreparation) {
+    // Compatibility migrations can rebuild a legacy table. Re-run the
+    // idempotent ensure before stamping so the database matches the generated
+    // schema after migration, including newer nullable columns the legacy
+    // rebuild did not know about.
+    if (migrationResult.schemaChanged) {
+      await ensureDrizzleRuntimeSchemaFromTables({ run: (query) => drizzleDb.run(query) }, options);
+    }
+    await storePreparedSchemaFingerprint(db, expectedFingerprint);
+  }
 
   // `interactiveTransactions: false` — D1 rejects interactive transactions, so
   // the fuma adapter runs transaction callbacks directly (auto-commit per

diff --git a/apps/host-cloudflare/src/db/data-migrations.ts b/apps/host-cloudflare/src/db/data-migrations.ts
@@ -108,7 +108,8 @@ const rebuildLegacyConnectionTable = async (db: D1Database): Promise<void> => {
   }
 };
 
-export const ensureCloudflareD1SchemaCompatibility = async (db: D1Database): Promise<void> => {
+export const ensureCloudflareD1SchemaCompatibility = async (db: D1Database): Promise<boolean> => {
+  let schemaChanged = false;
   const integrationColumns = await tableColumns(db, "integration");
   if (integrationColumns.has("config")) {
     await db
@@ -122,14 +123,17 @@ export const ensureCloudflareD1SchemaCompatibility = async (db: D1Database): Pro
   }
 
   const connectionColumns = await tableColumns(db, "connection");
-  if (connectionColumns.size === 0) return;
+  if (connectionColumns.size === 0) return schemaChanged;
   if (!connectionColumns.has("item_ids")) {
     await db.prepare(`ALTER TABLE connection ADD COLUMN item_ids json NOT NULL DEFAULT '{}'`).run();
+    schemaChanged = true;
   }
   const updatedConnectionColumns = await tableColumns(db, "connection");
   if (updatedConnectionColumns.has("item_id")) {
     await rebuildLegacyConnectionTable(db);
+    schemaChanged = true;
   }
+  return schemaChanged;
 };
 
 const r2ObjectName = (tenant: string, pluginId: string, key: string): string =>
@@ -248,14 +252,24 @@ const cloudflareDataMigrations = (bucket: R2Bucket | undefined): readonly Sqlite
   openApiNdjsonOutputDataMigration,
 ];
 
-export const runCloudflareDataMigrations = (
+export const prepareCloudflareD1Data = (
   db: D1Database,
   bucket: R2Bucket | undefined,
-): Promise<readonly string[]> =>
+): Promise<{
+  readonly appliedMigrations: readonly string[];
+  readonly schemaChanged: boolean;
+}> =>
   Effect.runPromise(
     Effect.promise(() => ensureCloudflareD1SchemaCompatibility(db)).pipe(
-      Effect.flatMap(() =>
-        runSqliteDataMigrations(d1DataMigrationClient(db), cloudflareDataMigrations(bucket)),
+      Effect.flatMap((schemaChanged) =>
+        runSqliteDataMigrations(d1DataMigrationClient(db), cloudflareDataMigrations(bucket)).pipe(
+          Effect.map((appliedMigrations) => ({ appliedMigrations, schemaChanged })),
+        ),
       ),
     ),
   );
+
+export const runCloudflareDataMigrations = async (
+  db: D1Database,
+  bucket: R2Bucket | undefined,
+): Promise<readonly string[]> => (await prepareCloudflareD1Data(db, bucket)).appliedMigrations;