Speed up automatic Media Usage indexing

.changeset/fast-media-usage-backfill.md

@@ -0,0 +1,6 @@
+---
+"emdash": patch
+"@emdash-cms/cloudflare": patch
+---
+
+Speeds up automatic Media Usage indexing by continuing bounded background work until the site catches up.

.changeset/media-usage-progress-locale.md

@@ -0,0 +1,5 @@
+---
+"@emdash-cms/admin": patch
+---
+
+Fixes Media Usage progress text displaying untranslated plural syntax.

demos/cloudflare/src/worker.ts

@@ -7,11 +7,19 @@
  */
 
 import handler from "@astrojs/cloudflare/entrypoints/server";
-import { createScheduledHandler, PluginBridge } from "@emdash-cms/cloudflare/worker";
+import {
+	createMediaUsageQueueHandler,
+	createScheduledHandler,
+	type MediaUsageWakeMessage,
+	PluginBridge,
+} from "@emdash-cms/cloudflare/worker";
 
 export { PluginBridge };
 
+const resolveMediaUsageQueue = (env: Env) => env.MEDIA_USAGE_QUEUE;
+
 export default {
 	...handler,
-	scheduled: createScheduledHandler(),
-} satisfies ExportedHandler<Env>;
+	scheduled: createScheduledHandler({ resolveMediaUsageQueue }),
+	queue: createMediaUsageQueueHandler(resolveMediaUsageQueue),
+} satisfies ExportedHandler<Env, MediaUsageWakeMessage>;

demos/cloudflare/worker-configuration.d.ts

@@ -8,6 +8,7 @@ declare namespace Cloudflare {
 	interface Env {
 		MEDIA: R2Bucket;
 		DB: D1Database;
+		MEDIA_USAGE_QUEUE: Queue;
 		LOADER: WorkerLoader;
 		AI_SEARCH: AiSearchNamespace;
 	}

demos/cloudflare/wrangler.jsonc

@@ -34,7 +34,24 @@
 			"bucket_name": "emdash-media",
 		},
 	],
-	// Cron trigger drives the AI Search reindex queue flush.
+	"queues": {
+		"producers": [
+			{
+				"binding": "MEDIA_USAGE_QUEUE",
+				"queue": "emdash-demo-media-usage",
+			},
+		],
+		"consumers": [
+			{
+				"queue": "emdash-demo-media-usage",
+				"max_batch_size": 1,
+				"max_batch_timeout": 0,
+				"max_retries": 3,
+				"max_concurrency": 1,
+			},
+		],
+	},
+	// Cron triggers drive general maintenance and Media Usage Queue recovery.
 	"triggers": {
 		"crons": ["* * * * *", "*/2 * * * *"],
 	},

docs/src/content/docs/deployment/cloudflare.mdx

@@ -80,32 +80,64 @@ On Cloudflare Workers, scheduled publishing, plugin cron, and maintenance tasks
 
 ```ts title="src/worker.ts"
 import handler, {
+	createMediaUsageQueueHandler,
 	createScheduledHandler,
+	type MediaUsageWakeMessage,
 	PluginBridge,
 } from "@emdash-cms/cloudflare/worker";
 
 export { PluginBridge };
 
+const resolveMediaUsageQueue = (env: Env) => env.MEDIA_USAGE_QUEUE;
+
 export default {
 	...handler,
-	scheduled: createScheduledHandler(),
-} satisfies ExportedHandler;
+	scheduled: createScheduledHandler({ resolveMediaUsageQueue }),
+	queue: createMediaUsageQueueHandler(resolveMediaUsageQueue),
+} satisfies ExportedHandler<Env, MediaUsageWakeMessage>;
 ```
 
-By default, `*/2 * * * *` runs Media Usage maintenance and every other expression runs general maintenance. Then add both Cron Triggers to `wrangler.jsonc`:
+By default, `*/2 * * * *` sends a Media Usage recovery wake and every other expression runs general maintenance. Then add both Cron Triggers and the automatically provisioned Queue to `wrangler.jsonc`:
 
 ```jsonc title="wrangler.jsonc"
 {
 	"triggers": {
 		"crons": ["* * * * *", "*/2 * * * *"],
 	},
+	"queues": {
+		"producers": [
+			{
+				"binding": "MEDIA_USAGE_QUEUE",
+				"queue": "my-emdash-site-media-usage",
+			},
+		],
+		"consumers": [
+			{
+				"queue": "my-emdash-site-media-usage",
+				"max_batch_size": 1,
+				"max_batch_timeout": 0,
+				"max_retries": 3,
+				"max_concurrency": 1,
+			},
+		],
+	},
 }
 ```
 
-To use different schedules, set the corresponding `generalCron` or `mediaUsageCron` option in `createScheduledHandler()` and use the same expression in `wrangler.jsonc`.
+Run `wrangler types` after changing the bindings so `Env.MEDIA_USAGE_QUEUE` is available to
+TypeScript.
+
+The Media Usage Cron sends a recovery wake to this Queue. The consumer processes one bounded,
+durable database unit at a time and sends another wake while work remains. Queue messages contain no
+content identity, and the existing database work rows remain authoritative if a message is retried or
+lost. Concurrency stays at one because D1 processes a database sequentially.
+
+Existing custom Workers without a Queue binding keep the previous bounded Cron path. To use different
+schedules, set the corresponding `generalCron` or `mediaUsageCron` option in
+`createScheduledHandler()` and use the same expression in `wrangler.jsonc`.
 
 <Aside type="caution">
-	Without the general trigger, scheduled publishing and plugin cron do not run. Without the dedicated Media Usage trigger, automatic historical reconciliation cannot progress. Local `astro dev` still uses the in-process scheduler.
+	Without the general trigger, scheduled publishing and plugin cron do not run. Without the dedicated Media Usage trigger, automatic historical reconciliation cannot start or recover after Queue retries are exhausted. Local `astro dev` still uses the in-process scheduler.
 </Aside>
 
 ### Enable automatic media usage indexing

docs/src/content/docs/deployment/nodejs.mdx

@@ -58,6 +58,10 @@ The server runs on `http://localhost:4321` by default. Migrations are applied on
 The built-in scheduler runs only while a Node process is running. It handles scheduled publishing,
 plugin tasks, and background media indexing.
 
+While Media Usage work is available, the scheduler continues one bounded indexing unit on each event
+loop turn until it catches up. It yields between units so requests and other scheduled work remain
+responsive, and durable database work resumes after a process restart.
+
 Keep at least one Node process running continuously in production. If all processes stop or sleep,
 scheduled tasks pause.
 

packages/admin/src/locales/en/messages.po

@@ -203,6 +203,13 @@ msgstr "{0, plural, one {Media file} other {Media files}}"
 msgid "{0, plural, one {User} other {Users}}"
 msgstr "{0, plural, one {User} other {Users}}"
 
+#. placeholder {0}: progress.totalCollections
+#. placeholder {1}: progress.readyCollections
+#. placeholder {2}: progress.readyCollections
+#: packages/admin/src/components/settings/MediaUsageSettings.tsx:374
+msgid "{0, plural, one {{1} of # content type ready} other {{2} of # content types ready}}"
+msgstr "{0, plural, one {{1} of # content type ready} other {{2} of # content types ready}}"
+
 #. placeholder {0}: envLabel(m.key)
 #. placeholder {1}: m.required
 #. placeholder {2}: m.host

packages/admin/src/locales/loadMessages.ts

@@ -7,12 +7,17 @@ const LOCALE_LOADERS = import.meta.glob<{ messages: Messages }>("./**/messages.m
 export async function loadMessages(locale: string): Promise<Messages> {
 	const key = `./${locale}/messages.mjs`;
 	const fallbackKey = `./${DEFAULT_LOCALE}/messages.mjs`;
-	const loader = LOCALE_LOADERS[key] ?? LOCALE_LOADERS[fallbackKey];
-	if (!loader) {
+	const fallbackLoader = LOCALE_LOADERS[fallbackKey];
+	if (!fallbackLoader) {
 		throw new Error(
 			`No locale catalog found for "${locale}" or "${DEFAULT_LOCALE}". Run \`pnpm locale:compile\` to generate catalogs.`,
 		);
 	}
-	const { messages } = await loader();
-	return messages;
+	const loader = LOCALE_LOADERS[key] ?? fallbackLoader;
+	if (loader === fallbackLoader) return (await loader()).messages;
+	const [{ messages: fallbackMessages }, { messages }] = await Promise.all([
+		fallbackLoader(),
+		loader(),
+	]);
+	return { ...fallbackMessages, ...messages };
 }

packages/admin/tests/lib/locales.test.ts

@@ -1,3 +1,4 @@
+import { setupI18n } from "@lingui/core";
 import { describe, expect, test } from "vitest";
 
 import {
@@ -22,6 +23,25 @@ test("loadMessages falls back to English for unknown locale", async () => {
 	expect(fallback).toEqual(english);
 });
 
+test("formats the Media Usage progress plural from the production English catalog", async () => {
+	const catalog = await loadMessages("en");
+	expect(catalog.zRXzWv).toBeDefined();
+	expect(Array.isArray(catalog.zRXzWv)).toBe(true);
+	const germanCatalog = await loadMessages("de");
+	expect(germanCatalog.zRXzWv).toEqual(catalog.zRXzWv);
+	const productionI18n = setupI18n();
+	productionI18n.loadAndActivate({ locale: "en", messages: catalog });
+
+	expect(
+		productionI18n._({
+			id: "zRXzWv",
+			message:
+				"{0, plural, one {{1} of # content type ready} other {{2} of # content types ready}}",
+			values: { 0: 2, 1: 1, 2: 1 },
+		}),
+	).toBe("1 of 2 content types ready");
+});
+
 // -- getLocaleDir ----------------------------------------------------------
 
 describe("getLocaleDir", () => {

packages/cloudflare/src/worker.ts

@@ -14,7 +14,11 @@
 // @ts-ignore - resolved against the consuming app's Astro build
 import astroHandler from "@astrojs/cloudflare/entrypoints/server";
 import { createApp } from "astro/app/entrypoint";
-import { runScheduledMediaUsageTasks, runScheduledTasks } from "emdash/middleware";
+import {
+	runMediaUsageMaintenanceStep,
+	runScheduledMediaUsageTasks,
+	runScheduledTasks,
+} from "emdash/middleware";
 
 export { PluginBridge } from "./sandbox/index.js";
 
@@ -45,16 +49,27 @@ async function invalidatePublishedTags(
  * general maintenance. Configuring a general expression changes that lane
  * from catch-all to exact.
  */
-export interface ScheduledHandlerOptions {
+export interface MediaUsageWakeMessage {
+	version: 1;
+}
+
+export type OptionalMediaUsageQueueResolver<Env> = (
+	env: Env,
+) => Queue<MediaUsageWakeMessage> | undefined;
+
+export type MediaUsageQueueResolver<Env> = (env: Env) => Queue<MediaUsageWakeMessage>;
+
+export interface ScheduledHandlerOptions<Env = unknown> {
 	generalCron?: string;
 	mediaUsageCron?: string;
+	resolveMediaUsageQueue?: OptionalMediaUsageQueueResolver<Env>;
 }
 
 const DEFAULT_MEDIA_USAGE_CRON = "*/2 * * * *";
 
-export function createScheduledHandler(
-	options?: ScheduledHandlerOptions,
-): ExportedHandlerScheduledHandler {
+export function createScheduledHandler<Env = unknown>(
+	options?: ScheduledHandlerOptions<Env>,
+): ExportedHandlerScheduledHandler<Env> {
 	const generalCron = options?.generalCron?.trim();
 	const mediaUsageCron = options?.mediaUsageCron?.trim() ?? DEFAULT_MEDIA_USAGE_CRON;
 	if ((options?.generalCron !== undefined && !generalCron) || !mediaUsageCron) {
@@ -64,8 +79,23 @@ export function createScheduledHandler(
 		throw new Error("General and Media Usage Cron expressions must differ");
 	}
 
-	return (controller, _env, ctx) => {
+	return (controller, env, ctx) => {
 		if (controller.cron === mediaUsageCron) {
+			let queue: Queue<MediaUsageWakeMessage> | undefined;
+			try {
+				queue = options?.resolveMediaUsageQueue?.(env);
+			} catch {
+				console.error("[scheduled] Failed to queue Media Usage maintenance wake");
+				return;
+			}
+			if (queue) {
+				ctx.waitUntil(
+					queue.send({ version: 1 }).catch(() => {
+						console.error("[scheduled] Failed to queue Media Usage maintenance wake");
+					}),
+				);
+				return;
+			}
 			ctx.waitUntil(
 				runScheduledMediaUsageTasks().catch((error: unknown) => {
 					console.error("[scheduled] Media Usage maintenance failed:", error);
@@ -97,6 +127,38 @@ export function createScheduledHandler(
 	};
 }
 
+export function createMediaUsageQueueHandler<Env>(
+	resolveMediaUsageQueue: MediaUsageQueueResolver<Env>,
+): ExportedHandlerQueueHandler<Env, MediaUsageWakeMessage> {
+	return async (batch, env) => {
+		let hasValidWake = false;
+		for (const message of batch.messages) {
+			if (isMediaUsageWakeMessage(message.body)) {
+				hasValidWake = true;
+			} else {
+				message.ack();
+				console.warn("[queue] Ignoring invalid Media Usage wake");
+			}
+		}
+		if (!hasValidWake) return;
+
+		const queue = resolveMediaUsageQueue(env);
+		if (!queue) throw new Error("Media Usage Queue binding is unavailable");
+
+		const result = await runMediaUsageMaintenanceStep();
+		if (result.continuation.kind === "none") return;
+		if (result.continuation.kind === "delayed") {
+			await queue.send({ version: 1 }, { delaySeconds: result.continuation.delaySeconds });
+			return;
+		}
+		await queue.send({ version: 1 });
+	};
+}
+
+function isMediaUsageWakeMessage(value: unknown): value is MediaUsageWakeMessage {
+	return typeof value === "object" && value !== null && "version" in value && value.version === 1;
+}
+
 // 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;