diff --git a/.changeset/add-cloudflare-d1-migrations.md b/.changeset/add-cloudflare-d1-migrations.md new file mode 100644 index 000000000..8cb5df782 --- /dev/null +++ b/.changeset/add-cloudflare-d1-migrations.md @@ -0,0 +1,5 @@ +--- +"@emdash-cms/cloudflare": minor +--- + +Adds deployment-managed D1 migrations through Cloudflare's authenticated REST API. diff --git a/.changeset/add-hyperdrive-direct-migrations.md b/.changeset/add-hyperdrive-direct-migrations.md new file mode 100644 index 000000000..dffdca05a --- /dev/null +++ b/.changeset/add-hyperdrive-direct-migrations.md @@ -0,0 +1,5 @@ +--- +"@emdash-cms/cloudflare": minor +--- + +Adds deployment-managed migrations for PostgreSQL databases reached directly alongside a Hyperdrive deployment. diff --git a/packages/cloudflare/package.json b/packages/cloudflare/package.json index 99047dbdf..fa7124d94 100644 --- a/packages/cloudflare/package.json +++ b/packages/cloudflare/package.json @@ -17,10 +17,18 @@ "types": "./dist/db/d1.d.mts", "default": "./dist/db/d1.mjs" }, + "./db/d1-migrations": { + "types": "./dist/db/d1-migrations.d.mts", + "default": "./dist/db/d1-migrations.mjs" + }, "./db/hyperdrive": { "types": "./dist/db/hyperdrive.d.mts", "default": "./dist/db/hyperdrive.mjs" }, + "./db/hyperdrive-migrations": { + "types": "./dist/db/hyperdrive-migrations.d.mts", + "default": "./dist/db/hyperdrive-migrations.mjs" + }, "./db/do": { "types": "./dist/db/do.d.mts", "default": "./dist/db/do.mjs" diff --git a/packages/cloudflare/src/db/d1-migration-target.ts b/packages/cloudflare/src/db/d1-migration-target.ts new file mode 100644 index 000000000..810069497 --- /dev/null +++ b/packages/cloudflare/src/db/d1-migration-target.ts @@ -0,0 +1,398 @@ +import { createRequire } from "node:module"; +import { isAbsolute, join, resolve } from "node:path"; +import { pathToFileURL } from "node:url"; + +import type { MigrationExecutorFactoryContext, MigrationTarget } from "emdash/migrations"; + +import { readBoundedJson } from "./d1-rest-dialect.js"; + +const ACCOUNT_ID_PATTERN = /^[a-f0-9]{32}$/i; +const DATABASE_ID_PATTERN = /^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}$/i; +const DATABASE_NAME_PATTERN = /^[A-Za-z0-9][A-Za-z0-9_-]*$/; +const BINDING_PATTERN = /^[A-Za-z_$][A-Za-z0-9_$]*$/; +const NIL_DATABASE_ID = "00000000-0000-0000-0000-000000000000"; +const MAX_DATABASE_ID = "ffffffff-ffff-ffff-ffff-ffffffffffff"; +const CONTROL_PLANE_TIMEOUT_MS = 15_000; +const CONTROL_PLANE_MAX_BYTES = 1_048_576; +const DATABASE_LIST_PAGE_SIZE = 100; +const MAX_DATABASE_LIST_PAGES = 1000; + +export interface D1MigrationManifestConfig { + binding: string; +} + +export interface WranglerD1Database { + binding: string; + databaseName?: string; + databaseId?: string; + previewDatabaseId?: string; +} + +export interface WranglerMigrationConfig { + accountId?: string; + d1Databases: WranglerD1Database[]; +} + +export interface ResolvedD1MigrationTarget { + target: MigrationTarget; + accountId: string; + databaseId: string; + databaseName: string; +} + +export interface D1TargetResolutionDependencies { + fetch?: typeof globalThis.fetch; + readWranglerConfig?: typeof loadProjectWranglerConfig; +} + +interface WranglerModule { + unstable_readConfig?: ( + args: { config: string; env?: string }, + options: { hideWarnings: boolean }, + ) => unknown; +} + +interface MetadataEnvelope { + result: unknown; + resultInfo?: unknown; +} + +interface ListResultInfo { + page: number; + perPage: number; + count: number; + totalCount: number; + totalPages: number; +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function canonicalAccountId(value: unknown): string { + if (typeof value !== "string" || !ACCOUNT_ID_PATTERN.test(value)) { + throw new Error("A valid Cloudflare account ID is required for D1 migrations."); + } + const id = value.toLowerCase(); + if (id === "0".repeat(32) || id === "f".repeat(32)) { + throw new Error("A valid Cloudflare account ID is required for D1 migrations."); + } + return id; +} + +function canonicalDatabaseId(value: unknown): string { + if (typeof value !== "string" || !DATABASE_ID_PATTERN.test(value)) { + throw new Error("A valid production D1 database UUID is required."); + } + const id = value.toLowerCase(); + if (id === NIL_DATABASE_ID || id === MAX_DATABASE_ID) { + throw new Error("A valid production D1 database UUID is required."); + } + return id; +} + +function databaseName(value: unknown): string { + if (typeof value !== "string" || !DATABASE_NAME_PATTERN.test(value)) { + throw new Error("A valid D1 database name is required."); + } + return value; +} + +function apiMessages(value: unknown): string[] { + if (!Array.isArray(value)) throw new Error("Cloudflare D1 metadata response is invalid."); + return value.map((item) => { + if (!isRecord(item) || typeof item.message !== "string") { + throw new Error("Cloudflare D1 metadata response is invalid."); + } + return item.message; + }); +} + +function validateApiEnvelope(value: unknown): MetadataEnvelope { + if (!isRecord(value)) throw new Error("Cloudflare D1 metadata response is invalid."); + const errors = apiMessages(value.errors); + apiMessages(value.messages); + if (value.success !== true) { + throw new Error( + errors.length > 0 + ? `Cloudflare D1 metadata request failed: ${errors.join("; ")}` + : "Cloudflare D1 metadata request failed.", + ); + } + if (errors.length !== 0 || !("result" in value)) { + throw new Error("Cloudflare D1 metadata response is invalid."); + } + return { result: value.result, resultInfo: value.result_info }; +} + +async function metadataRequest( + path: string, + token: string, + fetch: typeof globalThis.fetch, +): Promise { + const controller = new AbortController(); + const timer = setTimeout(controller.abort.bind(controller), CONTROL_PLANE_TIMEOUT_MS); + try { + const response = await fetch(`https://api.cloudflare.com/client/v4${path}`, { + headers: { authorization: `Bearer ${token}` }, + redirect: "error", + signal: controller.signal, + }); + if (!response.ok) { + throw new Error(`Cloudflare D1 metadata request failed with HTTP status ${response.status}.`); + } + return validateApiEnvelope(await readBoundedJson(response, CONTROL_PLANE_MAX_BYTES)); + } catch (error) { + if (error instanceof Error && !error.message.includes(token)) throw error; + // eslint-disable-next-line preserve-caught-error -- the cause may contain the authorization token or request details + throw new Error("Cloudflare D1 metadata request failed."); + } finally { + clearTimeout(timer); + } +} + +function metadataDatabase(value: unknown): { uuid: string; name: string } { + if (!isRecord(value)) throw new Error("Cloudflare D1 database metadata is invalid."); + if (value.version !== "production") { + throw new Error("Preview D1 databases cannot be used for deployment migrations."); + } + return { + uuid: canonicalDatabaseId(value.uuid), + name: databaseName(value.name), + }; +} + +async function databaseById( + accountId: string, + databaseId: string, + token: string, + fetch: typeof globalThis.fetch, +): Promise<{ uuid: string; name: string }> { + const metadata = metadataDatabase( + ( + await metadataRequest( + `/accounts/${encodeURIComponent(accountId)}/d1/database/${encodeURIComponent(databaseId)}`, + token, + fetch, + ) + ).result, + ); + if (metadata.uuid !== databaseId) + throw new Error("Cloudflare returned a different D1 database UUID."); + return metadata; +} + +async function databaseByName( + accountId: string, + name: string, + token: string, + fetch: typeof globalThis.fetch, +): Promise<{ uuid: string; name: string }> { + const matches: Array<{ uuid: string; name: string }> = []; + let page = 1; + let totalPages = 1; + let totalCount: number | undefined; + let seenCount = 0; + while (page <= totalPages) { + const envelope = await metadataRequest( + `/accounts/${encodeURIComponent(accountId)}/d1/database?name=${encodeURIComponent(name)}&page=${page}&per_page=${DATABASE_LIST_PAGE_SIZE}`, + token, + fetch, + ); + if (!Array.isArray(envelope.result)) { + throw new Error("Cloudflare D1 database list is invalid."); + } + const resultInfo = listResultInfo(envelope.resultInfo, page, envelope.result.length); + if (totalCount === undefined) { + totalCount = resultInfo.totalCount; + totalPages = resultInfo.totalPages; + } else if (resultInfo.totalCount !== totalCount || resultInfo.totalPages !== totalPages) { + throw new Error("Cloudflare D1 database list pagination is invalid."); + } + if (totalPages > MAX_DATABASE_LIST_PAGES) { + throw new Error("Cloudflare D1 database list pagination is invalid."); + } + seenCount += resultInfo.count; + matches.push( + ...envelope.result.map(metadataDatabase).filter((database) => database.name === name), + ); + page += 1; + } + if (seenCount !== totalCount) { + throw new Error("Cloudflare D1 database list pagination is invalid."); + } + if (matches.length !== 1) { + throw new Error( + matches.length === 0 + ? `No D1 database named ${name} exists in the selected account.` + : `More than one D1 database named ${name} exists in the selected account.`, + ); + } + return matches[0]!; +} + +function listInteger(value: unknown, name: string, minimum: number): number { + if (typeof value !== "number" || !Number.isSafeInteger(value) || value < minimum) { + throw new Error(`Cloudflare D1 database list ${name} is invalid.`); + } + return value; +} + +function listResultInfo(value: unknown, expectedPage: number, resultCount: number): ListResultInfo { + if (!isRecord(value)) throw new Error("Cloudflare D1 database list pagination is invalid."); + const result = { + page: listInteger(value.page, "page", 1), + perPage: listInteger(value.per_page, "per_page", 1), + count: listInteger(value.count, "count", 0), + totalCount: listInteger(value.total_count, "total_count", 0), + totalPages: listInteger(value.total_pages, "total_pages", 0), + }; + if ( + result.page !== expectedPage || + result.perPage > DATABASE_LIST_PAGE_SIZE || + result.count !== resultCount || + result.totalCount < result.count || + (result.totalPages === 0 + ? result.totalCount !== 0 || expectedPage !== 1 + : expectedPage > result.totalPages) + ) { + throw new Error("Cloudflare D1 database list pagination is invalid."); + } + return result; +} + +async function fingerprintTarget(accountId: string, databaseId: string): Promise { + const input = new TextEncoder().encode( + JSON.stringify({ kind: "d1", identity: [accountId, databaseId] }), + ); + const digest = new Uint8Array(await crypto.subtle.digest("SHA-256", input)); + return Array.from(digest, (byte) => byte.toString(16).padStart(2, "0")).join(""); +} + +export async function loadProjectWranglerConfig( + configPath: string, + environment: string | undefined, + projectRoot: string, +): Promise { + const require = createRequire(join(projectRoot, "package.json")); + let entrypoint: string; + try { + entrypoint = require.resolve("wrangler"); + } catch { + throw new Error("D1 target resolution requires Wrangler to be installed in the project."); + } + const loaded: unknown = await import(pathToFileURL(entrypoint).href); + if (!isRecord(loaded)) throw new Error("The project Wrangler module is invalid."); + const readConfig = (loaded as WranglerModule).unstable_readConfig; + if (typeof readConfig !== "function") { + throw new Error("The project Wrangler version does not expose its configuration reader."); + } + const config = await readConfig( + { config: configPath, ...(environment ? { env: environment } : {}) }, + { hideWarnings: true }, + ); + if (!isRecord(config) || !Array.isArray(config.d1_databases)) { + throw new Error("The selected Wrangler configuration is invalid."); + } + return { + accountId: typeof config.account_id === "string" ? config.account_id : undefined, + d1Databases: config.d1_databases.map((binding) => { + if (!isRecord(binding) || typeof binding.binding !== "string") { + throw new Error("The selected Wrangler D1 binding is invalid."); + } + return { + binding: binding.binding, + databaseName: typeof binding.database_name === "string" ? binding.database_name : undefined, + databaseId: typeof binding.database_id === "string" ? binding.database_id : undefined, + previewDatabaseId: + typeof binding.preview_database_id === "string" ? binding.preview_database_id : undefined, + }; + }), + }; +} + +export async function resolveD1MigrationTarget( + manifestConfig: D1MigrationManifestConfig, + context: MigrationExecutorFactoryContext, + dependencies: D1TargetResolutionDependencies = {}, +): Promise { + if (!isRecord(manifestConfig) || !BINDING_PATTERN.test(manifestConfig.binding)) { + throw new Error("The D1 migration binding is invalid."); + } + const configOverride = context.overrides?.wranglerConfig; + const environmentOverride = context.overrides?.wranglerEnv; + if (environmentOverride && !configOverride) { + throw new Error("A Wrangler environment requires an explicit Wrangler configuration path."); + } + const readWranglerConfig = dependencies.readWranglerConfig ?? loadProjectWranglerConfig; + let wranglerConfig: WranglerMigrationConfig | undefined; + if (configOverride) { + const configPath = isAbsolute(configOverride) + ? configOverride + : resolve(context.projectRoot, configOverride); + wranglerConfig = await readWranglerConfig(configPath, environmentOverride, context.projectRoot); + } + + const explicitAccountId = context.overrides?.accountId; + if ( + explicitAccountId && + wranglerConfig?.accountId && + canonicalAccountId(explicitAccountId) !== canonicalAccountId(wranglerConfig.accountId) + ) { + throw new Error("The explicit and Wrangler Cloudflare account IDs conflict."); + } + const accountId = canonicalAccountId( + explicitAccountId ?? wranglerConfig?.accountId ?? context.env.CLOUDFLARE_ACCOUNT_ID, + ); + const token = context.env.CLOUDFLARE_API_TOKEN; + if (!token) throw new Error("CLOUDFLARE_API_TOKEN is required for D1 migrations."); + const fetch = dependencies.fetch ?? globalThis.fetch; + const selector = context.overrides?.d1; + let metadata: { uuid: string; name: string }; + + if (selector) { + if (DATABASE_ID_PATTERN.test(selector)) { + metadata = await databaseById(accountId, canonicalDatabaseId(selector), token, fetch); + } else { + metadata = await databaseByName(accountId, databaseName(selector), token, fetch); + } + } else { + if (!wranglerConfig) { + throw new Error("D1 migrations require an explicit database selector or Wrangler config."); + } + const bindings = wranglerConfig.d1Databases.filter( + (binding) => binding.binding === manifestConfig.binding, + ); + if (bindings.length !== 1) { + throw new Error( + `Wrangler must contain exactly one D1 binding named ${manifestConfig.binding}.`, + ); + } + const binding = bindings[0]!; + if (binding.previewDatabaseId !== undefined) { + throw new Error("Preview D1 database IDs cannot be used for deployment migrations."); + } + const configuredId = canonicalDatabaseId(binding.databaseId); + const configuredName = databaseName(binding.databaseName); + metadata = await databaseById(accountId, configuredId, token, fetch); + if (metadata.name !== configuredName) { + throw new Error("Wrangler D1 database metadata does not match the selected remote database."); + } + } + + const environment = configOverride ? environmentOverride || "top-level" : undefined; + const target = Object.freeze({ + kind: "d1", + label: `${accountId}/${metadata.name}/${metadata.uuid}`, + fingerprint: await fingerprintTarget(accountId, metadata.uuid), + accountId, + resourceId: metadata.uuid, + ...(environment ? { environment } : {}), + }); + return Object.freeze({ + target, + accountId, + databaseId: metadata.uuid, + databaseName: metadata.name, + }); +} diff --git a/packages/cloudflare/src/db/d1-migrations.ts b/packages/cloudflare/src/db/d1-migrations.ts new file mode 100644 index 000000000..9fefc3739 --- /dev/null +++ b/packages/cloudflare/src/db/d1-migrations.ts @@ -0,0 +1,26 @@ +import { + createDirectMigrationExecutor, + type MigrationExecutor, + type MigrationExecutorFactoryContext, +} from "emdash/migrations"; + +import { resolveD1MigrationTarget, type D1MigrationManifestConfig } from "./d1-migration-target.js"; +import { D1RestDialect } from "./d1-rest-dialect.js"; + +export async function createMigrationExecutor( + manifestConfig: D1MigrationManifestConfig, + context: MigrationExecutorFactoryContext, +): Promise { + const resolved = await resolveD1MigrationTarget(manifestConfig, context); + const token = context.env.CLOUDFLARE_API_TOKEN; + if (!token) throw new Error("CLOUDFLARE_API_TOKEN is required for D1 migrations."); + return createDirectMigrationExecutor({ + target: resolved.target, + createDialect: () => + new D1RestDialect({ + accountId: resolved.accountId, + databaseId: resolved.databaseId, + token, + }), + }); +} diff --git a/packages/cloudflare/src/db/d1-rest-dialect.ts b/packages/cloudflare/src/db/d1-rest-dialect.ts new file mode 100644 index 000000000..c54e31bf6 --- /dev/null +++ b/packages/cloudflare/src/db/d1-rest-dialect.ts @@ -0,0 +1,356 @@ +import type { + CompiledQuery, + DatabaseConnection, + DatabaseIntrospector, + Dialect, + Driver, + Kysely, + QueryResult, +} from "kysely"; +import { SqliteQueryCompiler } from "kysely"; + +import { D1Adapter } from "./d1-dialect.js"; +import { D1Introspector } from "./d1-introspector.js"; + +const DEFAULT_TIMEOUT_MS = 30_000; +const DEFAULT_MAX_RESPONSE_BYTES = 1_048_576; +const JSON_CONTENT_TYPE_PATTERN = /^application\/(?:[a-z0-9.+-]+\+)?json\b/i; +const READ_STATEMENT_PATTERN = /^\s*(?:select|explain)\b/i; + +type JsonParameter = string | number | null | number[]; + +export interface D1RestDialectConfig { + accountId: string; + databaseId: string; + token: string; + fetch?: typeof globalThis.fetch; + timeoutMs?: number; + maxResponseBytes?: number; +} + +export class D1RestError extends Error { + constructor(message: string) { + super(message); + this.name = "D1RestError"; + } +} + +export class D1AmbiguousWriteError extends D1RestError { + constructor() { + super( + "The D1 write outcome is ambiguous because its response was not received safely. Run `emdash migrate --status` before retrying.", + ); + this.name = "D1AmbiguousWriteError"; + } +} + +class D1DefinitiveResponseError extends D1RestError {} + +interface ValidatedStatementResult { + rows: Record[]; + changes: number; + lastRowId: number | null; +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function safeInteger(value: unknown, name: string, nullable = false): number | null { + if (nullable && value === null) return null; + if (typeof value !== "number" || !Number.isSafeInteger(value) || value < 0) { + throw new D1RestError(`D1 response metadata field ${name} is invalid.`); + } + return value; +} + +function finiteNumber(value: unknown, name: string): number { + if (typeof value !== "number" || !Number.isFinite(value) || value < 0) { + throw new D1RestError(`D1 response metadata field ${name} is invalid.`); + } + return value; +} + +function validateMessages(value: unknown, name: string): Array<{ code: number; message: string }> { + if (!Array.isArray(value)) throw new D1RestError(`D1 response ${name} is invalid.`); + return value.map((item) => { + if ( + !isRecord(item) || + typeof item.code !== "number" || + !Number.isSafeInteger(item.code) || + typeof item.message !== "string" + ) { + throw new D1RestError(`D1 response ${name} is invalid.`); + } + return { code: item.code, message: item.message }; + }); +} + +function redactMessage(message: string, token: string): string { + const safe: string[] = []; + let length = 0; + for (let index = 0; index < message.length && length < 1_000;) { + let chunk: string; + if (token && message.startsWith(token, index)) { + chunk = "[redacted]"; + index += token.length; + } else { + const code = message.codePointAt(index)!; + chunk = code < 0x20 || code === 0x7f ? " " : String.fromCodePoint(code); + index += code > 0xffff ? 2 : 1; + } + const remaining = 1_000 - length; + const bounded = chunk.slice(0, remaining); + safe.push(bounded); + length += bounded.length; + } + return safe.join(""); +} + +function validateStatementResponse(value: unknown, token: string): ValidatedStatementResult { + if (!isRecord(value)) throw new D1RestError("D1 response envelope is invalid."); + const errors = validateMessages(value.errors, "errors"); + validateMessages(value.messages, "messages"); + if (value.success !== true) { + const message = errors.map((error) => error.message).join("; ") || "Cloudflare API failure"; + throw new D1DefinitiveResponseError(`D1 API request failed: ${redactMessage(message, token)}`); + } + if (errors.length !== 0 || !Array.isArray(value.result) || value.result.length !== 1) { + throw new D1RestError("D1 response envelope is invalid."); + } + const statement = value.result[0]; + if (!isRecord(statement) || typeof statement.success !== "boolean") { + throw new D1RestError("D1 statement response is invalid."); + } + if (!statement.success) { + if (typeof statement.error !== "string" || statement.error.length === 0) { + throw new D1RestError("D1 statement response is invalid."); + } + throw new D1DefinitiveResponseError( + `D1 query failed: ${redactMessage(statement.error, token)}`, + ); + } + if (!Array.isArray(statement.results) || !statement.results.every(isRecord)) { + throw new D1RestError("D1 statement rows are invalid."); + } + if (!isRecord(statement.meta)) throw new D1RestError("D1 response metadata is invalid."); + const meta = statement.meta; + if (typeof meta.changed_db !== "boolean") { + throw new D1RestError("D1 response metadata field changed_db is invalid."); + } + const changes = safeInteger(meta.changes, "changes"); + const lastRowId = safeInteger(meta.last_row_id, "last_row_id", true); + finiteNumber(meta.duration, "duration"); + safeInteger(meta.rows_read, "rows_read"); + safeInteger(meta.rows_written, "rows_written"); + safeInteger(meta.size_after, "size_after"); + if (changes === null) throw new D1RestError("D1 response metadata field changes is invalid."); + return { rows: statement.results, changes, lastRowId }; +} + +function normalizeParameter(value: unknown): JsonParameter { + if (value === null || typeof value === "string") return value; + if (typeof value === "boolean") return value ? 1 : 0; + if (typeof value === "number") { + if (!Number.isFinite(value) || (Number.isInteger(value) && !Number.isSafeInteger(value))) { + throw new D1RestError("D1 query parameter is not a safe finite number."); + } + return value; + } + if (value instanceof ArrayBuffer) return [...new Uint8Array(value)]; + if (ArrayBuffer.isView(value)) { + return [...new Uint8Array(value.buffer, value.byteOffset, value.byteLength)]; + } + throw new D1RestError("D1 query parameter type is unsupported."); +} + +export async function readBoundedJson(response: Response, maxBytes: number): Promise { + const contentType = response.headers.get("content-type") ?? ""; + if (!JSON_CONTENT_TYPE_PATTERN.test(contentType)) { + throw new D1RestError("D1 response content type is not JSON."); + } + const declaredLength = response.headers.get("content-length"); + if (declaredLength !== null) { + const parsedLength = Number(declaredLength); + if (!Number.isSafeInteger(parsedLength) || parsedLength < 0 || parsedLength > maxBytes) { + throw new D1RestError("D1 response is too large."); + } + } + if (!response.body) throw new D1RestError("D1 response body is missing."); + + const reader = response.body.getReader(); + const chunks: Uint8Array[] = []; + let total = 0; + while (true) { + const chunk = await reader.read(); + if (chunk.done) break; + total += chunk.value.byteLength; + if (total > maxBytes) { + await reader.cancel(); + throw new D1RestError("D1 response is too large."); + } + chunks.push(chunk.value); + } + const bytes = new Uint8Array(total); + let offset = 0; + for (const chunk of chunks) { + bytes.set(chunk, offset); + offset += chunk.byteLength; + } + try { + return JSON.parse(new TextDecoder().decode(bytes)); + } catch { + throw new D1RestError("D1 response body is not valid JSON."); + } +} + +class D1RestTransport { + readonly #config: Required> & { + fetch: typeof globalThis.fetch; + }; + readonly #controllers = new Set(); + #destroyed = false; + + constructor(config: D1RestDialectConfig) { + this.#config = { + accountId: config.accountId, + databaseId: config.databaseId, + token: config.token, + fetch: config.fetch ?? globalThis.fetch, + timeoutMs: config.timeoutMs ?? DEFAULT_TIMEOUT_MS, + maxResponseBytes: config.maxResponseBytes ?? DEFAULT_MAX_RESPONSE_BYTES, + }; + } + + async query(sql: string, parameters: readonly unknown[]): Promise { + const params = parameters.map(normalizeParameter); + if (this.#destroyed) throw new D1RestError("The D1 REST transport has been disposed."); + const controller = new AbortController(); + this.#controllers.add(controller); + let timedOut = false; + const timer = setTimeout(() => { + timedOut = true; + controller.abort(); + }, this.#config.timeoutMs); + const isWrite = !READ_STATEMENT_PATTERN.test(sql); + + try { + const response = await this.#config.fetch( + `https://api.cloudflare.com/client/v4/accounts/${encodeURIComponent(this.#config.accountId)}/d1/database/${encodeURIComponent(this.#config.databaseId)}/query`, + { + method: "POST", + headers: { + authorization: `Bearer ${this.#config.token}`, + "content-type": "application/json", + }, + body: JSON.stringify({ sql, params }), + redirect: "error", + signal: controller.signal, + }, + ); + if (!response.ok) { + if (isWrite && response.status >= 500) throw new D1AmbiguousWriteError(); + throw new D1DefinitiveResponseError( + `D1 API request failed with HTTP status ${response.status}.`, + ); + } + const body = await readBoundedJson(response, this.#config.maxResponseBytes); + return validateStatementResponse(body, this.#config.token); + } catch (error) { + if (error instanceof D1AmbiguousWriteError) throw error; + if (error instanceof D1DefinitiveResponseError) throw error; + if (isWrite) throw new D1AmbiguousWriteError(); + if (error instanceof D1RestError) throw error; + if (timedOut) throw new D1RestError("D1 query timed out."); + throw new D1RestError("D1 query request failed before a valid response was received."); + } finally { + clearTimeout(timer); + this.#controllers.delete(controller); + } + } + + destroy(): void { + this.#destroyed = true; + for (const controller of this.#controllers) controller.abort(); + this.#controllers.clear(); + } +} + +export class D1RestDialect implements Dialect { + readonly #config: D1RestDialectConfig; + + constructor(config: D1RestDialectConfig) { + this.#config = config; + } + + createAdapter(): D1Adapter { + return new D1Adapter(); + } + + createDriver(): Driver { + return new D1RestDriver(this.#config); + } + + createQueryCompiler(): SqliteQueryCompiler { + return new SqliteQueryCompiler(); + } + + createIntrospector(db: Kysely): DatabaseIntrospector { + return new D1Introspector(db); + } +} + +class D1RestDriver implements Driver { + readonly #transport: D1RestTransport; + + constructor(config: D1RestDialectConfig) { + this.#transport = new D1RestTransport(config); + } + + async init(): Promise {} + + async acquireConnection(): Promise { + return new D1RestConnection(this.#transport); + } + + async beginTransaction(): Promise { + throw new Error("Transactions are not supported yet."); + } + + async commitTransaction(): Promise { + throw new Error("Transactions are not supported yet."); + } + + async rollbackTransaction(): Promise { + throw new Error("Transactions are not supported yet."); + } + + async releaseConnection(): Promise {} + + async destroy(): Promise { + this.#transport.destroy(); + } +} + +class D1RestConnection implements DatabaseConnection { + readonly #transport: D1RestTransport; + + constructor(transport: D1RestTransport) { + this.#transport = transport; + } + + async executeQuery(compiledQuery: CompiledQuery): Promise> { + const result = await this.#transport.query(compiledQuery.sql, compiledQuery.parameters); + return { + // eslint-disable-next-line typescript/no-unsafe-type-assertion -- validated row objects are mapped to Kysely's caller-selected row type + rows: result.rows as Row[], + numAffectedRows: BigInt(result.changes), + insertId: result.lastRowId === null ? undefined : BigInt(result.lastRowId), + }; + } + + // eslint-disable-next-line require-yield -- the administrative D1 transport does not stream + async *streamQuery(): AsyncIterableIterator> { + throw new Error("D1 REST dialect does not support streaming."); + } +} diff --git a/packages/cloudflare/src/db/hyperdrive-migrations.ts b/packages/cloudflare/src/db/hyperdrive-migrations.ts new file mode 100644 index 000000000..6dd9b9385 --- /dev/null +++ b/packages/cloudflare/src/db/hyperdrive-migrations.ts @@ -0,0 +1,26 @@ +import { createMigrationExecutor as createPostgresMigrationExecutor } from "emdash/db/postgres-migrations"; +import type { MigrationExecutor, MigrationExecutorFactoryContext } from "emdash/migrations"; + +const BINDING_PATTERN = /^[A-Za-z_$][A-Za-z0-9_$]*$/; + +export interface HyperdriveMigrationManifestConfig { + binding: string; + connectionStringEnv: string; +} + +export async function createMigrationExecutor( + manifestConfig: HyperdriveMigrationManifestConfig, + context: MigrationExecutorFactoryContext, +): Promise { + if ( + typeof manifestConfig !== "object" || + manifestConfig === null || + !BINDING_PATTERN.test(manifestConfig.binding) + ) { + throw new Error("The Hyperdrive migration binding is invalid."); + } + return createPostgresMigrationExecutor( + { connectionStringEnv: manifestConfig.connectionStringEnv }, + context, + ); +} diff --git a/packages/cloudflare/src/db/hyperdrive.ts b/packages/cloudflare/src/db/hyperdrive.ts index 470d005d8..18177c5bd 100644 --- a/packages/cloudflare/src/db/hyperdrive.ts +++ b/packages/cloudflare/src/db/hyperdrive.ts @@ -41,7 +41,7 @@ * authenticated request, every write, and every request under `/_emdash` * (admin, setup, auth, internal APIs) — including anonymous GETs such as the * post-setup status check, which must observe a write made moments earlier. - * Migrations and the per-isolate singleton always use the primary binding. + * Runtime migrations and the per-isolate singleton always use the primary binding. * Omit `cachedBinding` and the adapter behaves exactly as before. * * Known limitation — sandboxed plugins are D1-only. The sandbox plugin bridge @@ -324,7 +324,7 @@ function getBinding(bindingName: string): HyperdriveBinding | null { } function requireBinding(config: HyperdriveConfig): HyperdriveBinding { - // Migrations and the per-isolate singleton always use the primary binding — + // Runtime migrations and the per-isolate singleton always use the primary binding — // never the cache-enabled one. const binding = getBinding(config.binding); if (!binding) { diff --git a/packages/cloudflare/src/index.ts b/packages/cloudflare/src/index.ts index 41782d0e6..8ab6e8b01 100644 --- a/packages/cloudflare/src/index.ts +++ b/packages/cloudflare/src/index.ts @@ -43,6 +43,8 @@ import type { import type { DurableObjectsConfig } from "./db/do-sql-types.js"; import type { PreviewDOConfig } from "./db/do-types.js"; +const ENVIRONMENT_VARIABLE_PATTERN = /^[A-Za-z_][A-Za-z0-9_]*$/; + /** * D1 configuration */ @@ -136,8 +138,8 @@ export interface HyperdriveConfig { * (uncached) to preserve read-after-write consistency: every authenticated * request, every write, and every request under `/_emdash` (admin, setup, * auth, internal APIs) — including anonymous GETs like the post-setup status - * check, which must observe a write made moments earlier. Migrations and the - * cold-start singleton always use `binding`. + * check, which must observe a write made moments earlier. Runtime migrations + * and the cold-start singleton always use `binding`. * * After a content publish, EmDash prefers the primary uncached binding for * anonymous public reads for a short window (see @@ -170,6 +172,17 @@ export interface HyperdriveConfig { */ preferUncachedAfterWriteMs?: number; + /** + * Environment variable containing a PostgreSQL connection string that the + * deployment migration command can use to reach the database origin + * directly. This value is read only by `emdash migrate`; Worker requests + * continue to use the Hyperdrive binding. + * + * @default `CLOUDFLARE_HYPERDRIVE_LOCAL_CONNECTION_STRING_` + * (`$` in the binding is represented as `_`.) + */ + migrationConnectionStringEnv?: string; + /** * Maximum size of the in-Worker node-postgres connection pool. * @@ -259,7 +272,8 @@ export interface AccessConfig { * Cloudflare D1 database adapter * * For Cloudflare Workers with D1 binding. - * Migrations run automatically at setup time - no need for manual SQL files. + * Runtime migrations run automatically by default; deployment-managed + * migrations can be applied from the build manifest before deploy. * * Uses a custom introspector that works around D1's restriction on * cross-joins with pragma_table_info(). @@ -274,6 +288,10 @@ export function d1(config: D1Config): DatabaseDescriptor { entrypoint: "@emdash-cms/cloudflare/db/d1", config, type: "sqlite", + migrations: { + entrypoint: "@emdash-cms/cloudflare/db/d1-migrations", + manifestConfig: { binding: config.binding }, + }, supportsRequestScope: true, supportsCoalescing: true, supportsCollectionDeletionGuard: true, @@ -352,10 +370,17 @@ export function d1(config: D1Config): DatabaseDescriptor { * ``` */ export function hyperdrive(config: HyperdriveConfig = {}): DatabaseDescriptor { + const binding = config.binding ?? "HYPERDRIVE"; + const connectionStringEnv = + config.migrationConnectionStringEnv ?? + `CLOUDFLARE_HYPERDRIVE_LOCAL_CONNECTION_STRING_${binding.replaceAll("$", "_")}`; + if (!ENVIRONMENT_VARIABLE_PATTERN.test(connectionStringEnv)) { + throw new Error("migrationConnectionStringEnv must be a valid environment variable name."); + } return { entrypoint: "@emdash-cms/cloudflare/db/hyperdrive", config: { - binding: config.binding ?? "HYPERDRIVE", + binding, max: config.max, ...(config.cachedBinding !== undefined ? { cachedBinding: config.cachedBinding } : {}), ...(config.preferUncachedAfterWriteMs !== undefined @@ -363,6 +388,10 @@ export function hyperdrive(config: HyperdriveConfig = {}): DatabaseDescriptor { : {}), }, type: "postgres", + migrations: { + entrypoint: "@emdash-cms/cloudflare/db/hyperdrive-migrations", + manifestConfig: { binding, connectionStringEnv }, + }, // Each request gets a fresh pg connection that is closed afterwards — // connections cannot be reused across Worker requests. supportsRequestScope: true, diff --git a/packages/cloudflare/tests/db/d1-migration-target.test.ts b/packages/cloudflare/tests/db/d1-migration-target.test.ts new file mode 100644 index 000000000..036099380 --- /dev/null +++ b/packages/cloudflare/tests/db/d1-migration-target.test.ts @@ -0,0 +1,264 @@ +import { resolve } from "node:path"; + +import { describe, expect, it, vi } from "vitest"; + +import { + loadProjectWranglerConfig, + resolveD1MigrationTarget, + type WranglerMigrationConfig, +} from "../../src/db/d1-migration-target.js"; + +const ACCOUNT_ID = "0123456789abcdef0123456789abcdef"; +const DATABASE_ID = "11111111-2222-4333-8444-555555555555"; +const TOKEN = "control-plane-secret"; + +function apiResponse(result: unknown): Response { + return Response.json({ success: true, errors: [], messages: [], result }); +} + +function listApiResponse( + result: unknown[], + page: number, + totalPages: number, + totalCount: number, +): Response { + return Response.json({ + success: true, + errors: [], + messages: [], + result, + result_info: { + page, + per_page: 100, + count: result.length, + total_count: totalCount, + total_pages: totalPages, + }, + }); +} + +function database(uuid = DATABASE_ID, name = "site-db"): Record { + return { uuid, name, version: "production" }; +} + +describe("resolveD1MigrationTarget", () => { + it("preflights an explicit UUID and freezes a credential-free target", async () => { + const fetch = vi.fn(async () => apiResponse(database())); + const resolved = await resolveD1MigrationTarget( + { binding: "DB" }, + { + projectRoot: "/project", + env: { CLOUDFLARE_API_TOKEN: TOKEN }, + overrides: { accountId: ACCOUNT_ID, d1: DATABASE_ID }, + }, + { fetch }, + ); + + expect(resolved.target).toEqual({ + kind: "d1", + label: `${ACCOUNT_ID}/site-db/${DATABASE_ID}`, + fingerprint: expect.stringMatching(/^[a-f0-9]{64}$/), + accountId: ACCOUNT_ID, + resourceId: DATABASE_ID, + }); + expect(Object.isFrozen(resolved)).toBe(true); + expect(Object.isFrozen(resolved.target)).toBe(true); + expect(JSON.stringify(resolved.target)).not.toContain(TOKEN); + expect(fetch).toHaveBeenCalledTimes(1); + }); + + it("resolves an explicit name only when exactly one exact match exists", async () => { + const fetch = vi.fn(async (input) => { + const url = new URL(input instanceof Request ? input.url : input.toString()); + return url.searchParams.get("page") === "1" + ? listApiResponse([database(undefined, "site-db-preview")], 1, 2, 2) + : listApiResponse([database(DATABASE_ID, "site-db")], 2, 2, 2); + }); + const resolved = await resolveD1MigrationTarget( + { binding: "DB" }, + { + projectRoot: "/project", + env: { CLOUDFLARE_ACCOUNT_ID: ACCOUNT_ID, CLOUDFLARE_API_TOKEN: TOKEN }, + overrides: { d1: "site-db" }, + }, + { fetch }, + ); + + expect(resolved.databaseId).toBe(DATABASE_ID); + const firstInput = fetch.mock.calls[0]?.[0]; + expect(firstInput instanceof Request ? firstInput.url : firstInput?.toString()).toContain( + "name=site-db", + ); + expect(fetch).toHaveBeenCalledTimes(2); + }); + + it("rejects duplicate exact names found on different result pages", async () => { + const fetch = vi.fn(async (input) => { + const url = new URL(input instanceof Request ? input.url : input.toString()); + const isFirstPage = url.searchParams.get("page") === "1"; + const id = isFirstPage ? DATABASE_ID : "aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee"; + return listApiResponse([database(id, "site-db")], isFirstPage ? 1 : 2, 2, 2); + }); + + await expect( + resolveD1MigrationTarget( + { binding: "DB" }, + { + projectRoot: "/project", + env: { CLOUDFLARE_ACCOUNT_ID: ACCOUNT_ID, CLOUDFLARE_API_TOKEN: TOKEN }, + overrides: { d1: "site-db" }, + }, + { fetch }, + ), + ).rejects.toThrow(/more than one/i); + expect(fetch).toHaveBeenCalledTimes(2); + }); + + it("rejects preview metadata returned for an explicit UUID", async () => { + const fetch = vi.fn(async () => + apiResponse({ ...database(), version: "preview" }), + ); + + await expect( + resolveD1MigrationTarget( + { binding: "DB" }, + { + projectRoot: "/project", + env: { CLOUDFLARE_API_TOKEN: TOKEN }, + overrides: { accountId: ACCOUNT_ID, d1: DATABASE_ID }, + }, + { fetch }, + ), + ).rejects.toThrow(/preview/i); + }); + + it("uses the selected Wrangler environment and its own binding array", async () => { + const config: WranglerMigrationConfig = { + accountId: ACCOUNT_ID, + d1Databases: [{ binding: "DB", databaseName: "production-db", databaseId: DATABASE_ID }], + }; + const readWranglerConfig = vi.fn(async () => config); + const fetch = vi.fn(async () => + apiResponse(database(DATABASE_ID, "production-db")), + ); + const resolved = await resolveD1MigrationTarget( + { binding: "DB" }, + { + projectRoot: "/project", + env: { CLOUDFLARE_API_TOKEN: TOKEN }, + overrides: { wranglerConfig: "wrangler.jsonc", wranglerEnv: "production" }, + }, + { fetch, readWranglerConfig }, + ); + + expect(readWranglerConfig).toHaveBeenCalledWith( + "/project/wrangler.jsonc", + "production", + "/project", + ); + expect(resolved.target.environment).toBe("production"); + }); + + it.each([ + [ + "preview ID", + [ + { + binding: "DB", + databaseName: "site-db", + databaseId: DATABASE_ID, + previewDatabaseId: DATABASE_ID, + }, + ], + ], + [ + "duplicate binding", + [ + { binding: "DB", databaseName: "one", databaseId: DATABASE_ID }, + { binding: "DB", databaseName: "two", databaseId: "aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee" }, + ], + ], + ["placeholder ID", [{ binding: "DB", databaseName: "site-db", databaseId: "" }]], + ])("rejects a configured %s before metadata lookup", async (_name, d1Databases) => { + const fetch = vi.fn(); + + await expect( + resolveD1MigrationTarget( + { binding: "DB" }, + { + projectRoot: "/project", + env: { CLOUDFLARE_API_TOKEN: TOKEN }, + overrides: { wranglerConfig: "wrangler.jsonc" }, + }, + { + fetch, + readWranglerConfig: async () => ({ accountId: ACCOUNT_ID, d1Databases }), + }, + ), + ).rejects.toThrow(); + expect(fetch).not.toHaveBeenCalled(); + }); + + it("rejects conflicting explicit and configured account IDs", async () => { + const fetch = vi.fn(); + await expect( + resolveD1MigrationTarget( + { binding: "DB" }, + { + projectRoot: "/project", + env: { CLOUDFLARE_API_TOKEN: TOKEN }, + overrides: { + accountId: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + d1: DATABASE_ID, + wranglerConfig: "wrangler.jsonc", + }, + }, + { + fetch, + readWranglerConfig: async () => ({ accountId: ACCOUNT_ID, d1Databases: [] }), + }, + ), + ).rejects.toThrow(/account.*conflict/i); + expect(fetch).not.toHaveBeenCalled(); + }); +}); + +describe("loadProjectWranglerConfig", () => { + it("uses project-local Wrangler and preserves account inheritance for a named environment", async () => { + const cloudflarePackage = resolve(import.meta.dirname, "../.."); + const projectRoot = resolve(cloudflarePackage, "../marketplace"); + const configPath = resolve(import.meta.dirname, "../fixtures/d1-wrangler.jsonc"); + + await expect(loadProjectWranglerConfig(configPath, "production", projectRoot)).resolves.toEqual( + { + accountId: ACCOUNT_ID, + d1Databases: [ + { + binding: "DB", + databaseName: "production-db", + databaseId: "aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee", + previewDatabaseId: undefined, + }, + ], + }, + ); + }); + + it("omits the environment when selecting the top-level configuration", async () => { + const cloudflarePackage = resolve(import.meta.dirname, "../.."); + const projectRoot = resolve(cloudflarePackage, "../marketplace"); + const configPath = resolve(import.meta.dirname, "../fixtures/d1-wrangler.jsonc"); + + await expect(loadProjectWranglerConfig(configPath, undefined, projectRoot)).resolves.toEqual({ + accountId: ACCOUNT_ID, + d1Databases: [ + { + binding: "DB", + databaseName: "top-level-db", + databaseId: DATABASE_ID, + previewDatabaseId: undefined, + }, + ], + }); + }); +}); diff --git a/packages/cloudflare/tests/db/d1-migrations.live.test.ts b/packages/cloudflare/tests/db/d1-migrations.live.test.ts new file mode 100644 index 000000000..89295232f --- /dev/null +++ b/packages/cloudflare/tests/db/d1-migrations.live.test.ts @@ -0,0 +1,88 @@ +import { getCoreMigrationIdentity } from "emdash/migrations"; +import { Kysely, sql } from "kysely"; +import { describe, expect, it } from "vitest"; + +import { createMigrationExecutor } from "../../src/db/d1-migrations.js"; +import { D1RestDialect } from "../../src/db/d1-rest-dialect.js"; + +const accountId = process.env.EMDASH_TEST_D1_ACCOUNT_ID ?? ""; +const databaseId = process.env.EMDASH_TEST_D1_DATABASE_ID ?? ""; +const token = process.env.CLOUDFLARE_API_TOKEN ?? ""; +const hasLiveD1 = + process.env.EMDASH_TEST_D1_DISPOSABLE === "1" && + accountId.length > 0 && + databaseId.length > 0 && + token.length > 0; + +describe.skipIf(!hasLiveD1)("D1 migration executor live", () => { + it("applies the registry to an explicitly disposable database and is idempotent", async () => { + const context = { + projectRoot: process.cwd(), + env: { CLOUDFLARE_API_TOKEN: token }, + overrides: { accountId, d1: databaseId }, + }; + const identity = await getCoreMigrationIdentity(); + const request = { + action: "apply" as const, + i18n: null, + artifact: { + emdashVersion: identity.emdashVersion, + migrationSetFingerprint: identity.fingerprint, + }, + }; + + const first = await createMigrationExecutor({ binding: "DB" }, context); + await expect(first.execute(request)).resolves.toMatchObject({ pending: [] }); + const second = await createMigrationExecutor({ binding: "DB" }, context); + await expect(second.execute(request)).resolves.toMatchObject({ pending: [], executed: [] }); + }); + + it("supports migration introspection, values, and duplicate-error recovery", async () => { + const db = new Kysely>({ + dialect: new D1RestDialect({ accountId, databaseId, token }), + }); + try { + await sql`drop table if exists _emdash_d1_live_contract`.execute(db); + await sql`create table _emdash_d1_live_contract ( + id integer primary key, + text_value text not null unique, + number_value real not null, + null_value text, + boolean_value integer not null, + blob_value blob not null + )`.execute(db); + await sql`insert into _emdash_d1_live_contract + (id, text_value, number_value, null_value, boolean_value, blob_value) + values (${1}, ${"contract"}, ${1.5}, ${null}, ${true}, ${new Uint8Array([0, 127, 255])})`.execute( + db, + ); + + const result = await sql<{ + text_value: string; + number_value: number; + null_value: null; + boolean_value: number; + blob_value: number[]; + }>`select text_value, number_value, null_value, boolean_value, blob_value + from _emdash_d1_live_contract where id = ${1}`.execute(db); + expect(result.rows[0]).toEqual({ + text_value: "contract", + number_value: 1.5, + null_value: null, + boolean_value: 1, + blob_value: [0, 127, 255], + }); + await expect(db.introspection.getTables()).resolves.toEqual( + expect.arrayContaining([expect.objectContaining({ name: "_emdash_d1_live_contract" })]), + ); + await expect( + sql`insert into _emdash_d1_live_contract + (id, text_value, number_value, boolean_value, blob_value) + values (${2}, ${"contract"}, ${2}, ${false}, ${new Uint8Array()})`.execute(db), + ).rejects.toThrow(/UNIQUE constraint failed: _emdash_d1_live_contract\.text_value/); + } finally { + await sql`drop table if exists _emdash_d1_live_contract`.execute(db); + await db.destroy(); + } + }); +}); diff --git a/packages/cloudflare/tests/db/d1-migrations.test.ts b/packages/cloudflare/tests/db/d1-migrations.test.ts new file mode 100644 index 000000000..9661783ff --- /dev/null +++ b/packages/cloudflare/tests/db/d1-migrations.test.ts @@ -0,0 +1,167 @@ +import { getCoreMigrationIdentity } from "emdash/migrations"; +import { afterEach, describe, expect, it, vi } from "vitest"; + +import { createMigrationExecutor } from "../../src/db/d1-migrations.js"; + +const ACCOUNT_ID = "0123456789abcdef0123456789abcdef"; +const DATABASE_ID = "11111111-2222-4333-8444-555555555555"; +const TOKEN = "d1-test-token"; +const originalFetch = globalThis.fetch; + +function response(result: unknown): Response { + return Response.json({ success: true, errors: [], messages: [], result }); +} + +function queryResponse(results: Record[] = []): Response { + return response([ + { + success: true, + results, + meta: { + changed_db: false, + changes: 0, + duration: 0.1, + last_row_id: null, + rows_read: results.length, + rows_written: 0, + size_after: 4096, + }, + }, + ]); +} + +afterEach(() => { + globalThis.fetch = originalFetch; +}); + +describe("D1 migration executor", () => { + it("constructs from remote metadata without issuing SQL, then checks through the REST dialect", async () => { + const fetch = vi.fn(async (input) => { + const url = input instanceof Request ? input.url : input.toString(); + if (!url.endsWith("/query")) { + return response({ uuid: DATABASE_ID, name: "site-db", version: "production" }); + } + return response([ + { + success: false, + error: "no such table: _emdash_migrations", + results: [], + }, + ]); + }); + globalThis.fetch = fetch; + const executor = await createMigrationExecutor( + { binding: "DB" }, + { + projectRoot: "/project", + env: { CLOUDFLARE_API_TOKEN: TOKEN }, + overrides: { accountId: ACCOUNT_ID, d1: DATABASE_ID }, + }, + ); + + expect(fetch).toHaveBeenCalledTimes(1); + const firstInput = fetch.mock.calls[0]?.[0]; + expect(firstInput instanceof Request ? firstInput.url : firstInput?.toString()).not.toContain( + "/query", + ); + const identity = await getCoreMigrationIdentity(); + await expect( + executor.execute({ + action: "check", + i18n: null, + artifact: { + emdashVersion: identity.emdashVersion, + migrationSetFingerprint: identity.fingerprint, + }, + }), + ).resolves.toMatchObject({ pending: identity.names, executed: [] }); + expect(fetch).toHaveBeenCalledTimes(2); + }); + + it("applies a pending migration through the REST dialect", async () => { + const identity = await getCoreMigrationIdentity(); + const pendingMigration = identity.names.at(-1); + if (!pendingMigration) throw new Error("Expected at least one core migration."); + + const applied = new Set(identity.names.slice(0, -1)); + const tables = ["_emdash_migrations", "_emdash_migrations_lock", "_emdash_collections"]; + const requests: Array<{ sql: string; params: unknown[] }> = []; + const fetch = vi.fn(async (input, init) => { + const url = input instanceof Request ? input.url : input.toString(); + if (!url.endsWith("/query")) { + return response({ uuid: DATABASE_ID, name: "site-db", version: "production" }); + } + if (typeof init?.body !== "string") throw new Error("Expected a JSON query body."); + const request = JSON.parse(init.body) as { sql: string; params: unknown[] }; + requests.push(request); + + if (/\bfrom\s+["`]?sqlite_master["`]?/i.test(request.sql)) { + return queryResponse( + tables.map((name) => ({ + name, + type: "table", + sql: `CREATE TABLE "${name}" (id TEXT)`, + })), + ); + } + if (/\bfrom\s+["`]?_emdash_migrations["`]?\b/i.test(request.sql)) { + if (/count\(\*\)/i.test(request.sql)) { + return queryResponse([{ count: applied.size }]); + } + return queryResponse( + Array.from(applied, (name, index) => ({ + name, + timestamp: new Date(index).toISOString(), + })), + ); + } + if (/\binsert\s+into\s+["`]?_emdash_migrations["`]?\b/i.test(request.sql)) { + applied.add(String(request.params[0])); + } + return queryResponse(); + }); + globalThis.fetch = fetch; + + const executor = await createMigrationExecutor( + { binding: "DB" }, + { + projectRoot: "/project", + env: { CLOUDFLARE_API_TOKEN: TOKEN }, + overrides: { accountId: ACCOUNT_ID, d1: DATABASE_ID }, + }, + ); + + await expect( + executor.execute({ + action: "apply", + i18n: null, + artifact: { + emdashVersion: identity.emdashVersion, + migrationSetFingerprint: identity.fingerprint, + }, + }), + ).resolves.toMatchObject({ pending: [], executed: [pendingMigration] }); + expect( + requests.find((request) => + /\binsert\s+into\s+["`]?_emdash_migrations["`]?\b/i.test(request.sql), + )?.params[0], + ).toBe(pendingMigration); + }); + + it("fails without the API token before making a metadata request", async () => { + const fetch = vi.fn(); + globalThis.fetch = fetch; + + await expect( + createMigrationExecutor( + { binding: "DB" }, + { + projectRoot: "/project", + env: {}, + overrides: { accountId: ACCOUNT_ID, d1: DATABASE_ID }, + }, + ), + ).rejects.toThrow("CLOUDFLARE_API_TOKEN"); + expect(fetch).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/cloudflare/tests/db/d1-rest-dialect.test.ts b/packages/cloudflare/tests/db/d1-rest-dialect.test.ts new file mode 100644 index 000000000..ae9ad06d1 --- /dev/null +++ b/packages/cloudflare/tests/db/d1-rest-dialect.test.ts @@ -0,0 +1,362 @@ +import { CompiledQuery, Kysely, sql } from "kysely"; +import { describe, expect, it, vi } from "vitest"; + +import { D1AmbiguousWriteError, D1RestDialect } from "../../src/db/d1-rest-dialect.js"; + +const ACCOUNT_ID = "0123456789abcdef0123456789abcdef"; +const DATABASE_ID = "11111111-2222-4333-8444-555555555555"; +const TOKEN = "cloudflare-secret-token"; + +function queryEnvelope( + results: Array> = [], + meta: Record = {}, +): Record { + return { + success: true, + errors: [], + messages: [], + result: [ + { + success: true, + results, + meta: { + changed_db: false, + changes: 0, + duration: 0.1, + last_row_id: 0, + rows_read: results.length, + rows_written: 0, + size_after: 4096, + ...meta, + }, + }, + ], + }; +} + +function jsonResponse(body: unknown, init: ResponseInit = {}): Response { + const headers = new Headers(init.headers); + headers.set("content-type", "application/json"); + return Response.json(body, { + ...init, + headers, + }); +} + +function database( + fetch: typeof globalThis.fetch, + options: { timeoutMs?: number; maxResponseBytes?: number } = {}, +) { + return new Kysely>({ + dialect: new D1RestDialect({ + accountId: ACCOUNT_ID, + databaseId: DATABASE_ID, + token: TOKEN, + fetch, + ...options, + }), + }); +} + +describe("D1RestDialect", () => { + it("sends one compiled query with ordered normalized parameters", async () => { + const fetch = vi.fn(async (_input, init) => { + expect(init?.method).toBe("POST"); + expect(init?.redirect).toBe("error"); + expect(new Headers(init?.headers).get("authorization")).toBe(`Bearer ${TOKEN}`); + if (typeof init?.body !== "string") throw new Error("Expected a JSON request body."); + expect(JSON.parse(init.body)).toEqual({ + sql: "select ?, ?, ?, ?", + params: ["text", 42, null, 1], + }); + return jsonResponse(queryEnvelope([{ value: "ok" }])); + }); + const db = database(fetch); + + await expect( + sql<{ value: string }>`select ${"text"}, ${42}, ${null}, ${true}`.execute(db), + ).resolves.toMatchObject({ rows: [{ value: "ok" }] }); + expect(fetch).toHaveBeenCalledTimes(1); + await db.destroy(); + }); + + it.each([ + [Number.NaN], + [Number.POSITIVE_INFINITY], + [Number.MAX_SAFE_INTEGER + 1], + [123n], + [{ unsafe: true }], + ])("rejects an unsupported parameter before sending it", async (parameter) => { + const fetch = vi.fn(); + const db = database(fetch); + + await expect(sql`select ${parameter}`.execute(db)).rejects.toThrow(/parameter/i); + expect(fetch).not.toHaveBeenCalled(); + await db.destroy(); + }); + + it("rejects a non-success HTTP status", async () => { + const fetch = vi.fn(async () => + jsonResponse( + { + success: false, + errors: [{ code: 9109, message: "Unauthorized" }], + messages: [], + result: null, + }, + { status: 401 }, + ), + ); + const db = database(fetch); + + await expect(sql`select 1`.execute(db)).rejects.toThrow(/HTTP status 401/i); + expect(fetch).toHaveBeenCalledTimes(1); + await db.destroy(); + }); + + it("preserves a top-level Cloudflare API failure", async () => { + const fetch = vi.fn(async () => + jsonResponse({ + success: false, + errors: [{ code: 7500, message: "D1 API unavailable" }], + messages: [], + result: null, + }), + ); + const db = database(fetch); + + await expect(sql`select 1`.execute(db)).rejects.toThrow(/D1 API unavailable/); + await db.destroy(); + }); + + it("maps validated affected-row and insert metadata", async () => { + const fetch = vi.fn(async () => + jsonResponse( + queryEnvelope([], { + changed_db: true, + changes: 2, + last_row_id: 17, + rows_written: 2, + }), + ), + ); + const db = database(fetch); + + await expect(sql`insert into example values (1)`.execute(db)).resolves.toMatchObject({ + insertId: 17n, + numAffectedRows: 2n, + }); + await db.destroy(); + }); + + it.each([ + ["non-JSON content", new Response("not json", { headers: { "content-type": "text/plain" } })], + ["malformed envelope", jsonResponse({ success: true, result: [] })], + ["multiple statement results", jsonResponse({ ...queryEnvelope(), result: [{}, {}] })], + ["invalid metadata", jsonResponse(queryEnvelope([], { changes: -1 }))], + ])("rejects a %s", async (_name, response) => { + const fetch = vi.fn(async () => response); + const db = database(fetch); + + await expect(sql`select 1`.execute(db)).rejects.toThrow(/D1/i); + await db.destroy(); + }); + + it("rejects a declared response larger than the bound", async () => { + const fetch = vi.fn( + async () => + new Response("{}", { + headers: { + "content-length": "1000", + "content-type": "application/json", + }, + }), + ); + const db = database(fetch, { maxResponseBytes: 100 }); + + await expect(sql`select 1`.execute(db)).rejects.toThrow(/response.*large/i); + await db.destroy(); + }); + + it("rejects a streamed response larger than the bound", async () => { + const fetch = vi.fn( + async () => + new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new Uint8Array(60)); + controller.enqueue(new Uint8Array(60)); + controller.close(); + }, + }), + { headers: { "content-type": "application/json" } }, + ), + ); + const db = database(fetch, { maxResponseBytes: 100 }); + + await expect(sql`select 1`.execute(db)).rejects.toThrow(/response.*large/i); + await db.destroy(); + }); + + it("preserves a statement error needed for duplicate-record recovery", async () => { + const fetch = vi.fn(async () => + jsonResponse({ + ...queryEnvelope(), + result: [ + { + success: false, + error: "UNIQUE constraint failed: _emdash_migrations.name", + results: [], + meta: queryEnvelope().result, + }, + ], + }), + ); + const db = database(fetch); + + await expect(sql`insert into _emdash_migrations values (1)`.execute(db)).rejects.toThrow( + "UNIQUE constraint failed: _emdash_migrations.name", + ); + await db.destroy(); + }); + + it("bounds and sanitizes upstream error messages", async () => { + const fetch = vi.fn(async () => + jsonResponse({ + success: false, + errors: [ + { + code: 10_000, + message: `before\t${TOKEN}\n${"x".repeat(2_000)}`, + }, + ], + messages: [], + result: [], + }), + ); + const db = database(fetch); + + let error: unknown; + try { + await sql`select 1`.execute(db); + } catch (caught) { + error = caught; + } + + expect(error).toBeInstanceOf(Error); + expect((error as Error).message).toContain("before [redacted] "); + expect((error as Error).message).not.toContain(TOKEN); + expect((error as Error).message).not.toMatch(/[\r\n\t]/); + expect((error as Error).message.length).toBeLessThanOrEqual( + "D1 API request failed: ".length + 1_000, + ); + await db.destroy(); + }); + + it("does not retry an ambiguous write failure and never leaks the token", async () => { + const fetch = vi.fn(async () => { + throw new TypeError(`network failed near ${TOKEN}`); + }); + const db = database(fetch); + + let error: unknown; + try { + await sql`create table example (id integer)`.execute(db); + } catch (caught) { + error = caught; + } + + expect(error).toBeInstanceOf(D1AmbiguousWriteError); + expect((error as Error).message).toMatch(/ambiguous.*--status/i); + expect((error as Error).message).not.toContain(TOKEN); + expect(fetch).toHaveBeenCalledTimes(1); + await db.destroy(); + }); + + it.each([ + ["malformed", jsonResponse({ success: true, result: [] })], + [ + "oversized", + new Response("{}", { + headers: { + "content-length": "1000", + "content-type": "application/json", + }, + }), + ], + ])("classifies a %s successful write response as ambiguous", async (_name, response) => { + const fetch = vi.fn(async () => response); + const db = database(fetch, { maxResponseBytes: 100 }); + + await expect(sql`create table example (id integer)`.execute(db)).rejects.toBeInstanceOf( + D1AmbiguousWriteError, + ); + expect(fetch).toHaveBeenCalledTimes(1); + await db.destroy(); + }); + + it("keeps an explicit statement failure definitive", async () => { + const fetch = vi.fn(async () => + jsonResponse({ + ...queryEnvelope(), + result: [{ success: false, error: "syntax error", results: [] }], + }), + ); + const db = database(fetch); + + let error: unknown; + try { + await sql`create table example (id integer`.execute(db); + } catch (caught) { + error = caught; + } + expect(error).toBeInstanceOf(Error); + expect(error).not.toBeInstanceOf(D1AmbiguousWriteError); + expect((error as Error).message).toContain("syntax error"); + await db.destroy(); + }); + + it("aborts a timed-out request", async () => { + const fetch = vi.fn( + (_input, init) => + new Promise((_resolve, reject) => { + init?.signal?.addEventListener("abort", () => + reject(new DOMException("Aborted", "AbortError")), + ); + }), + ); + const db = database(fetch, { timeoutMs: 5 }); + + await expect(sql`select 1`.execute(db)).rejects.toThrow(/timed out/i); + expect(fetch).toHaveBeenCalledTimes(1); + await db.destroy(); + }); + + it("aborts an active request when the driver is destroyed", async () => { + let signal: AbortSignal | undefined; + const fetch = vi.fn( + (_input, init) => + new Promise((_resolve, reject) => { + signal = init?.signal ?? undefined; + init?.signal?.addEventListener("abort", () => + reject(new DOMException("Aborted", "AbortError")), + ); + }), + ); + const dialect = new D1RestDialect({ + accountId: ACCOUNT_ID, + databaseId: DATABASE_ID, + token: TOKEN, + fetch, + }); + const driver = dialect.createDriver(); + await driver.init(); + const connection = await driver.acquireConnection(); + const query = connection.executeQuery(CompiledQuery.raw("select 1")); + await vi.waitFor(() => expect(signal).toBeDefined()); + + await driver.destroy(); + await expect(query).rejects.toThrow(/valid response/i); + expect(signal?.aborted).toBe(true); + }); +}); diff --git a/packages/cloudflare/tests/db/hyperdrive-migrations.test.ts b/packages/cloudflare/tests/db/hyperdrive-migrations.test.ts new file mode 100644 index 000000000..49d87781b --- /dev/null +++ b/packages/cloudflare/tests/db/hyperdrive-migrations.test.ts @@ -0,0 +1,64 @@ +import { describe, expect, it } from "vitest"; + +import { createMigrationExecutor } from "../../src/db/hyperdrive-migrations.js"; + +const CONNECTION_ENV = "CLOUDFLARE_HYPERDRIVE_LOCAL_CONNECTION_STRING_PRIMARY_DB"; + +describe("Hyperdrive migration executor", () => { + it("resolves a safe PostgreSQL origin identity without opening a connection", async () => { + const executor = await createMigrationExecutor( + { binding: "PRIMARY_DB", connectionStringEnv: CONNECTION_ENV }, + { + projectRoot: "/project", + env: { + [CONNECTION_ENV]: + "postgresql://migration-user:super-secret@db.example.com:5444/content?sslmode=require&application_name=emdash", + }, + }, + ); + + expect(executor.target).toEqual({ + kind: "postgres", + label: "db.example.com:5444/content", + fingerprint: expect.stringMatching(/^[a-f0-9]{64}$/), + }); + const serialized = JSON.stringify(executor.target); + expect(serialized).not.toContain("migration-user"); + expect(serialized).not.toContain("super-secret"); + expect(serialized).not.toContain("sslmode"); + }); + + it("fails before opening a connection when the direct-origin variable is missing", async () => { + await expect( + createMigrationExecutor( + { binding: "PRIMARY_DB", connectionStringEnv: CONNECTION_ENV }, + { projectRoot: "/project", env: {} }, + ), + ).rejects.toThrow(CONNECTION_ENV); + }); + + it("honors the common database URL environment override", async () => { + const executor = await createMigrationExecutor( + { binding: "PRIMARY_DB", connectionStringEnv: CONNECTION_ENV }, + { + projectRoot: "/project", + env: { OVERRIDE_DATABASE_URL: "postgres://user:secret@override.example.com/cms" }, + overrides: { databaseUrlEnv: "OVERRIDE_DATABASE_URL" }, + }, + ); + + expect(executor.target.label).toBe("override.example.com:5432/cms"); + }); + + it("rejects a non-PostgreSQL direct origin", async () => { + await expect( + createMigrationExecutor( + { binding: "PRIMARY_DB", connectionStringEnv: CONNECTION_ENV }, + { + projectRoot: "/project", + env: { [CONNECTION_ENV]: "mysql://user:secret@db.example.com/content" }, + }, + ), + ).rejects.toThrow(/PostgreSQL migration connection string is invalid/i); + }); +}); diff --git a/packages/cloudflare/tests/do-config.test.ts b/packages/cloudflare/tests/do-config.test.ts index 51ed9c004..a008ce08c 100644 --- a/packages/cloudflare/tests/do-config.test.ts +++ b/packages/cloudflare/tests/do-config.test.ts @@ -7,6 +7,10 @@ describe("d1()", () => { const result = d1({ binding: "DB" }); expect(result.supportsRequestScope).toBe(true); expect(result.supportsCoalescing).toBe(true); + expect(result.migrations).toEqual({ + entrypoint: "@emdash-cms/cloudflare/db/d1-migrations", + manifestConfig: { binding: "DB" }, + }); }); }); diff --git a/packages/cloudflare/tests/fixtures/d1-wrangler.jsonc b/packages/cloudflare/tests/fixtures/d1-wrangler.jsonc new file mode 100644 index 000000000..f0d6371e3 --- /dev/null +++ b/packages/cloudflare/tests/fixtures/d1-wrangler.jsonc @@ -0,0 +1,24 @@ +{ + "name": "d1-resolver-fixture", + "main": "./worker.ts", + "account_id": "0123456789abcdef0123456789abcdef", + "compatibility_date": "2026-08-12", + "d1_databases": [ + { + "binding": "DB", + "database_name": "top-level-db", + "database_id": "11111111-2222-4333-8444-555555555555", + }, + ], + "env": { + "production": { + "d1_databases": [ + { + "binding": "DB", + "database_name": "production-db", + "database_id": "aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee", + }, + ], + }, + }, +} diff --git a/packages/cloudflare/tests/hyperdrive-config.test.ts b/packages/cloudflare/tests/hyperdrive-config.test.ts index 6079fc384..b69000d67 100644 --- a/packages/cloudflare/tests/hyperdrive-config.test.ts +++ b/packages/cloudflare/tests/hyperdrive-config.test.ts @@ -9,6 +9,13 @@ describe("hyperdrive()", () => { entrypoint: "@emdash-cms/cloudflare/db/hyperdrive", config: { binding: "HYPERDRIVE", max: undefined }, type: "postgres", + migrations: { + entrypoint: "@emdash-cms/cloudflare/db/hyperdrive-migrations", + manifestConfig: { + binding: "HYPERDRIVE", + connectionStringEnv: "CLOUDFLARE_HYPERDRIVE_LOCAL_CONNECTION_STRING_HYPERDRIVE", + }, + }, supportsRequestScope: true, }); }); @@ -42,4 +49,47 @@ describe("hyperdrive()", () => { cachedBinding: "HYPERDRIVE_CACHED", }); }); + + it("keeps migration credentials out of runtime config and never selects the cached binding", () => { + const result = hyperdrive({ + binding: "PRIMARY_DB", + cachedBinding: "CACHED_DB", + migrationConnectionStringEnv: "DEPLOYMENT_DATABASE_URL", + }); + + expect(result.config).not.toHaveProperty("migrationConnectionStringEnv"); + expect(result.migrations).toEqual({ + entrypoint: "@emdash-cms/cloudflare/db/hyperdrive-migrations", + manifestConfig: { + binding: "PRIMARY_DB", + connectionStringEnv: "DEPLOYMENT_DATABASE_URL", + }, + }); + expect(JSON.stringify(result.migrations)).not.toContain("CACHED_DB"); + }); + + it("derives the default direct-origin environment variable from the primary binding", () => { + const result = hyperdrive({ binding: "CONTENT_DB" }); + + expect(result.migrations?.manifestConfig).toEqual({ + binding: "CONTENT_DB", + connectionStringEnv: "CLOUDFLARE_HYPERDRIVE_LOCAL_CONNECTION_STRING_CONTENT_DB", + }); + }); + + it("keeps JavaScript bindings containing dollar signs backwards compatible", () => { + const result = hyperdrive({ binding: "$CONTENT_DB" }); + + expect(result.config).toMatchObject({ binding: "$CONTENT_DB" }); + expect(result.migrations?.manifestConfig).toEqual({ + binding: "$CONTENT_DB", + connectionStringEnv: "CLOUDFLARE_HYPERDRIVE_LOCAL_CONNECTION_STRING__CONTENT_DB", + }); + }); + + it("rejects an invalid migration environment variable name", () => { + expect(() => hyperdrive({ migrationConnectionStringEnv: "DEPLOYMENT-DATABASE-URL" })).toThrow( + /valid environment variable name/i, + ); + }); }); diff --git a/packages/cloudflare/tsdown.config.ts b/packages/cloudflare/tsdown.config.ts index 8c8035064..2e2926880 100644 --- a/packages/cloudflare/tsdown.config.ts +++ b/packages/cloudflare/tsdown.config.ts @@ -4,7 +4,9 @@ export default defineConfig({ entry: [ "src/index.ts", "src/db/d1.ts", + "src/db/d1-migrations.ts", "src/db/hyperdrive.ts", + "src/db/hyperdrive-migrations.ts", "src/db/do.ts", "src/db/do-sql.ts", "src/db/playground.ts",