fix(scheduling): drive scheduled publishing from a real heartbeat

.changeset/scheduled-publishing-driver.md

@@ -0,0 +1,33 @@
+---
+"emdash": minor
+"@emdash-cms/cloudflare": minor
+---
+
+Drive scheduled publishing from a real heartbeat instead of request side effects (#1303).
+
+Content scheduled via the admin now actually transitions to `published` when its time arrives. Previously nothing promoted the row — `status` stayed `scheduled` and `published_at` stayed null forever.
+
+A new sweep (`publishDueContent`) promotes due content and runs alongside the existing cron tick and system cleanup:
+
+- **Node / single-process:** the timer-based scheduler already drives it — no action needed.
+- **Cloudflare Workers:** a `scheduled()` handler driven by a Cron Trigger now runs the sweep. The request-driven `PiggybackScheduler` is gone, so there are no maintenance side effects on visitor requests.
+
+`@emdash-cms/cloudflare` ships a Worker entry that wraps Astro's handler with the `scheduled()` handler (`@emdash-cms/cloudflare/worker`, plus `createScheduledHandler()` for hand-assembled Workers). When a cache provider is configured, the handler also purges edge-cache tags for whatever it published, so stale snapshots produced before the scheduled time are evicted.
+
+**Migration for existing Cloudflare sites.** New sites get this from the templates. Existing deployments must update two files:
+
+```ts
+// src/worker.ts
+export { default, PluginBridge } from "@emdash-cms/cloudflare/worker";
+```
+
+```jsonc
+// wrangler.jsonc
+"triggers": { "crons": ["* * * * *"] }
+```
+
+Without the Cron Trigger, scheduled publishing and plugin cron do not run on Workers.
+
+Scheduled publishing matches manual publishing exactly: it fires `content:afterPublish` hooks (search indexing, webhooks, syndication), and records the _scheduled_ time as `published_at` on first publication rather than the (later) sweep time. The sweep claims each row atomically before promoting it, so an entry unscheduled or rescheduled just before its time is never published, and overlapping sweeps can't double-publish. Local `astro dev` keeps running the timer-driven sweep even under the Cloudflare adapter (where production relies on the Cron Trigger).
+
+Each tick promotes at most 100 items per collection (a large backlog drains over successive ticks) so a single Worker invocation can't exhaust its CPU/subrequest budget, and edge-cache tags are purged incrementally after each collection's batch rather than only at the end. On Node, the maintenance interval is capped at 60s so scheduled-publish latency matches the Cloudflare Cron Trigger cadence instead of lagging up to five minutes when no plugin cron is due.

packages/cloudflare/package.json

@@ -41,6 +41,10 @@
       "types": "./dist/sandbox/index.d.mts",
       "default": "./dist/sandbox/index.mjs"
     },
+    "./worker": {
+      "types": "./dist/worker.d.mts",
+      "default": "./dist/worker.mjs"
+    },
     "./plugins": {
       "types": "./dist/plugins/index.d.mts",
       "default": "./dist/plugins/index.mjs"
@@ -77,12 +81,14 @@
     "ulidx": "^2.4.1"
   },
   "peerDependencies": {
+    "@astrojs/cloudflare": ">=12.0.0",
     "@cloudflare/workers-types": ">=4.0.0",
     "astro": ">=6.0.0-beta.0",
     "kysely": ">=0.28.17"
   },
   "devDependencies": {
     "@arethetypeswrong/cli": "catalog:",
+    "@astrojs/cloudflare": "catalog:",
     "@cloudflare/workers-types": "catalog:",
     "publint": "catalog:",
     "tsdown": "catalog:",

packages/cloudflare/src/worker.ts

@@ -0,0 +1,84 @@
+/**
+ * Cloudflare Worker entry for EmDash sites.
+ *
+ * Wraps the Astro Cloudflare server handler with a `scheduled()` handler so a
+ * Cron Trigger drives scheduled publishing, plugin cron, and system cleanup
+ * without any request side effects. Re-exports the `PluginBridge` Durable
+ * Object so the sandbox binding resolves against the entry module.
+ *
+ * Templates use this as their entire `src/worker.ts`:
+ *
+ *   export { default, PluginBridge } from "@emdash-cms/cloudflare/worker";
+ *
+ * and add a Cron Trigger to wrangler.jsonc:
+ *
+ *   "triggers": { "crons": ["* * * * *"] }
+ *
+ * The `@astrojs/cloudflare/entrypoints/server` import is resolved by the
+ * consuming app's Astro build (it pulls the build-time `virtual:astro:app`
+ * module), so this package keeps the adapter external.
+ */
+
+// @ts-ignore - resolved against the consuming app's Astro build
+import astroHandler from "@astrojs/cloudflare/entrypoints/server";
+import { createApp } from "astro/app/entrypoint";
+import { runScheduledTasks } from "emdash/middleware";
+
+export { PluginBridge } from "./sandbox/index.js";
+
+// The Astro App wraps the build manifest; reuse one per isolate so each tick
+// doesn't re-resolve the cache provider.
+let app: ReturnType<typeof createApp> | null = null;
+
+/**
+ * Purge edge-cache tags for content the sweep just published. Without a
+ * request there's no `locals.cache`, so we reach the configured cache provider
+ * through the Astro App pipeline — the same provider routes invalidate against.
+ * A no-op when no cache provider is configured.
+ */
+async function invalidatePublishedTags(
+	published: ReadonlyArray<{ collection: string; id: string }>,
+): Promise<void> {
+	if (published.length === 0) return;
+	app ??= createApp();
+	const provider = await app.pipeline.getCacheProvider();
+	if (!provider) return;
+	const tags = [...new Set(published.flatMap((ref) => [ref.collection, ref.id]))];
+	await provider.invalidate({ tags });
+}
+
+/**
+ * Build a Worker `scheduled()` handler that runs EmDash's scheduled
+ * maintenance batch and purges edge-cache tags for anything it published.
+ * Exported for sites that assemble their own Worker object; most sites get it
+ * via this module's default export.
+ */
+export function createScheduledHandler(): ExportedHandlerScheduledHandler {
+	return (_controller, _env, ctx) => {
+		ctx.waitUntil(
+			// Invalidate incrementally as each collection batch publishes, so a
+			// scheduled() invocation killed mid-sweep (CPU/wall-clock limits on a
+			// large backlog) still purged the cache tags for everything it managed
+			// to publish — not just whatever completed before a single end-of-sweep
+			// purge that may never run.
+			runScheduledTasks({ onPublished: invalidatePublishedTags })
+				.then(({ published }) => {
+					if (published.length > 0) {
+						console.log(`[scheduled] Published ${published.length} scheduled item(s)`);
+					}
+					return undefined;
+				})
+				.catch((error: unknown) => {
+					console.error("[scheduled] runScheduledTasks failed:", error);
+				}),
+		);
+	};
+}
+
+// eslint-disable-next-line typescript/no-unsafe-type-assertion -- astroHandler is the adapter's { fetch } worker object; resolved at app-build time
+const handler = astroHandler as ExportedHandler;
+
+export default {
+	...handler,
+	scheduled: createScheduledHandler(),
+} satisfies ExportedHandler;

packages/cloudflare/tsdown.config.ts

@@ -10,6 +10,7 @@ export default defineConfig({
 		"src/storage/r2.ts",
 		"src/auth/index.ts",
 		"src/sandbox/index.ts",
+		"src/worker.ts",
 		"src/plugins/index.ts",
 		// Media provider runtimes
 		"src/media/images-runtime.ts",
@@ -21,5 +22,9 @@ export default defineConfig({
 	format: ["esm"],
 	dts: true,
 	clean: true,
-	external: ["cloudflare:workers", "cloudflare:email"],
+	// @astrojs/cloudflare's server entrypoint and `astro/app/entrypoint` both
+	// resolve the build-time `virtual:astro:app` module — only available in the
+	// consuming app's Astro build, never here. Keep them external so the bare
+	// imports survive to be resolved downstream.
+	external: ["cloudflare:workers", "cloudflare:email", /^@astrojs\/cloudflare/, /^astro($|\/)/],
 });

packages/core/src/api/handlers/content.ts

@@ -14,6 +14,7 @@ import { RevisionRepository } from "../../database/repositories/revision.js";
 import { SeoRepository } from "../../database/repositories/seo.js";
 import {
 	EmDashValidationError,
+	ScheduledNotDueError,
 	InvalidCursorError,
 	type BylineSummary,
 	type ContentBylineCredit,
@@ -1268,13 +1269,13 @@ export async function handleContentPublish(
 	db: Kysely<Database>,
 	collection: string,
 	id: string,
-	options: { publishedAt?: string } = {},
+	options: { publishedAt?: string; requireScheduledDue?: boolean } = {},
 ): Promise<ApiResult<ContentResponse>> {
 	try {
 		const item = await withTransaction(db, async (trx) => {
 			const repo = new ContentRepository(trx);
 			const resolvedId = (await resolveId(repo, collection, id)) ?? id;
-			return repo.publish(collection, resolvedId, options.publishedAt);
+			return repo.publish(collection, resolvedId, options.publishedAt, options.requireScheduledDue);
 		});
 
 		const hasSeo = await collectionHasSeo(db, collection);
@@ -1285,6 +1286,17 @@ export async function handleContentPublish(
 			data: { item },
 		};
 	} catch (error) {
+		// The scheduled sweep gates publish on the row still being due; a row
+		// unscheduled in the meantime is a silent skip, not a failure.
+		if (error instanceof ScheduledNotDueError) {
+			return {
+				success: false,
+				error: {
+					code: "NOT_DUE",
+					message: error.message,
+				},
+			};
+		}
 		if (error instanceof EmDashValidationError) {
 			return {
 				success: false,

packages/core/src/astro/integration/virtual-modules.ts

@@ -63,6 +63,9 @@ export const RESOLVED_VIRTUAL_SEED_ID = "\0" + VIRTUAL_SEED_ID;
 export const VIRTUAL_WAIT_UNTIL_ID = "virtual:emdash/wait-until";
 export const RESOLVED_VIRTUAL_WAIT_UNTIL_ID = "\0" + VIRTUAL_WAIT_UNTIL_ID;
 
+export const VIRTUAL_SCHEDULER_ID = "virtual:emdash/scheduler";
+export const RESOLVED_VIRTUAL_SCHEDULER_ID = "\0" + VIRTUAL_SCHEDULER_ID;
+
 /**
  * Generates the config virtual module.
  */
@@ -413,6 +416,42 @@ export function generateWaitUntilModule(adapterName: string | undefined): string
 	return `export const waitUntil = undefined;`;
 }
 
+/**
+ * Generates the scheduler virtual module.
+ *
+ * Decides — at build time, from the Astro adapter — whether the runtime gets a
+ * long-lived timer heartbeat. A *production* Cloudflare build has no persistent
+ * timers, so the Worker's `scheduled()` handler (a Cron Trigger) drives
+ * `runScheduledTasks()` instead and this exports `null`. Every other case — any
+ * other adapter (Node, Bun), and crucially local `astro dev` even under the
+ * Cloudflare adapter (no Cron Trigger fires in dev) — gets a `NodeCronScheduler`
+ * factory so plugin cron, scheduled publishing, and cleanup still run.
+ *
+ * Keeping the adapter check here — rather than in core's runtime — means the
+ * runtime has no Cloudflare-specific code path; it just calls `createScheduler`
+ * if one was injected. Mirrors the wait-until module's approach.
+ */
+export function generateSchedulerModule(
+	adapterName: string | undefined,
+	command: "build" | "serve" | undefined,
+): string {
+	// Only suppress the timer for an actual Cloudflare *build* — that artifact
+	// runs in workerd where a Cron Trigger drives scheduled work. In `serve`
+	// (local dev) nothing fires the Cron Trigger, so fall through to the timer.
+	if (adapterName === "@astrojs/cloudflare" && command !== "serve") {
+		return `// Serverless build: an external Cron Trigger drives scheduled work.
+export const createScheduler = null;
+`;
+	}
+	return `// Long-lived runtime (or local dev): drive scheduled work from an in-process timer.
+import { NodeCronScheduler } from "emdash";
+
+export function createScheduler(executor) {
+	return new NodeCronScheduler(executor);
+}
+`;
+}
+
 /**
  * Generates the seed virtual module.
  * Reads the user's seed file at build time (in Node context) and embeds it,

packages/core/src/astro/integration/vite-config.ts

@@ -42,8 +42,11 @@ import {
 	RESOLVED_VIRTUAL_SEED_ID,
 	VIRTUAL_WAIT_UNTIL_ID,
 	RESOLVED_VIRTUAL_WAIT_UNTIL_ID,
+	VIRTUAL_SCHEDULER_ID,
+	RESOLVED_VIRTUAL_SCHEDULER_ID,
 	generateSeedModule,
 	generateWaitUntilModule,
+	generateSchedulerModule,
 	generateConfigModule,
 	generateDialectModule,
 	generateStorageModule,
@@ -203,6 +206,9 @@ export function createVirtualModulesPlugin(options: VitePluginOptions): Plugin {
 			if (id === VIRTUAL_WAIT_UNTIL_ID) {
 				return RESOLVED_VIRTUAL_WAIT_UNTIL_ID;
 			}
+			if (id === VIRTUAL_SCHEDULER_ID) {
+				return RESOLVED_VIRTUAL_SCHEDULER_ID;
+			}
 		},
 		load(id: string) {
 			if (id === RESOLVED_VIRTUAL_CONFIG_ID) {
@@ -271,6 +277,12 @@ export function createVirtualModulesPlugin(options: VitePluginOptions): Plugin {
 			if (id === RESOLVED_VIRTUAL_WAIT_UNTIL_ID) {
 				return generateWaitUntilModule(astroConfig.adapter?.name);
 			}
+			// Generate scheduler module — a NodeCronScheduler factory on
+			// long-lived runtimes, or null under the Cloudflare adapter where
+			// a Cron Trigger drives scheduled work instead.
+			if (id === RESOLVED_VIRTUAL_SCHEDULER_ID) {
+				return generateSchedulerModule(astroConfig.adapter?.name, viteCommand);
+			}
 		},
 	};
 }

packages/core/src/astro/middleware.ts

@@ -25,6 +25,8 @@ import * as virtualSandboxRunnerModule from "virtual:emdash/sandbox-runner";
 // @ts-ignore - virtual module
 import { sandboxedPlugins as virtualSandboxedPlugins } from "virtual:emdash/sandboxed-plugins";
 // @ts-ignore - virtual module
+import { createScheduler as virtualCreateScheduler } from "virtual:emdash/scheduler";
+// @ts-ignore - virtual module
 import { createStorage as virtualCreateStorage } from "virtual:emdash/storage";
 
 import { after } from "../after.js";
@@ -39,6 +41,7 @@ import {
 	type RuntimeDependencies,
 	type SandboxedPluginEntry,
 	type MediaProviderEntry,
+	type CreateSchedulerFn,
 } from "../emdash-runtime.js";
 import { setI18nConfig } from "../i18n/config.js";
 import type { Database, Storage } from "../index.js";
@@ -52,6 +55,7 @@ import {
 	type RequestMetrics,
 	runWithContext,
 } from "../request-context.js";
+import type { PublishedRef } from "../scheduled-publish.js";
 import { isMissingTableError } from "../utils/db-errors.js";
 import { createInitLock, type InitLock, initWithLock } from "../utils/init-lock.js";
 import type { EmDashConfig } from "./integration/runtime.js";
@@ -174,6 +178,7 @@ function buildDependencies(config: EmDashConfig): RuntimeDependencies {
 		plugins: getPlugins(),
 		createDialect: virtualCreateDialect as (config: Record<string, unknown>) => unknown,
 		createStorage: virtualCreateStorage as ((config: Record<string, unknown>) => Storage) | null,
+		createScheduler: virtualCreateScheduler as CreateSchedulerFn | null,
 		sandboxEnabled: sandboxModule.sandboxEnabled as boolean,
 		sandboxBypassed: (sandboxModule.sandboxBypassed as boolean) ?? false,
 		sandboxedPluginEntries: (virtualSandboxedPlugins as SandboxedPluginEntry[]) || [],
@@ -243,6 +248,29 @@ async function getRuntime(
 	);
 }
 
+/**
+ * Run scheduled maintenance (cron tasks, scheduled publishing, system cleanup)
+ * outside any request. Resolves the runtime from the build-time virtual config
+ * and the cached singleton — the same instance request handlers use.
+ *
+ * Wired into a platform heartbeat that is not a request: the Cloudflare Worker's
+ * `scheduled()` handler (Cron Trigger) calls this. On Node the runtime's own
+ * timer-based scheduler already drives the same work, so this isn't needed there.
+ *
+ * Returns the content promoted by the publishing sweep so the caller can purge
+ * edge-cache tags for it. `onPublished` (optional) is awaited after each
+ * collection's batch so the caller can invalidate edge-cache tags incrementally
+ * rather than only after the whole sweep.
+ */
+export async function runScheduledTasks(
+	options: { onPublished?: (refs: PublishedRef[]) => Promise<void> } = {},
+): Promise<{ published: PublishedRef[] }> {
+	const config = getConfig();
+	if (!config) return { published: [] };
+	const runtime = await getRuntime(config);
+	return runtime.runScheduledTasks(options);
+}
+
 /**
  * Astro attaches AstroCookies to outgoing responses via a well-known global
  * symbol. Cloning a Response (`new Response(body, init)`) drops non-header

packages/core/src/astro/types.ts

@@ -313,7 +313,7 @@ export interface EmDashHandlers {
 	handleContentPublish: (
 		collection: string,
 		id: string,
-		options?: { publishedAt?: string },
+		options?: { publishedAt?: string; requireScheduledDue?: boolean },
 	) => Promise<HandlerResponse>;
 
 	handleContentUnpublish: (collection: string, id: string) => Promise<HandlerResponse>;