diff --git a/.changeset/event-scoped-db-cron-plugins.md b/.changeset/event-scoped-db-cron-plugins.md new file mode 100644 index 0000000000..bf6cb45ed3 --- /dev/null +++ b/.changeset/event-scoped-db-cron-plugins.md @@ -0,0 +1,5 @@ +--- +"emdash": patch +--- + +Resolves the database connection at use-time for the cron sweep, plugin hook contexts, and media providers instead of capturing it once at startup. This makes scheduled publishing, plugin cron, and database-querying plugin hooks work on connection-backed adapters like Postgres over Cloudflare Hyperdrive, where a connection is bound to the event that opened it. Stateless adapters (D1, Node SQLite) are unaffected. diff --git a/.changeset/hyperdrive-postgres-adapter.md b/.changeset/hyperdrive-postgres-adapter.md index 2ace641fbf..44f6677633 100644 --- a/.changeset/hyperdrive-postgres-adapter.md +++ b/.changeset/hyperdrive-postgres-adapter.md @@ -4,4 +4,4 @@ Adds a `hyperdrive()` database adapter for connecting EmDash on Cloudflare Workers to a PostgreSQL (or PostgreSQL-compatible, e.g. PlanetScale Postgres) database through a Hyperdrive binding. Configure it with `database: hyperdrive({ binding: "HYPERDRIVE" })`. Each request gets its own pooled connection that is opened and closed within that request — connections cannot be reused across Worker requests. Requires `pg >= 8.16.3`, the `nodejs_compat` compatibility flag, and a compatibility date of `2024-09-23` or later. Disable Hyperdrive query caching for the configuration so the admin's read-after-write stays consistent. -The content read/write path (pages, content API routes, loaders) is fully supported. Cron Triggers (scheduled publishing, plugin cron, system cleanup), plugin hooks that query the database, and sandboxed plugins are not yet supported on this adapter — they use a per-isolate connection that workerd will not reuse across events. Use `d1()` if your deployment depends on those. +The content read/write path, scheduled publishing, plugin cron, and database-querying plugin hooks are all supported. Sandboxed plugins remain D1-only (the sandbox bridge talks to a D1 binding directly, independent of the configured adapter). diff --git a/packages/cloudflare/src/db/hyperdrive.ts b/packages/cloudflare/src/db/hyperdrive.ts index 042995a339..468419f98b 100644 --- a/packages/cloudflare/src/db/hyperdrive.ts +++ b/packages/cloudflare/src/db/hyperdrive.ts @@ -20,33 +20,20 @@ * ALS, and the runtime/loader db getters prefer it over the singleton — so all * request-path queries use a connection opened in the current request. * - * `createDialect` still builds the per-isolate singleton Kysely. Its socket is - * opened by whatever event first queries it — normally the cold-start - * migrations during the first HTTP request — and, because a pg socket is bound - * to the request that opened it, it is only safe to use again from within that - * same event. The request path never does: routes and loaders read through the - * per-request scoped Kysely (ALS), not the singleton. + * `createDialect` still builds the per-isolate singleton Kysely, used only for + * cold-start migrations (which run inside the first request, so the socket is + * valid there). Everything else resolves the connection from ALS at use-time: + * the request path through the runtime/loader db getters, and the background and + * plugin paths (Cron Trigger sweep, plugin hook contexts, media providers) + * through resolvers threaded by the core runtime. The Cron Trigger handler opens + * its own event-scoped connection for the sweep. So no warm-isolate path reuses + * the singleton's request-bound socket across events. * - * Known limitation — background and plugin paths still use the singleton - * -------------------------------------------------------------------------- - * Several subsystems capture the runtime's singleton db at construction and do - * not consult the per-request scoped connection: - * - the Cron Trigger handler (`scheduled()` → scheduled publishing, plugin - * cron, system cleanup), - * - plugin hook contexts (a hook's `content` / `media` / `users` / `cron` - * access), - * - media providers and sandboxed plugins. - * - * On a warm isolate the singleton's socket belongs to an earlier request, so - * these paths can fail under workerd's cross-request I/O guard ("Cannot perform - * I/O on behalf of a different request"). It is not a data-corruption risk — the - * work errors and is logged — but it means scheduled publishing and - * database-querying plugin hooks are not yet supported on the Hyperdrive - * adapter. The core read/write path (pages, content API routes, loaders) is - * unaffected. Closing this requires the core runtime to thread an event-scoped - * connection through those subsystems; tracked in - * https://github.com/emdash-cms/emdash/issues/1622. Until then, use D1 for - * deployments that rely on Cron Triggers or DB-querying plugins. + * Known limitation — sandboxed plugins are D1-only. The sandbox plugin bridge + * (a Durable Object) talks to a D1 binding directly, independent of the + * configured adapter, so sandboxed plugins are not available on a Hyperdrive + * deployment. This is a pre-existing bridge constraint, unrelated to connection + * scoping; tracked in https://github.com/emdash-cms/emdash/issues/1623. * * This module imports directly from cloudflare:workers to access the binding. * Do NOT import it at config time — use { hyperdrive } from @@ -108,10 +95,10 @@ function createPool(connectionString: string, max: number): Pool { /** * Create a PostgreSQL dialect backed by a Hyperdrive binding. * - * Used for the per-isolate singleton Kysely. The request path never touches it - * (it reads through `createRequestScopedDb`); in practice the singleton serves - * cold-start migrations, plus the background/plugin paths noted in the module - * header that are not yet safe across event boundaries on this adapter. + * Used for the per-isolate singleton Kysely, which serves cold-start migrations + * only. The request path reads through `createRequestScopedDb`, and the + * background/plugin paths resolve an event-scoped connection from ALS, so + * neither reuses this singleton's request-bound socket across events. */ export function createDialect(config: HyperdriveConfig): Dialect { const binding = requireBinding(config); diff --git a/packages/cloudflare/src/index.ts b/packages/cloudflare/src/index.ts index 3bf51adc63..32b9dc0a53 100644 --- a/packages/cloudflare/src/index.ts +++ b/packages/cloudflare/src/index.ts @@ -262,21 +262,16 @@ export function d1(config: D1Config): DatabaseDescriptor { * { "placement": { "region": "aws:us-east-1" } } * ``` * - * **Known limitation — request path only (for now).** Each request gets its own - * pg connection, so the content read/write path (pages, content API routes, - * loaders) is fully supported. But several background and plugin paths still use - * the per-isolate singleton connection, whose socket is bound to the request - * that opened it; on a warm isolate workerd refuses to reuse it from a later - * event. Until the core runtime threads an event-scoped connection through them - * (tracked in https://github.com/emdash-cms/emdash/issues/1622), the following - * are **not yet supported** on the Hyperdrive adapter: - * - Cron Triggers — scheduled publishing, plugin cron, and system cleanup. - * - Plugin hooks that query the database via their plugin context. - * - Media providers and sandboxed plugins that hold the singleton db. - * - * Use `d1()` for deployments that depend on those. (This is a Hyperdrive-adapter - * limitation, not a data-safety risk: affected work errors and is logged rather - * than corrupting anything.) + * Each request gets its own pg connection, and the Cron Trigger sweep, plugin + * hook contexts, and media providers resolve an event-scoped connection too, so + * the content read/write path, scheduled publishing, plugin cron, and + * DB-querying plugin hooks are all supported. + * + * **Known limitation — sandboxed plugins are D1-only.** The sandbox plugin + * bridge talks to a D1 binding directly (independent of the configured + * adapter), so sandboxed plugins aren't available on a Hyperdrive deployment. + * This is a pre-existing bridge constraint, unrelated to connection scoping; + * tracked in https://github.com/emdash-cms/emdash/issues/1623. * * @example * ```ts diff --git a/packages/core/src/astro/middleware.ts b/packages/core/src/astro/middleware.ts index 80f407db50..167ca58dee 100644 --- a/packages/core/src/astro/middleware.ts +++ b/packages/core/src/astro/middleware.ts @@ -276,9 +276,72 @@ export async function runScheduledTasks( const config = getConfig(); if (!config) return { published: [] }; const runtime = await getRuntime(config); - return runtime.runScheduledTasks(options); + + // Connection-backed adapters (e.g. Postgres over Hyperdrive) cannot reuse + // the per-isolate singleton from a Cron Trigger: its socket belongs to the + // request that opened it, and workerd rejects cross-event I/O. Open an + // event-scoped connection for the sweep and run the batch under it in ALS — + // the runtime's db getter, the cron executor, and plugin cron contexts all + // resolve the connection from ALS — then close it. Gated on the adapter + // being connection-backed (it exposes `close()`); stateless adapters (D1, + // Node SQLite) return null or a close-less scope and keep using the + // singleton, so their cron path is unchanged. + const scoped = createRequestScopedDb({ + config: config.database?.config, + isAuthenticated: false, + // The sweep publishes and cleans up — a write workload — so a + // connection-backed adapter routes it to the primary. + isWrite: true, + cookies: NOOP_COOKIE_JAR, + url: CRON_EVENT_URL, + }); + if (!scoped?.close) { + // Stateless adapter (or no per-request scoping): the singleton is safe + // outside a request. Any close-less scope created above is discarded. + return runtime.runScheduledTasks(options); + } + + const parent = getRequestContext(); + const ctx = parent + ? { ...parent, db: scoped.db } + : { editMode: false, db: scoped.db, metrics: createRequestMetrics(performance.now()) }; + try { + return await runWithContext(ctx, () => runtime.runScheduledTasks(options)); + } finally { + // Guard both so a throw in teardown can't mask the sweep result or skip + // close() and leak the connection. Mirrors closeSafely() in scoped-db.ts. + try { + scoped.commit(); + } catch (error) { + console.error("[scheduled] request-scoped db commit failed:", error); + } + try { + scoped.close(); + } catch (error) { + console.error("[scheduled] request-scoped db close failed:", error); + } + } } +/** + * A cookie jar that reads nothing and writes nothing, for request-scoped db + * adapters invoked outside an HTTP request (the Cron Trigger sweep). Connection + * adapters like Hyperdrive ignore cookies entirely; the D1 session adapter + * reads/writes a bookmark cookie, but cron never reaches that path (it has no + * `close()`), so the no-ops are never observed. + */ +const NOOP_COOKIE_JAR = { + get: () => undefined, + set: () => {}, +}; + +/** + * Synthetic URL for the cron sweep's request-scoped db opts. Only the D1 + * session adapter inspects `url` (for cookie `secure`), and cron doesn't take + * that path, so the value is never used — it exists to satisfy the contract. + */ +const CRON_EVENT_URL = new URL("https://cron.emdash.internal/"); + /** * Baseline security headers applied to all responses. * Admin routes get additional headers (strict CSP) from auth middleware. diff --git a/packages/core/src/emdash-runtime.ts b/packages/core/src/emdash-runtime.ts index 48f2c50905..1bf91dd137 100644 --- a/packages/core/src/emdash-runtime.ts +++ b/packages/core/src/emdash-runtime.ts @@ -244,6 +244,15 @@ export interface MediaProviderEntry { */ export interface MediaProviderContext { db: Kysely; + /** + * Resolver for the live connection, preferred over `db` by providers that + * query EmDash's database. Resolves the current request/event-scoped + * connection from ALS so connection-backed adapters (Postgres over + * Hyperdrive) don't reuse the per-isolate singleton's socket across events. + * Providers should resolve per operation rather than capturing `db` once. + * Omitted-safe: falls back to `db` for stateless adapters (D1, Node SQLite). + */ + getDb?: () => Kysely; storage: Storage | null; } @@ -333,6 +342,7 @@ export interface EmDashRuntimeParts { allPipelinePlugins: ResolvedPlugin[]; pipelineFactoryOptions: { db: Kysely; + getDb?: () => Kysely; storage?: Storage; siteInfo?: { siteName?: string; siteUrl?: string; locale?: string }; }; @@ -446,7 +456,20 @@ export class EmDashRuntime { readonly configuredPlugins: ResolvedPlugin[]; readonly sandboxedPlugins: Map; readonly sandboxedPluginEntries: SandboxedPluginEntry[]; - readonly schemaRegistry: SchemaRegistry; + /** + * Schema registry bound to the current request/event-scoped connection. + * Built per access (SchemaRegistry just wraps a db) against `this.db`, the + * ALS-aware getter — never a captured snapshot of the singleton. On a + * connection-backed adapter (Postgres over Hyperdrive) a captured singleton + * would query a socket opened by an earlier event and trip workerd's + * cross-request I/O guard; the catch in handlers like handleContentUpdate + * would then silently treat a revision-enabled collection as non-revisioned + * and write draft edits to live columns. Same reasoning as the per-call + * registry in _buildManifest(). + */ + get schemaRegistry(): SchemaRegistry { + return new SchemaRegistry(this.db); + } private _hooks!: HookPipeline; readonly config: EmDashConfig; readonly mediaProviders: Map; @@ -477,6 +500,7 @@ export class EmDashRuntime { /** Factory options for the hook pipeline context factory */ private pipelineFactoryOptions: { db: Kysely; + getDb?: () => Kysely; storage?: Storage; siteInfo?: { siteName?: string; siteUrl?: string; locale?: string }; }; @@ -507,7 +531,6 @@ export class EmDashRuntime { this.configuredPlugins = parts.configuredPlugins; this.sandboxedPlugins = parts.sandboxedPlugins; this.sandboxedPluginEntries = parts.sandboxedPluginEntries; - this.schemaRegistry = new SchemaRegistry(parts.db); this._hooks = parts.hooks; this.enabledPlugins = parts.enabledPlugins; this.pluginStates = parts.pluginStates; @@ -653,7 +676,9 @@ export class EmDashRuntime { // The old pipeline's contextFactoryOptions were built up incrementally // via setContextFactory calls during create(). We replay them here. if (this.email) { - newPipeline.setContextFactory({ db: this.db, emailPipeline: this.email }); + // db/getDb are already wired by createHookPipeline above (they live in + // pipelineFactoryOptions), so the merge only adds emailPipeline. + newPipeline.setContextFactory({ emailPipeline: this.email }); } if (this.cronScheduler) { const scheduler = this.cronScheduler; @@ -1017,6 +1042,22 @@ export class EmDashRuntime { // Initialize database (connects, runs migrations if needed) const db = await phase("rt.db", "DB init + migrations", () => EmDashRuntime.getDatabase(deps)); + // Resolver for the live connection, mirroring the `get db()` getter + // below (which can't be used here — the runtime instance doesn't exist + // yet). Long-lived subsystems built during create() (cron executor, + // plugin context factory, media providers) capture this resolver rather + // than the `db` snapshot, so a connection-backed adapter (Postgres over + // Hyperdrive) serves their queries from the current request/event-scoped + // connection in ALS instead of the per-isolate singleton — whose socket + // belongs to an earlier request and would trip workerd's cross-request + // I/O guard. Stateless adapters (D1, Node SQLite) set no ALS db on most + // paths, so this falls back to the singleton: unchanged behavior. + const resolveDb = (): Kysely => { + const ctx = getRequestContext(); + // eslint-disable-next-line typescript/no-unsafe-type-assertion -- ALS db is typed unknown to avoid a circular import; middleware always sets a Kysely + return (ctx?.db as Kysely | undefined) ?? db; + }; + // Validate EMDASH_ENCRYPTION_KEY once here so a malformed value // surfaces in startup logs instead of as request-time 500s. The key // itself is not yet consumed (a follow-up PR adds plugin-secret @@ -1295,9 +1336,15 @@ export class EmDashRuntime { // Filter to currently enabled plugins for the initial pipeline const enabledPluginList = allPipelinePlugins.filter((p) => enabledPlugins.has(p.id)); - // Create hook pipeline + // Create hook pipeline. getDb travels here (not just via the email + // setContextFactory call below) so it survives rebuildHookPipeline(), + // which reconstructs the factory from pipelineFactoryOptions. Without it, + // toggling a plugin on an email-less deployment would silently revert + // plugin contexts to the singleton db — re-breaking connection-backed + // adapters. See #1622. const pipelineFactoryOptions = { db, + getDb: resolveDb, storage: storage ?? undefined, siteInfo, }; @@ -1349,7 +1396,7 @@ export class EmDashRuntime { // Initialize media providers const mediaProviders = new Map(); const mediaProviderEntries = deps.mediaProviderEntries ?? []; - const providerContext: MediaProviderContext = { db, storage }; + const providerContext: MediaProviderContext = { db, storage, getDb: resolveDb }; for (const entry of mediaProviderEntries) { try { @@ -1390,8 +1437,10 @@ export class EmDashRuntime { }; // Wire email pipeline into context factory (independent of cron — - // must not be inside the cron try/catch or ctx.email breaks when cron fails) - pipeline.setContextFactory({ db, emailPipeline }); + // must not be inside the cron try/catch or ctx.email breaks when cron fails). + // db/getDb were already set via pipelineFactoryOptions above; merge only + // adds emailPipeline. + pipeline.setContextFactory({ emailPipeline }); let cronExecutor: CronExecutor | null = null; let cronScheduler: CronScheduler | null = null; @@ -1403,7 +1452,7 @@ export class EmDashRuntime { await phase("rt.cron", "Cron init (recovery deferred post-response)", async () => { try { - cronExecutor = new CronExecutor(db, invokeCronHook); + cronExecutor = new CronExecutor(resolveDb, invokeCronHook); // Recover stale locks from previous crashes. Pure bookkeeping // against the _emdash_cron_tasks table — no request needs the diff --git a/packages/core/src/media/local-runtime.ts b/packages/core/src/media/local-runtime.ts index d257f0be04..baca6586a8 100644 --- a/packages/core/src/media/local-runtime.ts +++ b/packages/core/src/media/local-runtime.ts @@ -29,6 +29,14 @@ export interface LocalMediaRuntimeConfig { enabled?: boolean; // These are injected by the runtime, not from user config db?: Kysely; + /** + * Resolver for the live connection, preferred over `db`. The runtime + * injects it so a connection-backed adapter (Postgres over Hyperdrive) + * serves provider queries from the current request-scoped connection in ALS + * rather than a snapshot of the per-isolate singleton. Omitted for stateless + * adapters (D1, Node SQLite), where `db` is used directly. + */ + getDb?: () => Kysely; storage?: Storage; } @@ -42,11 +50,15 @@ export const createMediaProvider: CreateMediaProviderFn throw new Error("Local media provider requires database connection"); } - const repo = new MediaRepository(db); + // Resolve the connection per operation (not captured once) so a + // connection-backed adapter uses the current event-scoped connection; falls + // back to the injected `db` for stateless adapters. + const resolveDb = config.getDb ?? (() => db); + const repo = () => new MediaRepository(resolveDb()); const provider: MediaProvider = { async list(options: MediaListOptions) { - const result = await repo.findMany({ + const result = await repo().findMany({ cursor: options.cursor, limit: options.limit, mimeType: options.mimeType, @@ -75,7 +87,7 @@ export const createMediaProvider: CreateMediaProviderFn }, async get(id: string) { - const item = await repo.findById(id); + const item = await repo().findById(id); if (!item) return null; return { @@ -108,7 +120,8 @@ export const createMediaProvider: CreateMediaProviderFn }, async delete(id: string) { - const item = await repo.findById(id); + const repoInstance = repo(); + const item = await repoInstance.findById(id); if (!item) return; // Delete from storage if available @@ -120,7 +133,7 @@ export const createMediaProvider: CreateMediaProviderFn } } - await repo.delete(id); + await repoInstance.delete(id); // If this row was referenced by `logo`, `favicon`, or // `seo.defaultOgImage`, the worker-scoped settings cache now diff --git a/packages/core/src/plugins/context.ts b/packages/core/src/plugins/context.ts index 036eeb7577..533a87bffa 100644 --- a/packages/core/src/plugins/context.ts +++ b/packages/core/src/plugins/context.ts @@ -868,6 +868,16 @@ export function createUserAccess(db: Kysely): UserAccess { export interface PluginContextFactoryOptions { db: Kysely; + /** + * Resolver for the database connection, preferred over `db` when present. + * Called per `createContext()` so connection-backed adapters (e.g. Postgres + * over Hyperdrive) get the current request/event-scoped connection from ALS + * rather than a snapshot of the per-isolate singleton — reusing the + * singleton's socket from a later event trips workerd's cross-request I/O + * guard. When omitted, `db` is used directly (correct for stateless + * adapters like D1 and Node SQLite). `db` remains required as the fallback. + */ + getDb?: () => Kysely; /** * Storage backend for direct media uploads. * If not provided, upload() will throw. @@ -911,8 +921,7 @@ export interface PluginContextFactoryOptions { * Factory for creating plugin contexts */ export class PluginContextFactory { - private optionsRepo: OptionsRepository; - private db: Kysely; + private resolveDb: () => Kysely; private storage?: Storage; private getUploadUrl?: ( filename: string, @@ -930,8 +939,8 @@ export class PluginContextFactory { private warnedMissingMediaBackend = new Set(); constructor(options: PluginContextFactoryOptions) { - this.db = options.db; - this.optionsRepo = new OptionsRepository(options.db); + const fixedDb = options.db; + this.resolveDb = options.getDb ?? (() => fixedDb); this.storage = options.storage; this.getUploadUrl = options.getUploadUrl; this.site = createSiteInfo(options.siteInfo ?? {}); @@ -946,10 +955,17 @@ export class PluginContextFactory { createContext(plugin: ResolvedPlugin): PluginContext { const capabilities = new Set(plugin.capabilities); + // Resolve the connection once per context. For stateless adapters this + // is the singleton; for connection-backed adapters it's the current + // request/event-scoped connection from ALS. All repos below are built + // from this local `db` so a hook never queries a stale singleton socket. + const db = this.resolveDb(); + const optionsRepo = new OptionsRepository(db); + // Always available - const kv = createKVAccess(this.optionsRepo, plugin.id); + const kv = createKVAccess(optionsRepo, plugin.id); const log = createLogAccess(plugin.id); - const storage = createStorageAccess(this.db, plugin.id, plugin.storage); + const storage = createStorageAccess(db, plugin.id, plugin.storage); // Capability-gated: content // Note: capabilities reach this point already normalized to the @@ -957,9 +973,9 @@ export class PluginContextFactory { // names ("read:content", "write:content") never appear here. let content: ContentAccess | ContentAccessWithWrite | undefined; if (capabilities.has("content:write")) { - content = createContentAccessWithWrite(this.db); + content = createContentAccessWithWrite(db); } else if (capabilities.has("content:read")) { - content = createContentAccess(this.db); + content = createContentAccess(db); } // Capability-gated: media @@ -970,7 +986,7 @@ export class PluginContextFactory { let media: MediaAccess | MediaAccessWithWrite | undefined; if (capabilities.has("media:write")) { if (this.getUploadUrl || this.storage) { - media = createMediaAccessWithWrite(this.db, this.getUploadUrl, this.storage); + media = createMediaAccessWithWrite(db, this.getUploadUrl, this.storage); } else { if (!this.warnedMissingMediaBackend.has(plugin.id)) { this.warnedMissingMediaBackend.add(plugin.id); @@ -979,11 +995,11 @@ export class PluginContextFactory { ); } if (capabilities.has("media:read")) { - media = createMediaAccess(this.db); + media = createMediaAccess(db); } } } else if (capabilities.has("media:read")) { - media = createMediaAccess(this.db); + media = createMediaAccess(db); } // Capability-gated: http @@ -997,14 +1013,14 @@ export class PluginContextFactory { // Capability-gated: users let users: UserAccess | undefined; if (capabilities.has("users:read")) { - users = createUserAccess(this.db); + users = createUserAccess(db); } - // Cron access ��� always available (scoped to plugin), but only if + // Cron access — always available (scoped to plugin), but only if // the runtime provided a reschedule callback (i.e. cron is wired up). let cron: CronAccess | undefined; if (this.cronReschedule) { - cron = new CronAccessImpl(this.db, plugin.id, this.cronReschedule); + cron = new CronAccessImpl(db, plugin.id, this.cronReschedule); } // Email access — requires email:send capability AND a configured provider diff --git a/packages/core/src/plugins/cron.ts b/packages/core/src/plugins/cron.ts index c44b407266..ed3d19fc32 100644 --- a/packages/core/src/plugins/cron.ts +++ b/packages/core/src/plugins/cron.ts @@ -39,10 +39,26 @@ export type RescheduleFn = () => void; * Stateless — all state lives in the database. */ export class CronExecutor { + /** + * Resolves the database connection to use for this tick. A resolver (not a + * captured instance) so connection-backed adapters work across events: on + * Cloudflare the `scheduled()` handler installs an event-scoped connection + * in ALS, and this resolves to it instead of the per-isolate singleton + * whose socket belongs to an earlier request. Accepts a plain `Kysely` too + * (wrapped in a constant resolver) for callers/tests that don't need ALS. + */ + private readonly resolveDb: () => Kysely; + constructor( - private db: Kysely, + db: Kysely | (() => Kysely), private invokeCronHook: InvokeCronHookFn, - ) {} + ) { + this.resolveDb = typeof db === "function" ? db : () => db; + } + + private get db(): Kysely { + return this.resolveDb(); + } /** * Process all overdue tasks. diff --git a/packages/core/tests/integration/plugins/event-scoped-db.test.ts b/packages/core/tests/integration/plugins/event-scoped-db.test.ts new file mode 100644 index 0000000000..caeff9c036 --- /dev/null +++ b/packages/core/tests/integration/plugins/event-scoped-db.test.ts @@ -0,0 +1,326 @@ +/** + * Event-scoped DB resolution (#1622) + * + * Long-lived subsystems built once at runtime init — the plugin context + * factory, the cron executor, and media providers — must resolve the database + * connection at use-time, not capture it at construction. On connection-backed + * adapters (Postgres over Hyperdrive) the per-isolate singleton's socket is + * bound to the request that opened it, so a later request or the Cron Trigger + * must use the current event-scoped connection instead. + * + * These tests prove resolution happens per operation by pointing a resolver at + * two independent databases and asserting each operation reads/writes the one + * the resolver currently returns. They use plain SQLite (stateless across + * events) where the resolver simply makes the indirection observable; the same + * indirection is what lets Hyperdrive swap the ALS-scoped connection in. + */ + +import { randomUUID } from "node:crypto"; + +import Database from "better-sqlite3"; +import { Kysely, SqliteDialect } from "kysely"; +import { describe, it, expect, beforeEach, afterEach, vi } from "vitest"; + +import { runMigrations } from "../../../src/database/migrations/runner.js"; +import { MediaRepository } from "../../../src/database/repositories/media.js"; +import type { Database as DbSchema } from "../../../src/database/types.js"; +import { EmDashRuntime, type RuntimeDependencies } from "../../../src/emdash-runtime.js"; +import { createMediaProvider } from "../../../src/media/local-runtime.js"; +import { PluginContextFactory } from "../../../src/plugins/context.js"; +import { CronExecutor } from "../../../src/plugins/cron.js"; +import { createHookPipeline } from "../../../src/plugins/hooks.js"; +import type { CronHandler, ResolvedHook, ResolvedPlugin } from "../../../src/plugins/types.js"; +import { runWithContext } from "../../../src/request-context.js"; +import { SchemaRegistry } from "../../../src/schema/registry.js"; + +function createTestPlugin(overrides: Partial = {}): ResolvedPlugin { + return { + id: "test-plugin", + version: "1.0.0", + capabilities: [], + allowedHosts: [], + storage: {}, + admin: { pages: [], widgets: [], fieldWidgets: {} }, + hooks: {}, + routes: {}, + settings: undefined, + ...overrides, + }; +} + +/** A plugin whose cron hook writes a marker through ctx.kv (always available). */ +function cronMarkerPlugin(id: string): ResolvedPlugin { + const cron: ResolvedHook = { + priority: 0, + timeout: 5_000, + dependencies: [], + errorPolicy: "continue", + exclusive: false, + pluginId: id, + handler: async (_event, ctx) => { + await ctx.kv.set("ran", "yes"); + }, + }; + return createTestPlugin({ id, hooks: { cron } }); +} + +/** Read a raw option value (the kv store) from a db. */ +async function optionValue(db: Kysely, name: string): Promise { + const row = await db + .selectFrom("options") + .select("value") + .where("name", "=", name) + .executeTakeFirst(); + return row?.value; +} + +async function makeDb(): Promise> { + const db = new Kysely({ + dialect: new SqliteDialect({ database: new Database(":memory:") }), + }); + await runMigrations(db); + return db; +} + +/** Insert a single due one-shot cron task into the given db. */ +async function seedDueOneshot(db: Kysely, pluginId: string, taskName: string) { + const past = new Date(Date.now() - 60_000).toISOString(); + await db + .insertInto("_emdash_cron_tasks" as never) + .values({ + id: `task_${taskName}`, + plugin_id: pluginId, + task_name: taskName, + schedule: past, + is_oneshot: 1, + data: null, + status: "idle", + enabled: 1, + next_run_at: past, + locked_at: null, + last_run_at: null, + created_at: past, + } as never) + .execute(); +} + +describe("event-scoped DB resolution (#1622)", () => { + let dbA: Kysely; + let dbB: Kysely; + + beforeEach(async () => { + dbA = await makeDb(); + dbB = await makeDb(); + }); + + afterEach(async () => { + await dbA.destroy(); + await dbB.destroy(); + }); + + describe("PluginContextFactory", () => { + it("resolves the connection per createContext() via getDb", async () => { + let current = dbA; + const getDb = vi.fn(() => current); + // `db` is the required fallback; getDb takes precedence. + const factory = new PluginContextFactory({ db: dbA, getDb }); + const plugin = createTestPlugin({ id: "kv-plugin" }); + + // Write through a context resolved to dbB. + current = dbB; + const ctxWrite = factory.createContext(plugin); + await ctxWrite.kv.set("color", "blue"); + + // A context resolved to dbA must not see it (different connection). + current = dbA; + const ctxA = factory.createContext(plugin); + expect(await ctxA.kv.get("color")).toBeNull(); + + // Back to dbB: the value is there. Proves each createContext built its + // repos from the currently-resolved connection, not a snapshot. + current = dbB; + const ctxB = factory.createContext(plugin); + expect(await ctxB.kv.get("color")).toBe("blue"); + + // Resolver consulted once per createContext call. + expect(getDb).toHaveBeenCalledTimes(3); + }); + + it("falls back to the fixed db when getDb is omitted (stateless adapters)", async () => { + const factory = new PluginContextFactory({ db: dbA }); + const plugin = createTestPlugin({ id: "kv-plugin" }); + + const ctx = factory.createContext(plugin); + await ctx.kv.set("k", "v"); + + // Written to dbA, the fixed connection. + const repoOptions = await dbA + .selectFrom("options") + .select("value") + .where("name", "=", "plugin:kv-plugin:k") + .executeTakeFirst(); + expect(repoOptions?.value).toBe(JSON.stringify("v")); + }); + }); + + describe("CronExecutor", () => { + it("resolves the connection at tick time, not construction", async () => { + let current = dbA; + const getDb = vi.fn(() => current); + const invoked: string[] = []; + const executor = new CronExecutor(getDb, async (pluginId) => { + invoked.push(pluginId); + }); + + // A due task exists only in dbB. + await seedDueOneshot(dbB, "cron-plugin", "sweep"); + + // Resolver points at dbA (empty): nothing to process. + current = dbA; + expect(await executor.tick()).toBe(0); + expect(invoked).toEqual([]); + + // Repoint at dbB: the same executor now processes the due task. + current = dbB; + expect(await executor.tick()).toBe(1); + expect(invoked).toEqual(["cron-plugin"]); + + // One-shot was consumed from dbB. + const remaining = await dbB + .selectFrom("_emdash_cron_tasks" as never) + .selectAll() + .execute(); + expect(remaining).toHaveLength(0); + expect(getDb).toHaveBeenCalled(); + }); + + it("accepts a plain Kysely for callers that don't need ALS", async () => { + await seedDueOneshot(dbA, "cron-plugin", "sweep"); + const executor = new CronExecutor(dbA, async () => {}); + expect(await executor.tick()).toBe(1); + }); + }); + + describe("hook pipeline rebuild (#1622 regression)", () => { + // rebuildHookPipeline() reconstructs the pipeline from + // pipelineFactoryOptions. getDb must live in those options (not only in + // the conditional email setContextFactory call), or a plugin toggle on an + // email-less deployment would silently revert plugin contexts to the + // singleton db. + it("resolves via getDb when the pipeline is built from factory options", async () => { + let current = dbA; + const pipeline = createHookPipeline([cronMarkerPlugin("cron-hook")], { + db: dbA, + getDb: () => current, + }); + + current = dbB; + const res = await pipeline.invokeCronHook("cron-hook", { name: "t" }); + expect(res.success).toBe(true); + + // Wrote to dbB (the resolved connection), not dbA (the fixed fallback). + expect(await optionValue(dbB, "plugin:cron-hook:ran")).toBe(JSON.stringify("yes")); + expect(await optionValue(dbA, "plugin:cron-hook:ran")).toBeUndefined(); + }); + + it("keeps getDb across a setContextFactory merge (cron/email wiring)", async () => { + let current = dbA; + const pipeline = createHookPipeline([cronMarkerPlugin("cron-hook")], { + db: dbA, + getDb: () => current, + }); + // A partial merge like the rebuild's email / cron-reschedule calls must + // not drop the previously-set getDb. + pipeline.setContextFactory({ cronReschedule: () => {} }); + + current = dbB; + await pipeline.invokeCronHook("cron-hook", { name: "t" }); + expect(await optionValue(dbB, "plugin:cron-hook:ran")).toBe(JSON.stringify("yes")); + }); + }); + + describe("local media provider", () => { + it("resolves the connection per operation via getDb", async () => { + let current = dbA; + const getDb = vi.fn(() => current); + const provider = createMediaProvider({ db: dbA, getDb }); + + // Seed a media row into dbB only. + await new MediaRepository(dbB).create({ + filename: "photo.jpg", + mimeType: "image/jpeg", + storageKey: "media/photo.jpg", + }); + + // Resolver at dbA: empty. + current = dbA; + expect((await provider.list({})).items).toHaveLength(0); + + // Resolver at dbB: the row is visible without rebuilding the provider. + current = dbB; + const listed = await provider.list({}); + expect(listed.items).toHaveLength(1); + expect(listed.items[0]!.filename).toBe("photo.jpg"); + + expect(getDb).toHaveBeenCalled(); + }); + }); + + describe("runtime schemaRegistry (#1622 regression)", () => { + function createDeps(): RuntimeDependencies { + return { + config: { + database: { + entrypoint: `test-schema-registry-${randomUUID()}`, + config: {}, + type: "sqlite", + }, + }, + plugins: [], + createDialect: () => new SqliteDialect({ database: new Database(":memory:") }), + createStorage: null, + sandboxEnabled: false, + sandboxedPluginEntries: [], + createSandboxRunner: null, + }; + } + + it("resolves against the event-scoped db, not the captured singleton", async () => { + // Regression: the runtime used to capture `new SchemaRegistry(parts.db)` + // at construction. On a connection-backed adapter that singleton's + // socket belongs to an earlier event; handleContentUpdate's catch would + // then treat a revision-enabled collection as non-revisioned and write + // draft edits to live columns. The registry must resolve `this.db`. + const runtime = await EmDashRuntime.create(createDeps()); + try { + // A separate event-scoped db with a revision-enabled collection the + // runtime's singleton does not have. + const scopedDb = new Kysely({ + dialect: new SqliteDialect({ database: new Database(":memory:") }), + }); + await runMigrations(scopedDb); + await new SchemaRegistry(scopedDb).createCollection({ + slug: "widgets", + label: "Widgets", + supports: ["drafts", "revisions"], + }); + + try { + // No ALS context: the registry resolves the singleton, which lacks it. + expect(await runtime.schemaRegistry.getCollectionWithFields("widgets")).toBeNull(); + + // Under the scoped db: the registry resolves it (and sees revisions). + const found = await runWithContext({ editMode: false, db: scopedDb }, () => + runtime.schemaRegistry.getCollectionWithFields("widgets"), + ); + expect(found?.slug).toBe("widgets"); + expect(found?.supports).toContain("revisions"); + } finally { + await scopedDb.destroy(); + } + } finally { + await runtime.stopCron(); + } + }); + }); +});