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