From 851e082cc95ffc831f4e1ab2c06dff2e5aa3017a Mon Sep 17 00:00:00 2001 From: khoinguyenpham04 <137921741+khoinguyenpham04@users.noreply.github.com> Date: Mon, 17 Aug 2026 17:18:47 +0100 Subject: [PATCH 1/4] feat(core): process one due media usage unit --- packages/core/src/astro/middleware.ts | 14 ++ packages/core/src/emdash-runtime.ts | 18 ++- .../usage/collection-deletion-processor.ts | 1 - .../src/media/usage/maintenance-engine.ts | 108 +++++++++++++++ .../media/usage/reconciliation-processor.ts | 41 ++++-- ...-usage-reconciliation-finalization.test.ts | 15 ++- .../media-usage-scheduled-driver.test.ts | 126 ++++++++++++++++++ 7 files changed, 308 insertions(+), 15 deletions(-) create mode 100644 packages/core/src/media/usage/maintenance-engine.ts diff --git a/packages/core/src/astro/middleware.ts b/packages/core/src/astro/middleware.ts index eab7aa1245..b66bc9b4e9 100644 --- a/packages/core/src/astro/middleware.ts +++ b/packages/core/src/astro/middleware.ts @@ -44,6 +44,7 @@ import { DB_INIT_DEADLINE_MS, EmDashRuntime, type MediaUsageMaintenanceResult, + type MediaUsageMaintenanceStepResult, type RuntimeDependencies, type SandboxedPluginEntry, type MediaProviderEntry, @@ -295,6 +296,19 @@ export async function runScheduledMediaUsageTasks(): Promise runtime.runScheduledMediaUsageTasks()); } +export async function runMediaUsageMaintenanceStep(): Promise { + const config = getConfig(); + if (!config) { + return { + state: "inactive", + continuation: { kind: "none" }, + taskClass: null, + turn: null, + }; + } + return runOutsideRequest(config, (runtime) => runtime.runMediaUsageMaintenanceStep()); +} + /** * Run a callback against the EmDash runtime outside any HTTP request — from a * Cloudflare Queue consumer, a `scheduled()` handler, or any other diff --git a/packages/core/src/emdash-runtime.ts b/packages/core/src/emdash-runtime.ts index ce4dd347cb..38b521ce89 100644 --- a/packages/core/src/emdash-runtime.ts +++ b/packages/core/src/emdash-runtime.ts @@ -52,6 +52,11 @@ import { markContentMediaUsageCollectionStale, refreshContentMediaUsageAfterWrite, } from "./media/usage/content-refresh.js"; +import { + runMediaUsageMaintenanceStep, + type MediaUsageMaintenanceStepResult, + type MediaUsageMaintenanceTaskClass, +} from "./media/usage/maintenance-engine.js"; import { MEDIA_USAGE_RECONCILIATION_LIMITS, processDueMediaUsageReconciliation, @@ -552,10 +557,11 @@ export const MEDIA_USAGE_MAINTENANCE_QUERY_RESERVATIONS = Object.freeze({ eventCeiling: 40, }); -export type MediaUsageMaintenanceTaskClass = - | "entry_work" - | "collection_deletion" - | "reconciliation"; +export type { + MediaUsageMaintenanceContinuation, + MediaUsageMaintenanceStepResult, + MediaUsageMaintenanceTaskClass, +} from "./media/usage/maintenance-engine.js"; export type MediaUsageMaintenanceResult = | { outcome: "inactive" | "admission_closed"; taskClass: null; turn: null } @@ -804,6 +810,10 @@ export class EmDashRuntime { return runScheduledMediaUsageLane(this.db); } + async runMediaUsageMaintenanceStep(): Promise { + return runMediaUsageMaintenanceStep(this.db); + } + /** * Materialize plugin-declared storage indexes, once per process. * diff --git a/packages/core/src/media/usage/collection-deletion-processor.ts b/packages/core/src/media/usage/collection-deletion-processor.ts index f0adb7ce98..342671a217 100644 --- a/packages/core/src/media/usage/collection-deletion-processor.ts +++ b/packages/core/src/media/usage/collection-deletion-processor.ts @@ -12,7 +12,6 @@ import { export const MEDIA_USAGE_COLLECTION_DELETION_LIMITS = Object.freeze({ candidatesPerTick: 4, - deletionsPerTick: 1, rowsPerBatch: 50, leaseDurationSeconds: 5 * 60, maxAttempts: 5, diff --git a/packages/core/src/media/usage/maintenance-engine.ts b/packages/core/src/media/usage/maintenance-engine.ts new file mode 100644 index 0000000000..618dbbd354 --- /dev/null +++ b/packages/core/src/media/usage/maintenance-engine.ts @@ -0,0 +1,108 @@ +import { sql, type Kysely } from "kysely"; + +import type { Database } from "../../database/types.js"; +import { processDueMediaUsageCollectionDeletions } from "./collection-deletion-processor.js"; +import { processDueMediaUsageReconciliationDetailed } from "./reconciliation-processor.js"; +import { processDueMediaUsageWork } from "./work-processor.js"; + +export type MediaUsageMaintenanceTaskClass = + | "entry_work" + | "collection_deletion" + | "reconciliation"; + +export type MediaUsageMaintenanceContinuation = + | { kind: "none" } + | { kind: "immediate" } + | { kind: "delayed"; delaySeconds: 30 }; + +export interface MediaUsageMaintenanceStepResult { + state: "inactive" | "idle" | "blocked" | "progress"; + continuation: MediaUsageMaintenanceContinuation; + taskClass: MediaUsageMaintenanceTaskClass | null; + turn: number | null; +} + +const TASK_CLASSES: readonly MediaUsageMaintenanceTaskClass[] = [ + "entry_work", + "collection_deletion", + "reconciliation", +]; + +export async function runMediaUsageMaintenanceStep( + db: Kysely, +): Promise { + const activation = await db + .updateTable("_emdash_media_usage_activation") + .set({ + media_usage_maintenance_turn: sql`(media_usage_maintenance_turn + 1) % 3`, + }) + .where("task_key", "=", "incremental_capture") + .where("state", "=", "active") + .returning("media_usage_maintenance_turn") + .executeTakeFirst(); + if (!activation) return inactiveResult(); + + const startingTurn = activation.media_usage_maintenance_turn; + let firstBlocked: { taskClass: MediaUsageMaintenanceTaskClass; turn: number } | null = null; + + for (let offset = 0; offset < TASK_CLASSES.length; offset++) { + const turn = (startingTurn + offset) % TASK_CLASSES.length; + const taskClass = TASK_CLASSES[turn]; + const outcome = await runTaskClass(db, taskClass); + if (outcome === "inactive") return inactiveResult(); + if (outcome === "progress") { + return { + state: "progress", + continuation: { kind: "immediate" }, + taskClass, + turn, + }; + } + if (outcome === "blocked" && !firstBlocked) firstBlocked = { taskClass, turn }; + } + + if (firstBlocked) { + return { + state: "blocked", + continuation: { kind: "delayed", delaySeconds: 30 }, + ...firstBlocked, + }; + } + + return { + state: "idle", + continuation: { kind: "none" }, + taskClass: TASK_CLASSES[startingTurn], + turn: startingTurn, + }; +} + +async function runTaskClass( + db: Kysely, + taskClass: MediaUsageMaintenanceTaskClass, +): Promise<"inactive" | "idle" | "blocked" | "progress"> { + if (taskClass === "entry_work") { + const result = await processDueMediaUsageWork(db); + if (result.claimedCount > 0) return "progress"; + return result.candidateCount > 0 ? "blocked" : "idle"; + } + if (taskClass === "collection_deletion") { + const result = await processDueMediaUsageCollectionDeletions(db); + if (result.claimedCount > 0) return "progress"; + return result.candidateCount > 0 ? "blocked" : "idle"; + } + + const result = await processDueMediaUsageReconciliationDetailed(db); + if (result.outcome === "inactive") return "inactive"; + if (result.consumedUnit) return "progress"; + return result.outcome === "claim_lost" ? "blocked" : "idle"; +} + +function inactiveResult(): MediaUsageMaintenanceStepResult { + return { + state: "inactive", + continuation: { kind: "none" }, + taskClass: null, + turn: null, + }; +} diff --git a/packages/core/src/media/usage/reconciliation-processor.ts b/packages/core/src/media/usage/reconciliation-processor.ts index 3bf944f437..fad6b67deb 100644 --- a/packages/core/src/media/usage/reconciliation-processor.ts +++ b/packages/core/src/media/usage/reconciliation-processor.ts @@ -34,6 +34,11 @@ export type MediaUsageReconciliationOutcome = | "retry" | "failed"; +export interface MediaUsageReconciliationDetailedResult { + outcome: MediaUsageReconciliationOutcome; + consumedUnit: boolean; +} + export type MediaUsageReconciliationScanOutcome = | "advanced" | "exhausted" @@ -43,24 +48,34 @@ export type MediaUsageReconciliationScanOutcome = export async function processDueMediaUsageReconciliation( db: Kysely, ): Promise { + return (await processDueMediaUsageReconciliationDetailed(db)).outcome; +} + +export async function processDueMediaUsageReconciliationDetailed( + db: Kysely, +): Promise { const activation = await db .selectFrom("_emdash_media_usage_activation") .select("state") .where("task_key", "=", "incremental_capture") .executeTakeFirst(); - if (activation?.state !== "active") return "inactive"; + if (activation?.state !== "active") return { outcome: "inactive", consumedUnit: false }; const reconciliation = new MediaUsageReconciliationRepository(db); - if (await reconciliation.deleteOneObsolete()) return "completed"; + if (await reconciliation.deleteOneObsolete()) { + return { outcome: "completed", consumedUnit: true }; + } const [failed] = await reconciliation.findFailed(1); if (failed) { if (await reconciliation.finishFailedCoverage(failed.collectionId, failed.runToken)) { - return "failed"; + return { outcome: "failed", consumedUnit: true }; + } + if (await reconciliation.resetFailedForNewEpoch(failed)) { + return { outcome: "advanced", consumedUnit: true }; } - if (await reconciliation.resetFailedForNewEpoch(failed)) return "advanced"; } - await reconciliation.seedNextCandidate(); + const seeded = await reconciliation.seedNextCandidate(); const candidates = await reconciliation.findDue( MEDIA_USAGE_RECONCILIATION_LIMITS.candidatesPerTick, ); @@ -73,10 +88,18 @@ export async function processDueMediaUsageReconciliation( }); if (claim) break; } - if (!claim) return candidates.length === 0 ? "not_due" : "claim_lost"; + if (!claim) { + return { + outcome: candidates.length === 0 ? "not_due" : "claim_lost", + consumedUnit: seeded, + }; + } try { - return await processClaimedReconciliation(db, claim); + return { + outcome: await processClaimedReconciliation(db, claim), + consumedUnit: true, + }; } catch (error) { const terminal = claim.attemptCount + 1 >= MEDIA_USAGE_RECONCILIATION_LIMITS.maxAttempts; const recorded = await reconciliation.recordFailure({ @@ -87,10 +110,10 @@ export async function processDueMediaUsageReconciliation( retryDelaySeconds: retryDelaySeconds(claim.attemptCount), terminal, }); - if (!recorded) return "claim_lost"; + if (!recorded) return { outcome: "claim_lost", consumedUnit: true }; if (terminal) await reconciliation.finishFailedCoverage(claim.collectionId, claim.runToken); console.error("[media-usage:reconciliation] Processing failed:", error); - return terminal ? "failed" : "retry"; + return { outcome: terminal ? "failed" : "retry", consumedUnit: true }; } } diff --git a/packages/core/tests/integration/database/media-usage-reconciliation-finalization.test.ts b/packages/core/tests/integration/database/media-usage-reconciliation-finalization.test.ts index c3ae649d77..bd14464f61 100644 --- a/packages/core/tests/integration/database/media-usage-reconciliation-finalization.test.ts +++ b/packages/core/tests/integration/database/media-usage-reconciliation-finalization.test.ts @@ -4,7 +4,10 @@ import { afterEach, beforeEach, expect, it } from "vitest"; import { MediaUsageRepository } from "../../../src/database/repositories/media-usage.js"; import { processDueMediaUsageCollectionDeletions } from "../../../src/media/usage/collection-deletion-processor.js"; import { loadContentMediaUsageSnapshots } from "../../../src/media/usage/content-snapshots.js"; -import { processDueMediaUsageReconciliation } from "../../../src/media/usage/reconciliation-processor.js"; +import { + processDueMediaUsageReconciliation, + processDueMediaUsageReconciliationDetailed, +} from "../../../src/media/usage/reconciliation-processor.js"; import { MediaUsageReconciliationRepository } from "../../../src/media/usage/reconciliation.js"; import { buildContentMediaUsageSourceKey } from "../../../src/media/usage/source-key.js"; import { processDueMediaUsageWork } from "../../../src/media/usage/work-processor.js"; @@ -103,6 +106,16 @@ describeEachDialect("media usage reconciliation finalization", (dialect) => { expect(await usage.findSource(quarantinedSourceKey)).not.toBeNull(); }); + it("reports a seeded or claimed reconciliation unit as consumed", async () => { + const collection = await createCollection(ctx, "consumed_reconciliation", true); + await activateCollection(ctx, collection); + + await expect(processDueMediaUsageReconciliationDetailed(ctx.db)).resolves.toMatchObject({ + consumedUnit: true, + outcome: "advanced", + }); + }); + it("preserves automatic ownership until failed work reaches terminal coverage", async () => { const collection = await createCollection(ctx, "articles", true); await sql` diff --git a/packages/core/tests/integration/runtime/media-usage-scheduled-driver.test.ts b/packages/core/tests/integration/runtime/media-usage-scheduled-driver.test.ts index f7d6e89322..787482f018 100644 --- a/packages/core/tests/integration/runtime/media-usage-scheduled-driver.test.ts +++ b/packages/core/tests/integration/runtime/media-usage-scheduled-driver.test.ts @@ -43,6 +43,132 @@ describe("media usage scheduled drivers", () => { ).not.toBeNull(); }); + it("skips idle maintenance classes and processes due entry work", async () => { + runtime = await EmDashRuntime.create(createDeps(null)); + const fixture = await activateCollection(runtime, "work_conserving_posts"); + await insertEntry(runtime, fixture.tableName, "entry-1"); + await runtime.db + .updateTable("_emdash_media_usage_activation") + .set({ media_usage_maintenance_turn: 0 }) + .where("task_key", "=", "incremental_capture") + .execute(); + + await expect(runtime.runMediaUsageMaintenanceStep()).resolves.toEqual({ + state: "progress", + continuation: { kind: "immediate" }, + taskClass: "entry_work", + turn: 0, + }); + expect(await countWork(runtime)).toBe(0); + }); + + it("processes at most one useful maintenance unit per step", async () => { + runtime = await EmDashRuntime.create(createDeps(null)); + const fixture = await activateCollection(runtime, "one_unit_posts"); + await insertEntry(runtime, fixture.tableName, "entry-1"); + await insertEntry(runtime, fixture.tableName, "entry-2"); + + await expect(runtime.runMediaUsageMaintenanceStep()).resolves.toMatchObject({ + state: "progress", + continuation: { kind: "immediate" }, + taskClass: "entry_work", + }); + expect(await countWork(runtime)).toBe(1); + }); + + it("delays one continuation when every visible claim is blocked", async () => { + runtime = await EmDashRuntime.create(createDeps(null)); + const fixture = await activateCollection(runtime, "blocked_claim_posts"); + await insertEntry(runtime, fixture.tableName, "entry-1"); + await sql` + CREATE TRIGGER block_media_usage_work_claim + BEFORE UPDATE OF state ON _emdash_media_usage_work + WHEN NEW.state = 'leased' + BEGIN + SELECT RAISE(IGNORE); + END + `.execute(runtime.db); + + await expect(runtime.runMediaUsageMaintenanceStep()).resolves.toEqual({ + state: "blocked", + continuation: { kind: "delayed", delaySeconds: 30 }, + taskClass: "entry_work", + turn: 0, + }); + expect(await countWork(runtime)).toBe(1); + }); + + it("does not fall through after seeding reconciliation when its claim is lost", async () => { + runtime = await EmDashRuntime.create(createDeps(null)); + const work = await activateCollection(runtime, "seed_claim_work"); + await insertEntry(runtime, work.tableName, "entry-1"); + const reconciliation = await activateCollection(runtime, "seed_claim_reconciliation"); + await runtime.db + .updateTable("_emdash_media_usage_index_status") + .set({ status: "stale", reconciliation_required: 1 }) + .where("collection_id", "=", reconciliation.collectionId) + .execute(); + await runtime.db + .updateTable("_emdash_media_usage_activation") + .set({ media_usage_maintenance_turn: 1 }) + .where("task_key", "=", "incremental_capture") + .execute(); + await sql` + CREATE TRIGGER block_media_usage_reconciliation_claim + BEFORE UPDATE OF state ON _emdash_media_usage_reconciliations + WHEN NEW.state = 'leased' + BEGIN + SELECT RAISE(IGNORE); + END + `.execute(runtime.db); + + await expect(runtime.runMediaUsageMaintenanceStep()).resolves.toEqual({ + state: "progress", + continuation: { kind: "immediate" }, + taskClass: "reconciliation", + turn: 2, + }); + expect(await countWork(runtime)).toBe(1); + }); + + it("stops continuation only after a full idle maintenance pass", async () => { + runtime = await EmDashRuntime.create(createDeps(null)); + await activateCollection(runtime, "idle_posts"); + + await expect(runtime.runMediaUsageMaintenanceStep()).resolves.toEqual({ + state: "idle", + continuation: { kind: "none" }, + taskClass: "entry_work", + turn: 0, + }); + }); + + it("rotates first consideration across continuously due maintenance classes", async () => { + runtime = await EmDashRuntime.create(createDeps(null)); + const work = await activateCollection(runtime, "fair_engine_work"); + await insertEntry(runtime, work.tableName, "entry-1"); + await activateCollection(runtime, "fair_engine_delete"); + await runtime.schemaRegistry.deleteCollection("fair_engine_delete", { force: true }); + const reconciliation = await activateCollection(runtime, "fair_engine_reconciliation"); + await runtime.db + .updateTable("_emdash_media_usage_index_status") + .set({ status: "stale", reconciliation_required: 1 }) + .where("collection_id", "=", reconciliation.collectionId) + .execute(); + + const classes = []; + for (let index = 0; index < 3; index++) { + const result = await runtime.runMediaUsageMaintenanceStep(); + classes.push(result.taskClass); + expect(result).toMatchObject({ + state: "progress", + continuation: { kind: "immediate" }, + }); + } + + expect(classes).toEqual(["entry_work", "collection_deletion", "reconciliation"]); + }); + it("drains bounded work from the Node timer maintenance callback", async () => { const scheduler = new CapturingScheduler(); runtime = await EmDashRuntime.create(createDeps(() => scheduler)); From a02ab2fe0235fb8ef2ce9cf3ff938fc867908a84 Mon Sep 17 00:00:00 2001 From: khoinguyenpham04 <137921741+khoinguyenpham04@users.noreply.github.com> Date: Mon, 17 Aug 2026 17:30:32 +0100 Subject: [PATCH 2/4] feat(core): continue media usage maintenance on Node --- packages/core/src/emdash-runtime.ts | 19 ++- packages/core/src/index.ts | 1 + packages/core/src/plugins/index.ts | 6 +- packages/core/src/plugins/scheduler/node.ts | 79 +++++++++- packages/core/src/plugins/scheduler/types.ts | 6 + .../media-usage-scheduled-driver.test.ts | 49 +++++- .../unit/plugins/node-cron-scheduler.test.ts | 148 ++++++++++++++++++ 7 files changed, 302 insertions(+), 6 deletions(-) create mode 100644 packages/core/tests/unit/plugins/node-cron-scheduler.test.ts diff --git a/packages/core/src/emdash-runtime.ts b/packages/core/src/emdash-runtime.ts index 38b521ce89..22f76a0a8e 100644 --- a/packages/core/src/emdash-runtime.ts +++ b/packages/core/src/emdash-runtime.ts @@ -1742,6 +1742,14 @@ export class EmDashRuntime { await runScheduledMediaUsageLane(db); } }); + const runContinuousMediaUsageMaintenance = () => + runWithContext({ editMode: false }, async () => { + const runtime = runtimeRef.current; + const result = runtime + ? await runtime.runMediaUsageMaintenanceStep() + : await runMediaUsageMaintenanceStep(db); + return result.continuation; + }); // Run scheduled publishing and system cleanup alongside each tick. // Pass storage so cleanupPendingUploads can delete orphaned files. @@ -1774,7 +1782,10 @@ export class EmDashRuntime { } // Never throws; no-op unless scheduled backups are enabled and due. await maybeRunScheduledBackup(db, storage ?? undefined); - if (!scheduler.setMediaUsageMaintenance) { + if ( + !scheduler.setContinuousMediaUsageMaintenance && + !scheduler.setMediaUsageMaintenance + ) { try { await runMediaUsageMaintenance(); } catch (error) { @@ -1782,7 +1793,11 @@ export class EmDashRuntime { } } }); - scheduler.setMediaUsageMaintenance?.(runMediaUsageMaintenance); + if (scheduler.setContinuousMediaUsageMaintenance) { + scheduler.setContinuousMediaUsageMaintenance(runContinuousMediaUsageMaintenance); + } else { + scheduler.setMediaUsageMaintenance?.(runMediaUsageMaintenance); + } // start() is void on the timer scheduler but the interface // allows a promise (alarm-backed schedulers); we don't block on it. diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index c2f90ecfd0..556420949e 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -292,6 +292,7 @@ export type { // Scheduler types CronScheduler, + MediaUsageContinuationFn, SystemCleanupFn, // Sandbox runtime types diff --git a/packages/core/src/plugins/index.ts b/packages/core/src/plugins/index.ts index c58624916a..4cf36df5c5 100644 --- a/packages/core/src/plugins/index.ts +++ b/packages/core/src/plugins/index.ts @@ -66,7 +66,11 @@ export type { PluginManagerOptions, PluginState } from "./manager.js"; // Scheduler (Node timer-based heartbeat; consumed by the generated // virtual:emdash/scheduler module on non-serverless adapters) export { NodeCronScheduler } from "./scheduler/node.js"; -export type { CronScheduler, SystemCleanupFn } from "./scheduler/types.js"; +export type { + CronScheduler, + MediaUsageContinuationFn, + SystemCleanupFn, +} from "./scheduler/types.js"; // Sandbox export { diff --git a/packages/core/src/plugins/scheduler/node.ts b/packages/core/src/plugins/scheduler/node.ts index 32e63d9ded..c417114151 100644 --- a/packages/core/src/plugins/scheduler/node.ts +++ b/packages/core/src/plugins/scheduler/node.ts @@ -10,7 +10,7 @@ */ import type { CronExecutor } from "../cron.js"; -import type { CronScheduler, SystemCleanupFn } from "./types.js"; +import type { CronScheduler, MediaUsageContinuationFn, SystemCleanupFn } from "./types.js"; /** Minimum polling interval (ms) — prevents tight loops if next_run_at is in the past */ const MIN_INTERVAL_MS = 1000; @@ -30,6 +30,11 @@ export class NodeCronScheduler implements CronScheduler { private running = false; private systemCleanup: SystemCleanupFn | null = null; private mediaUsageMaintenance: SystemCleanupFn | null = null; + private continuousMediaUsageMaintenance: MediaUsageContinuationFn | null = null; + private mediaUsageTimer: ReturnType | null = null; + private mediaUsageInFlight = false; + private mediaUsagePending = false; + private mediaUsageGeneration = 0; constructor(private executor: CronExecutor) {} @@ -41,6 +46,10 @@ export class NodeCronScheduler implements CronScheduler { this.mediaUsageMaintenance = fn; } + setContinuousMediaUsageMaintenance(fn: MediaUsageContinuationFn): void { + this.continuousMediaUsageMaintenance = fn; + } + start(): void { this.running = true; this.arm(); @@ -52,6 +61,12 @@ export class NodeCronScheduler implements CronScheduler { clearTimeout(this.timer); this.timer = null; } + if (this.mediaUsageTimer) { + clearTimeout(this.mediaUsageTimer); + this.mediaUsageTimer = null; + } + this.mediaUsagePending = false; + this.mediaUsageGeneration++; } reschedule(): void { @@ -123,7 +138,9 @@ export class NodeCronScheduler implements CronScheduler { console.error("[cron:node] Tick task failed:", r.reason); } } - if (this.mediaUsageMaintenance) { + if (this.continuousMediaUsageMaintenance) { + this.requestContinuousMediaUsageMaintenance(); + } else if (this.mediaUsageMaintenance) { try { await this.mediaUsageMaintenance(); } catch (error) { @@ -138,4 +155,62 @@ export class NodeCronScheduler implements CronScheduler { } }); } + + private requestContinuousMediaUsageMaintenance(): void { + if (!this.running || !this.continuousMediaUsageMaintenance) return; + if (this.mediaUsageInFlight) { + this.mediaUsagePending = true; + return; + } + if (this.mediaUsageTimer) return; + this.armMediaUsageTimer(0); + } + + private armMediaUsageTimer(delayMs: number): void { + if (!this.running || !this.continuousMediaUsageMaintenance) return; + this.mediaUsageTimer = setTimeout(() => { + this.mediaUsageTimer = null; + void this.executeContinuousMediaUsageMaintenance(); + }, delayMs); + if ( + this.mediaUsageTimer && + typeof this.mediaUsageTimer === "object" && + "unref" in this.mediaUsageTimer + ) { + this.mediaUsageTimer.unref(); + } + } + + private async executeContinuousMediaUsageMaintenance(): Promise { + if (!this.running || !this.continuousMediaUsageMaintenance) return; + if (this.mediaUsageInFlight) { + this.mediaUsagePending = true; + return; + } + + const generation = this.mediaUsageGeneration; + this.mediaUsageInFlight = true; + let continuation: Awaited> | null = null; + try { + continuation = await this.continuousMediaUsageMaintenance(); + } catch (error) { + console.error("[cron:node] Media Usage maintenance failed:", error); + } finally { + this.mediaUsageInFlight = false; + } + + if (!this.running || generation !== this.mediaUsageGeneration) return; + const pending = this.mediaUsagePending; + this.mediaUsagePending = false; + if (!continuation) return; + if (continuation.kind === "immediate") { + this.armMediaUsageTimer(0); + return; + } + if (continuation.kind === "delayed") { + this.armMediaUsageTimer(continuation.delaySeconds * 1_000); + return; + } + if (pending) this.armMediaUsageTimer(0); + } } diff --git a/packages/core/src/plugins/scheduler/types.ts b/packages/core/src/plugins/scheduler/types.ts index f4ee58afbb..6deffcf65a 100644 --- a/packages/core/src/plugins/scheduler/types.ts +++ b/packages/core/src/plugins/scheduler/types.ts @@ -9,6 +9,10 @@ * */ +import type { MediaUsageMaintenanceContinuation } from "../../media/usage/maintenance-engine.js"; + +export type MediaUsageContinuationFn = () => Promise; + export interface CronScheduler { /** Start the scheduler. */ start(): void | Promise; @@ -20,6 +24,8 @@ export interface CronScheduler { setSystemCleanup(fn: SystemCleanupFn): void; /** Register bounded Media Usage maintenance to run after the general tick settles. */ setMediaUsageMaintenance?(fn: SystemCleanupFn): void; + /** Register continuation-capable Media Usage maintenance. */ + setContinuousMediaUsageMaintenance?(fn: MediaUsageContinuationFn): void; } /** diff --git a/packages/core/tests/integration/runtime/media-usage-scheduled-driver.test.ts b/packages/core/tests/integration/runtime/media-usage-scheduled-driver.test.ts index 787482f018..c7db0653f4 100644 --- a/packages/core/tests/integration/runtime/media-usage-scheduled-driver.test.ts +++ b/packages/core/tests/integration/runtime/media-usage-scheduled-driver.test.ts @@ -12,7 +12,11 @@ import { } from "../../../src/emdash-runtime.js"; import { activateMediaUsageCapture } from "../../../src/media/usage/activation.js"; import { installMediaUsageCaptureTriggers } from "../../../src/media/usage/capture-triggers.js"; -import type { CronScheduler, SystemCleanupFn } from "../../../src/plugins/scheduler/types.js"; +import type { + CronScheduler, + MediaUsageContinuationFn, + SystemCleanupFn, +} from "../../../src/plugins/scheduler/types.js"; import { createRequestMetrics, runWithContext } from "../../../src/request-context.js"; describe("media usage scheduled drivers", () => { @@ -185,6 +189,20 @@ describe("media usage scheduled drivers", () => { ).not.toBeNull(); }); + it("registers the continuation-capable Node scheduler without changing legacy hooks", async () => { + const scheduler = new ContinuousCapturingScheduler(); + runtime = await EmDashRuntime.create(createDeps(() => scheduler)); + const fixture = await activateCollection(runtime, "continuous_node_posts"); + await insertEntry(runtime, fixture.tableName, "entry-1"); + await insertEntry(runtime, fixture.tableName, "entry-2"); + + await expect(scheduler.runMaintenance()).resolves.toEqual({ kind: "immediate" }); + expect(await countWork(runtime)).toBe(1); + await expect(scheduler.runContinuation()).resolves.toEqual({ kind: "immediate" }); + expect(await countWork(runtime)).toBe(0); + await expect(scheduler.runContinuation()).resolves.toEqual({ kind: "none" }); + }); + it("starts reconciliation when Node maintenance inherits an expensive request context", async () => { const scheduler = new CapturingScheduler(); const metrics = createRequestMetrics(performance.now()); @@ -394,6 +412,35 @@ class CapturingScheduler implements CronScheduler { } } +class ContinuousCapturingScheduler implements CronScheduler { + private maintenance: SystemCleanupFn | null = null; + private mediaUsageMaintenance: MediaUsageContinuationFn | null = null; + + setSystemCleanup(fn: SystemCleanupFn): void { + this.maintenance = fn; + } + setContinuousMediaUsageMaintenance(fn: MediaUsageContinuationFn): void { + this.mediaUsageMaintenance = fn; + } + + start(): void {} + stop(): void {} + reschedule(): void {} + + async runMaintenance() { + if (!this.maintenance) throw new Error("Expected Node maintenance callback"); + await this.maintenance(); + return this.runContinuation(); + } + + async runContinuation() { + if (!this.mediaUsageMaintenance) { + throw new Error("Expected continuous Media Usage maintenance callback"); + } + return this.mediaUsageMaintenance(); + } +} + class LegacyCapturingScheduler implements CronScheduler { private maintenance: SystemCleanupFn | null = null; diff --git a/packages/core/tests/unit/plugins/node-cron-scheduler.test.ts b/packages/core/tests/unit/plugins/node-cron-scheduler.test.ts new file mode 100644 index 0000000000..e3a6566d15 --- /dev/null +++ b/packages/core/tests/unit/plugins/node-cron-scheduler.test.ts @@ -0,0 +1,148 @@ +import type { Kysely } from "kysely"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import type { Database } from "../../../src/database/types.js"; +import { CronExecutor } from "../../../src/plugins/cron.js"; +import { NodeCronScheduler } from "../../../src/plugins/scheduler/node.js"; +import { setupTestDatabase, teardownTestDatabase } from "../../utils/test-db.js"; + +describe("NodeCronScheduler Media Usage continuation", () => { + let db: Kysely; + let executor: CronExecutor; + let scheduler: NodeCronScheduler; + + beforeEach(async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-08-17T12:00:00.000Z")); + db = await setupTestDatabase(); + executor = new CronExecutor(db, async () => {}); + vi.spyOn(executor, "getNextDueTime").mockResolvedValue(null); + vi.spyOn(executor, "tick").mockResolvedValue(0); + vi.spyOn(executor, "recoverStaleLocks").mockResolvedValue(0); + scheduler = new NodeCronScheduler(executor); + }); + + afterEach(async () => { + scheduler.stop(); + vi.useRealTimers(); + await teardownTestDatabase(db); + vi.restoreAllMocks(); + }); + + it("yields to a new timer turn between immediate units", async () => { + const maintenance = vi + .fn() + .mockResolvedValueOnce({ kind: "immediate" } as const) + .mockResolvedValueOnce({ kind: "none" } as const); + scheduler.setContinuousMediaUsageMaintenance(maintenance); + scheduler.start(); + + await vi.advanceTimersToNextTimerAsync(); + expect(maintenance).toHaveBeenCalledTimes(1); + await vi.advanceTimersToNextTimerAsync(); + expect(maintenance).toHaveBeenCalledTimes(2); + }); + + it("does not let heartbeats shorten a delayed continuation", async () => { + vi.mocked(executor.getNextDueTime).mockImplementation(async () => + new Date(Date.now()).toISOString(), + ); + const maintenance = vi + .fn() + .mockResolvedValueOnce({ kind: "delayed", delaySeconds: 30 } as const) + .mockResolvedValue({ kind: "none" } as const); + scheduler.setContinuousMediaUsageMaintenance(maintenance); + scheduler.start(); + + await vi.advanceTimersToNextTimerAsync(); + expect(maintenance).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(29_000); + expect(maintenance).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(1_000); + expect(maintenance).toHaveBeenCalledTimes(2); + }); + + it("keeps general heartbeats running without overlapping a held unit", async () => { + vi.mocked(executor.getNextDueTime).mockImplementation(async () => + new Date(Date.now()).toISOString(), + ); + let releaseFirst!: () => void; + const first = new Promise((resolve) => { + releaseFirst = resolve; + }); + let active = 0; + let maximumActive = 0; + const maintenance = vi.fn(async () => { + active++; + maximumActive = Math.max(maximumActive, active); + if (maintenance.mock.calls.length === 1) await first; + active--; + return { kind: "none" } as const; + }); + const cleanup = vi.fn(async () => {}); + scheduler.setSystemCleanup(cleanup); + scheduler.setContinuousMediaUsageMaintenance(maintenance); + scheduler.start(); + + await vi.advanceTimersToNextTimerAsync(); + expect(maintenance).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(3_000); + expect(executor.tick).toHaveBeenCalledTimes(4); + expect(cleanup).toHaveBeenCalledTimes(4); + expect(maximumActive).toBe(1); + + releaseFirst(); + await vi.advanceTimersByTimeAsync(0); + expect(maintenance).toHaveBeenCalledTimes(2); + expect(maximumActive).toBe(1); + }); + + it("clears a pending continuation when stopped", async () => { + const maintenance = vi.fn().mockResolvedValue({ kind: "immediate" } as const); + scheduler.setContinuousMediaUsageMaintenance(maintenance); + scheduler.start(); + + await vi.advanceTimersToNextTimerAsync(); + expect(maintenance).toHaveBeenCalledTimes(1); + scheduler.stop(); + await vi.runAllTimersAsync(); + expect(maintenance).toHaveBeenCalledTimes(1); + }); + + it("unrefs Media Usage continuation timers", async () => { + const timeout = vi.spyOn(globalThis, "setTimeout"); + const maintenance = vi.fn().mockResolvedValue({ kind: "none" } as const); + scheduler.setContinuousMediaUsageMaintenance(maintenance); + scheduler.start(); + + await vi.advanceTimersToNextTimerAsync(); + const mediaTimerIndex = timeout.mock.calls.findIndex((call) => call[1] === 0); + expect(mediaTimerIndex).toBeGreaterThanOrEqual(0); + expect(isUnreferencedTimer(timeout.mock.results[mediaTimerIndex]?.value)).toBe(true); + }); + + it("logs a failed unit once and waits for heartbeat recovery", async () => { + const error = vi.spyOn(console, "error").mockImplementation(() => {}); + const maintenance = vi + .fn() + .mockRejectedValueOnce(new Error("maintenance failed")) + .mockResolvedValue({ kind: "none" } as const); + scheduler.setContinuousMediaUsageMaintenance(maintenance); + scheduler.start(); + + await vi.advanceTimersToNextTimerAsync(); + expect(maintenance).toHaveBeenCalledTimes(1); + expect(error).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(0); + expect(maintenance).toHaveBeenCalledTimes(1); + await vi.advanceTimersToNextTimerAsync(); + expect(maintenance).toHaveBeenCalledTimes(2); + expect(error).toHaveBeenCalledTimes(1); + }); +}); + +function isUnreferencedTimer(value: unknown): boolean { + if (!value || typeof value !== "object" || !("hasRef" in value)) return false; + const hasRef = value.hasRef; + return typeof hasRef === "function" && hasRef.call(value) === false; +} From c62935a0c4c01faaebf3ff8a533112c71227f44c Mon Sep 17 00:00:00 2001 From: khoinguyenpham04 <137921741+khoinguyenpham04@users.noreply.github.com> Date: Mon, 17 Aug 2026 17:58:49 +0100 Subject: [PATCH 3/4] feat(cloudflare): continue media usage work through queue wakeups --- .changeset/fast-media-usage-backfill.md | 6 + demos/cloudflare/src/worker.ts | 14 +- demos/cloudflare/worker-configuration.d.ts | 1 + demos/cloudflare/wrangler.jsonc | 19 +- .../content/docs/deployment/cloudflare.mdx | 42 +++- docs/src/content/docs/deployment/nodejs.mdx | 4 + packages/cloudflare/src/worker.ts | 74 ++++++- .../cloudflare/tests/worker-scheduled.test.ts | 199 +++++++++++++++++- .../media-usage-maintenance-engine-d1.test.ts | 159 ++++++++++++++ templates/blog-cloudflare/src/worker.ts | 14 +- templates/blog-cloudflare/tsconfig.json | 2 +- .../blog-cloudflare/worker-configuration.d.ts | 1 + templates/blog-cloudflare/wrangler.jsonc | 17 ++ templates/marketing-cloudflare/src/worker.ts | 14 +- .../worker-configuration.d.ts | 1 + templates/marketing-cloudflare/wrangler.jsonc | 17 ++ templates/portfolio-cloudflare/src/worker.ts | 14 +- .../worker-configuration.d.ts | 1 + templates/portfolio-cloudflare/wrangler.jsonc | 17 ++ templates/starter-cloudflare/src/worker.ts | 14 +- templates/starter-cloudflare/tsconfig.json | 2 +- .../worker-configuration.d.ts | 1 + templates/starter-cloudflare/wrangler.jsonc | 17 ++ 23 files changed, 618 insertions(+), 32 deletions(-) create mode 100644 .changeset/fast-media-usage-backfill.md create mode 100644 packages/core/tests/workerd/media-usage-maintenance-engine-d1.test.ts diff --git a/.changeset/fast-media-usage-backfill.md b/.changeset/fast-media-usage-backfill.md new file mode 100644 index 0000000000..60d66c7ba6 --- /dev/null +++ b/.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. diff --git a/demos/cloudflare/src/worker.ts b/demos/cloudflare/src/worker.ts index d154c752d9..9a0e621779 100644 --- a/demos/cloudflare/src/worker.ts +++ b/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; + scheduled: createScheduledHandler({ resolveMediaUsageQueue }), + queue: createMediaUsageQueueHandler(resolveMediaUsageQueue), +} satisfies ExportedHandler; diff --git a/demos/cloudflare/worker-configuration.d.ts b/demos/cloudflare/worker-configuration.d.ts index 6603007357..6dc5705216 100644 --- a/demos/cloudflare/worker-configuration.d.ts +++ b/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; } diff --git a/demos/cloudflare/wrangler.jsonc b/demos/cloudflare/wrangler.jsonc index 71bbfb825c..d58bad3d5e 100644 --- a/demos/cloudflare/wrangler.jsonc +++ b/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 * * * *"], }, diff --git a/docs/src/content/docs/deployment/cloudflare.mdx b/docs/src/content/docs/deployment/cloudflare.mdx index 75987a190f..0efa17218a 100644 --- a/docs/src/content/docs/deployment/cloudflare.mdx +++ b/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; ``` -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`. ### Enable automatic media usage indexing diff --git a/docs/src/content/docs/deployment/nodejs.mdx b/docs/src/content/docs/deployment/nodejs.mdx index 4d3cd9a2dc..6cbc0a7923 100644 --- a/docs/src/content/docs/deployment/nodejs.mdx +++ b/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. diff --git a/packages/cloudflare/src/worker.ts b/packages/cloudflare/src/worker.ts index d2272d38a3..fe9470f446 100644 --- a/packages/cloudflare/src/worker.ts +++ b/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, +) => Queue | undefined; + +export type MediaUsageQueueResolver = (env: Env) => Queue; + +export interface ScheduledHandlerOptions { generalCron?: string; mediaUsageCron?: string; + resolveMediaUsageQueue?: OptionalMediaUsageQueueResolver; } const DEFAULT_MEDIA_USAGE_CRON = "*/2 * * * *"; -export function createScheduledHandler( - options?: ScheduledHandlerOptions, -): ExportedHandlerScheduledHandler { +export function createScheduledHandler( + options?: ScheduledHandlerOptions, +): ExportedHandlerScheduledHandler { 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 | 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( + resolveMediaUsageQueue: MediaUsageQueueResolver, +): ExportedHandlerQueueHandler { + 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; diff --git a/packages/cloudflare/tests/worker-scheduled.test.ts b/packages/cloudflare/tests/worker-scheduled.test.ts index 034844f207..c605d2448d 100644 --- a/packages/cloudflare/tests/worker-scheduled.test.ts +++ b/packages/cloudflare/tests/worker-scheduled.test.ts @@ -1,8 +1,21 @@ import { beforeEach, expect, it, vi } from "vitest"; +type MaintenanceStepResult = { + state: "inactive" | "idle" | "blocked" | "progress"; + continuation: { kind: "none" } | { kind: "immediate" } | { kind: "delayed"; delaySeconds: 30 }; + taskClass: "entry_work" | "collection_deletion" | "reconciliation" | null; + turn: number | null; +}; + const scheduled = vi.hoisted(() => ({ general: vi.fn(async () => ({ published: [] })), mediaUsage: vi.fn(async () => ({ outcome: "inactive", taskClass: null, turn: null })), + mediaUsageStep: vi.fn<() => Promise>(async () => ({ + state: "idle", + continuation: { kind: "none" }, + taskClass: "entry_work", + turn: 0, + })), })); vi.mock("@astrojs/cloudflare/entrypoints/server", () => ({ default: { fetch: vi.fn() } })); @@ -12,14 +25,27 @@ vi.mock("astro/app/entrypoint", () => ({ vi.mock("emdash/middleware", () => ({ runScheduledTasks: scheduled.general, runScheduledMediaUsageTasks: scheduled.mediaUsage, + runMediaUsageMaintenanceStep: scheduled.mediaUsageStep, })); vi.mock("../src/sandbox/index.js", () => ({ PluginBridge: vi.fn() })); -import { createScheduledHandler } from "../src/worker.js"; +import { + createMediaUsageQueueHandler, + createScheduledHandler, + type MediaUsageWakeMessage, + type ScheduledHandlerOptions, +} from "../src/worker.js"; beforeEach(() => { scheduled.general.mockClear(); scheduled.mediaUsage.mockClear(); + scheduled.mediaUsageStep.mockClear(); + scheduled.mediaUsageStep.mockResolvedValue({ + state: "idle", + continuation: { kind: "none" }, + taskClass: "entry_work", + turn: 0, + }); }); it("uses the default Media Usage expression and treats every other expression as general", async () => { @@ -95,13 +121,180 @@ it("rejects empty or aliased configured expressions", () => { expect(() => createScheduledHandler({ generalCron: " */2 * * * * " })).toThrow(/must differ/i); }); -async function invoke(handler: ExportedHandlerScheduledHandler, cron: string): Promise { +it("keeps the existing non-generic scheduled options type usable", () => { + const options: ScheduledHandlerOptions = { mediaUsageCron: "*/5 * * * *" }; + expect(createScheduledHandler(options)).toBeTypeOf("function"); +}); + +it("uses configured Cron as a Queue wake without initializing Media Usage", async () => { + const send = vi.fn(async () => {}); + const queue = queueBinding(send); + const handler = createScheduledHandler<{ MEDIA_USAGE_QUEUE: Queue }>({ + generalCron: "* * * * *", + mediaUsageCron: "*/2 * * * *", + resolveMediaUsageQueue: (env) => env.MEDIA_USAGE_QUEUE, + }); + + await invoke(handler, "*/2 * * * *", { MEDIA_USAGE_QUEUE: queue }); + + expect(send).toHaveBeenCalledExactlyOnceWith({ version: 1 }); + expect(scheduled.mediaUsage).not.toHaveBeenCalled(); + expect(scheduled.mediaUsageStep).not.toHaveBeenCalled(); +}); + +it("keeps direct scheduled maintenance when the optional Queue is unavailable", async () => { + const handler = createScheduledHandler({ + resolveMediaUsageQueue: () => undefined, + }); + + await invoke(handler, "*/2 * * * *", {}); + + expect(scheduled.mediaUsage).toHaveBeenCalledOnce(); + expect(scheduled.mediaUsageStep).not.toHaveBeenCalled(); +}); + +it("logs a redacted Cron wake failure and leaves recovery to the next trigger", async () => { + const queue = queueBinding( + vi.fn(async () => { + throw new Error("private binding detail"); + }), + ); + const error = vi.spyOn(console, "error").mockImplementation(() => {}); + const handler = createScheduledHandler<{ MEDIA_USAGE_QUEUE: Queue }>({ + resolveMediaUsageQueue: (env) => env.MEDIA_USAGE_QUEUE, + }); + + await invoke(handler, "*/2 * * * *", { MEDIA_USAGE_QUEUE: queue }); + + expect(error).toHaveBeenCalledExactlyOnceWith( + "[scheduled] Failed to queue Media Usage maintenance wake", + ); + expect(JSON.stringify(error.mock.calls)).not.toContain("private binding detail"); + expect(scheduled.mediaUsage).not.toHaveBeenCalled(); +}); + +it("coalesces a delivered batch into one step and one successor", async () => { + const send = vi.fn(async () => {}); + const handler = createMediaUsageQueueHandler(() => queueBinding(send)); + scheduled.mediaUsageStep.mockResolvedValue({ + state: "progress", + continuation: { kind: "immediate" }, + taskClass: "entry_work", + turn: 0, + }); + + await invokeQueue(handler, [wakeMessage(), wakeMessage()], {}); + + expect(scheduled.mediaUsageStep).toHaveBeenCalledOnce(); + expect(send).toHaveBeenCalledExactlyOnceWith({ version: 1 }); +}); + +it("lets the Queue drain when the durable database is idle", async () => { + const send = vi.fn(async () => {}); + const handler = createMediaUsageQueueHandler(() => queueBinding(send)); + + await invokeQueue(handler, [wakeMessage()], {}); + + expect(scheduled.mediaUsageStep).toHaveBeenCalledOnce(); + expect(send).not.toHaveBeenCalled(); +}); + +it("acknowledges invalid wakes without logging their body or running work", async () => { + const send = vi.fn(async () => {}); + const handler = createMediaUsageQueueHandler(() => queueBinding(send)); + const invalid = wakeMessage({ version: 2, secret: "do-not-log" }); + const warning = vi.spyOn(console, "warn").mockImplementation(() => {}); + + await invokeQueue(handler, [invalid], {}); + + expect(invalid.ack).toHaveBeenCalledOnce(); + expect(warning).toHaveBeenCalledWith("[queue] Ignoring invalid Media Usage wake"); + expect(JSON.stringify(warning.mock.calls)).not.toContain("do-not-log"); + expect(scheduled.mediaUsageStep).not.toHaveBeenCalled(); + expect(send).not.toHaveBeenCalled(); +}); + +it("retries valid wakes if a delayed successor cannot be sent", async () => { + const send = vi.fn(async () => { + throw new Error("send failed"); + }); + const handler = createMediaUsageQueueHandler(() => queueBinding(send)); + const invalid = wakeMessage({ version: 9 }); + const valid = wakeMessage(); + scheduled.mediaUsageStep.mockResolvedValue({ + state: "blocked", + continuation: { kind: "delayed", delaySeconds: 30 }, + taskClass: "entry_work", + turn: 0, + }); + vi.spyOn(console, "warn").mockImplementation(() => {}); + + await expect(invokeQueue(handler, [invalid, valid], {})).rejects.toThrow("send failed"); + + expect(invalid.ack).toHaveBeenCalledOnce(); + expect(valid.ack).not.toHaveBeenCalled(); + expect(send).toHaveBeenCalledExactlyOnceWith({ version: 1 }, { delaySeconds: 30 }); +}); + +it("fails before database work when the required Queue binding is missing", async () => { + type MissingQueueEnv = { MEDIA_USAGE_QUEUE?: Queue }; + const handler = createMediaUsageQueueHandler((env) => env.MEDIA_USAGE_QUEUE!); + const invalid = wakeMessage({ version: 4 }); + const valid = wakeMessage(); + vi.spyOn(console, "warn").mockImplementation(() => {}); + + await expect(invokeQueue(handler, [invalid, valid], {})).rejects.toThrow(/binding/i); + expect(invalid.ack).toHaveBeenCalledOnce(); + expect(valid.ack).not.toHaveBeenCalled(); + expect(scheduled.mediaUsageStep).not.toHaveBeenCalled(); +}); + +async function invoke( + handler: ExportedHandlerScheduledHandler, + cron: string, + env: Env, +): Promise; +async function invoke(handler: ExportedHandlerScheduledHandler, cron: string): Promise; +async function invoke( + handler: ExportedHandlerScheduledHandler, + cron: string, + env: unknown = {}, +): Promise { const pending: Promise[] = []; const context = { waitUntil(promise: Promise) { pending.push(promise); }, }; - Reflect.apply(handler, undefined, [{ cron }, {}, context]); + Reflect.apply(handler, undefined, [{ cron }, env, context]); await Promise.all(pending); } + +async function invokeQueue( + handler: ExportedHandlerQueueHandler, + messages: Message[], + env: Env, +): Promise { + const batch: MessageBatch = { + messages, + queue: "media-usage", + retryAll: vi.fn(), + ackAll: vi.fn(), + }; + await Reflect.apply(handler, undefined, [batch, env, {}]); +} + +function wakeMessage(body: unknown = { version: 1 }): Message { + return { + id: crypto.randomUUID(), + timestamp: new Date(), + body, + attempts: 1, + retry: vi.fn(), + ack: vi.fn(), + }; +} + +function queueBinding(send: Queue["send"]): Queue { + return { send, sendBatch: vi.fn(async () => {}) }; +} diff --git a/packages/core/tests/workerd/media-usage-maintenance-engine-d1.test.ts b/packages/core/tests/workerd/media-usage-maintenance-engine-d1.test.ts new file mode 100644 index 0000000000..8aa354a1c9 --- /dev/null +++ b/packages/core/tests/workerd/media-usage-maintenance-engine-d1.test.ts @@ -0,0 +1,159 @@ +import { env } from "cloudflare:test"; +import { Kysely, sql } from "kysely"; +import { afterAll, beforeAll, expect, it } from "vitest"; + +import { RawBindingD1Dialect } from "../../../cloudflare/src/db/d1-dialect.js"; +import { runMigrations } from "../../src/database/migrations/runner.js"; +import type { Database } from "../../src/database/types.js"; +import { installMediaUsageCaptureTriggers } from "../../src/media/usage/capture-triggers.js"; +import { runMediaUsageMaintenanceStep } from "../../src/media/usage/maintenance-engine.js"; +import { SchemaRegistry } from "../../src/schema/registry.js"; + +declare module "cloudflare:test" { + interface ProvidedEnv { + DB: D1Database; + } +} + +interface D1Measurement { + queries: number; + rowsRead: number; + rowsWritten: number; + durationMs: number; + wallDurationMs: number; + maxBinds: number; + maxSqlBytes: number; +} + +let adminDb: Kysely; + +beforeAll(async () => { + adminDb = new Kysely({ + dialect: new RawBindingD1Dialect({ database: env.DB }), + }); + await runMigrations(adminDb); +}); + +afterAll(async () => { + await adminDb.destroy(); +}); + +it("keeps two idle classes plus one useful D1 unit below the hard query boundary", async () => { + const registry = new SchemaRegistry(adminDb); + await registry.createCollection({ slug: "d1_engine_work", label: "D1 engine work" }); + await registry.createField("d1_engine_work", { + slug: "title", + label: "Title", + type: "string", + }); + const collection = await registry.getCollection("d1_engine_work"); + if (!collection) throw new Error("Expected D1 engine collection"); + await adminDb + .updateTable("_emdash_media_usage_index_status") + .set({ + collection_id: collection.id, + status: "complete", + capture_state: "installing", + reconciliation_required: 0, + }) + .where("adapter_id", "=", "content-media") + .where("scope_type", "=", "collection") + .where("scope_key", "=", collection.slug) + .execute(); + await installMediaUsageCaptureTriggers(adminDb, { + collectionId: collection.id, + collectionSlug: collection.slug, + }); + await adminDb + .updateTable("_emdash_media_usage_index_status") + .set({ capture_state: "active" }) + .where("collection_id", "=", collection.id) + .execute(); + await adminDb + .updateTable("_emdash_media_usage_activation") + .set({ state: "active", media_usage_maintenance_turn: 0 }) + .where("task_key", "=", "incremental_capture") + .execute(); + await sql` + INSERT INTO ${sql.ref("ec_d1_engine_work")} (id, slug, status, title) + VALUES ('entry-1', 'entry-1', 'published', 'Entry 1') + `.execute(adminDb); + + const measurement = emptyMeasurement(); + const db = new Kysely({ + dialect: new RawBindingD1Dialect({ database: captureD1(env.DB, measurement) }), + }); + const startedAt = performance.now(); + const result = await runMediaUsageMaintenanceStep(db); + measurement.wallDurationMs = Number((performance.now() - startedAt).toFixed(3)); + await db.destroy(); + + expect(result).toEqual({ + state: "progress", + continuation: { kind: "immediate" }, + taskClass: "entry_work", + turn: 0, + }); + expect(measurement.queries).toBeLessThan(50); + expect(measurement.maxBinds).toBeLessThanOrEqual(100); + expect(measurement.maxSqlBytes).toBeLessThan(100 * 1024); + expect(measurement.wallDurationMs).toBeLessThan(2_500); + console.info(`PR8_D1_MAINTENANCE_STEP=${JSON.stringify(measurement)}`); +}); + +function emptyMeasurement(): D1Measurement { + return { + queries: 0, + rowsRead: 0, + rowsWritten: 0, + durationMs: 0, + wallDurationMs: 0, + maxBinds: 0, + maxSqlBytes: 0, + }; +} + +function captureD1(database: D1Database, measurement: D1Measurement): D1Database { + return new Proxy(database, { + get(target, property) { + if (property === "prepare") { + return (query: string) => captureStatement(target.prepare(query), query, [], measurement); + } + const value: unknown = Reflect.get(target, property, target); + return typeof value === "function" ? value.bind(target) : value; + }, + }); +} + +function captureStatement( + statement: D1PreparedStatement, + query: string, + binds: unknown[], + measurement: D1Measurement, +): D1PreparedStatement { + return new Proxy(statement, { + get(target, property) { + if (property === "bind") { + return (...values: unknown[]) => + captureStatement(target.bind(...values), query, values, measurement); + } + if (property === "all") { + return async () => { + measurement.queries++; + measurement.maxBinds = Math.max(measurement.maxBinds, binds.length); + measurement.maxSqlBytes = Math.max( + measurement.maxSqlBytes, + new TextEncoder().encode(query).byteLength, + ); + const result = await target.all(); + measurement.rowsRead += result.meta.rows_read; + measurement.rowsWritten += result.meta.rows_written; + measurement.durationMs += result.meta.duration; + return result; + }; + } + const value: unknown = Reflect.get(target, property, target); + return typeof value === "function" ? value.bind(target) : value; + }, + }); +} diff --git a/templates/blog-cloudflare/src/worker.ts b/templates/blog-cloudflare/src/worker.ts index 75fe3e2743..09acc79965 100644 --- a/templates/blog-cloudflare/src/worker.ts +++ b/templates/blog-cloudflare/src/worker.ts @@ -1,8 +1,16 @@ -import handler, { createScheduledHandler, PluginBridge } from "@emdash-cms/cloudflare/worker"; +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(), -}; + scheduled: createScheduledHandler({ resolveMediaUsageQueue }), + queue: createMediaUsageQueueHandler(resolveMediaUsageQueue), +} satisfies ExportedHandler; diff --git a/templates/blog-cloudflare/tsconfig.json b/templates/blog-cloudflare/tsconfig.json index 0903753115..c2e1508a58 100644 --- a/templates/blog-cloudflare/tsconfig.json +++ b/templates/blog-cloudflare/tsconfig.json @@ -3,5 +3,5 @@ "compilerOptions": { "types": ["node"] }, - "include": ["src", ".astro/types.d.ts", "emdash-env.d.ts"] + "include": ["src", ".astro/types.d.ts", "emdash-env.d.ts", "worker-configuration.d.ts"] } diff --git a/templates/blog-cloudflare/worker-configuration.d.ts b/templates/blog-cloudflare/worker-configuration.d.ts index 8d6646e145..0f021c3461 100644 --- a/templates/blog-cloudflare/worker-configuration.d.ts +++ b/templates/blog-cloudflare/worker-configuration.d.ts @@ -5,6 +5,7 @@ declare namespace Cloudflare { interface Env { MEDIA: R2Bucket; DB: D1Database; + MEDIA_USAGE_QUEUE: Queue; } } interface Env extends Cloudflare.Env {} diff --git a/templates/blog-cloudflare/wrangler.jsonc b/templates/blog-cloudflare/wrangler.jsonc index 23632b0e98..14a84e55a7 100644 --- a/templates/blog-cloudflare/wrangler.jsonc +++ b/templates/blog-cloudflare/wrangler.jsonc @@ -22,6 +22,23 @@ "binding": "LOADER", }, ], + "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, + }, + ], + }, // General maintenance plus the bounded Media Usage lane (see src/worker.ts) "triggers": { "crons": ["* * * * *", "*/2 * * * *"], diff --git a/templates/marketing-cloudflare/src/worker.ts b/templates/marketing-cloudflare/src/worker.ts index 75fe3e2743..09acc79965 100644 --- a/templates/marketing-cloudflare/src/worker.ts +++ b/templates/marketing-cloudflare/src/worker.ts @@ -1,8 +1,16 @@ -import handler, { createScheduledHandler, PluginBridge } from "@emdash-cms/cloudflare/worker"; +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(), -}; + scheduled: createScheduledHandler({ resolveMediaUsageQueue }), + queue: createMediaUsageQueueHandler(resolveMediaUsageQueue), +} satisfies ExportedHandler; diff --git a/templates/marketing-cloudflare/worker-configuration.d.ts b/templates/marketing-cloudflare/worker-configuration.d.ts index 8d6646e145..0f021c3461 100644 --- a/templates/marketing-cloudflare/worker-configuration.d.ts +++ b/templates/marketing-cloudflare/worker-configuration.d.ts @@ -5,6 +5,7 @@ declare namespace Cloudflare { interface Env { MEDIA: R2Bucket; DB: D1Database; + MEDIA_USAGE_QUEUE: Queue; } } interface Env extends Cloudflare.Env {} diff --git a/templates/marketing-cloudflare/wrangler.jsonc b/templates/marketing-cloudflare/wrangler.jsonc index e5e23e764d..e85e2ffaad 100644 --- a/templates/marketing-cloudflare/wrangler.jsonc +++ b/templates/marketing-cloudflare/wrangler.jsonc @@ -22,6 +22,23 @@ "binding": "LOADER", }, ], + "queues": { + "producers": [ + { + "binding": "MEDIA_USAGE_QUEUE", + "queue": "my-marketing-site-media-usage", + }, + ], + "consumers": [ + { + "queue": "my-marketing-site-media-usage", + "max_batch_size": 1, + "max_batch_timeout": 0, + "max_retries": 3, + "max_concurrency": 1, + }, + ], + }, // General maintenance plus the bounded Media Usage lane (see src/worker.ts) "triggers": { "crons": ["* * * * *", "*/2 * * * *"], diff --git a/templates/portfolio-cloudflare/src/worker.ts b/templates/portfolio-cloudflare/src/worker.ts index 75fe3e2743..09acc79965 100644 --- a/templates/portfolio-cloudflare/src/worker.ts +++ b/templates/portfolio-cloudflare/src/worker.ts @@ -1,8 +1,16 @@ -import handler, { createScheduledHandler, PluginBridge } from "@emdash-cms/cloudflare/worker"; +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(), -}; + scheduled: createScheduledHandler({ resolveMediaUsageQueue }), + queue: createMediaUsageQueueHandler(resolveMediaUsageQueue), +} satisfies ExportedHandler; diff --git a/templates/portfolio-cloudflare/worker-configuration.d.ts b/templates/portfolio-cloudflare/worker-configuration.d.ts index 8d6646e145..0f021c3461 100644 --- a/templates/portfolio-cloudflare/worker-configuration.d.ts +++ b/templates/portfolio-cloudflare/worker-configuration.d.ts @@ -5,6 +5,7 @@ declare namespace Cloudflare { interface Env { MEDIA: R2Bucket; DB: D1Database; + MEDIA_USAGE_QUEUE: Queue; } } interface Env extends Cloudflare.Env {} diff --git a/templates/portfolio-cloudflare/wrangler.jsonc b/templates/portfolio-cloudflare/wrangler.jsonc index 792c1fda4d..3bd27f6a3a 100644 --- a/templates/portfolio-cloudflare/wrangler.jsonc +++ b/templates/portfolio-cloudflare/wrangler.jsonc @@ -22,6 +22,23 @@ "binding": "LOADER", }, ], + "queues": { + "producers": [ + { + "binding": "MEDIA_USAGE_QUEUE", + "queue": "my-portfolio-site-media-usage", + }, + ], + "consumers": [ + { + "queue": "my-portfolio-site-media-usage", + "max_batch_size": 1, + "max_batch_timeout": 0, + "max_retries": 3, + "max_concurrency": 1, + }, + ], + }, // General maintenance plus the bounded Media Usage lane (see src/worker.ts) "triggers": { "crons": ["* * * * *", "*/2 * * * *"], diff --git a/templates/starter-cloudflare/src/worker.ts b/templates/starter-cloudflare/src/worker.ts index 75fe3e2743..09acc79965 100644 --- a/templates/starter-cloudflare/src/worker.ts +++ b/templates/starter-cloudflare/src/worker.ts @@ -1,8 +1,16 @@ -import handler, { createScheduledHandler, PluginBridge } from "@emdash-cms/cloudflare/worker"; +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(), -}; + scheduled: createScheduledHandler({ resolveMediaUsageQueue }), + queue: createMediaUsageQueueHandler(resolveMediaUsageQueue), +} satisfies ExportedHandler; diff --git a/templates/starter-cloudflare/tsconfig.json b/templates/starter-cloudflare/tsconfig.json index 0903753115..c2e1508a58 100644 --- a/templates/starter-cloudflare/tsconfig.json +++ b/templates/starter-cloudflare/tsconfig.json @@ -3,5 +3,5 @@ "compilerOptions": { "types": ["node"] }, - "include": ["src", ".astro/types.d.ts", "emdash-env.d.ts"] + "include": ["src", ".astro/types.d.ts", "emdash-env.d.ts", "worker-configuration.d.ts"] } diff --git a/templates/starter-cloudflare/worker-configuration.d.ts b/templates/starter-cloudflare/worker-configuration.d.ts index 8d6646e145..0f021c3461 100644 --- a/templates/starter-cloudflare/worker-configuration.d.ts +++ b/templates/starter-cloudflare/worker-configuration.d.ts @@ -5,6 +5,7 @@ declare namespace Cloudflare { interface Env { MEDIA: R2Bucket; DB: D1Database; + MEDIA_USAGE_QUEUE: Queue; } } interface Env extends Cloudflare.Env {} diff --git a/templates/starter-cloudflare/wrangler.jsonc b/templates/starter-cloudflare/wrangler.jsonc index 23632b0e98..14a84e55a7 100644 --- a/templates/starter-cloudflare/wrangler.jsonc +++ b/templates/starter-cloudflare/wrangler.jsonc @@ -22,6 +22,23 @@ "binding": "LOADER", }, ], + "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, + }, + ], + }, // General maintenance plus the bounded Media Usage lane (see src/worker.ts) "triggers": { "crons": ["* * * * *", "*/2 * * * *"], From 1fcbc86302f0221ef6c33706bbb0fe6fdcb325c3 Mon Sep 17 00:00:00 2001 From: khoinguyenpham04 <137921741+khoinguyenpham04@users.noreply.github.com> Date: Tue, 18 Aug 2026 10:56:44 +0100 Subject: [PATCH 4/4] fix(admin): format media usage progress fallback --- .changeset/media-usage-progress-locale.md | 5 +++++ packages/admin/src/locales/en/messages.po | 7 +++++++ packages/admin/src/locales/loadMessages.ts | 13 +++++++++---- packages/admin/tests/lib/locales.test.ts | 20 ++++++++++++++++++++ 4 files changed, 41 insertions(+), 4 deletions(-) create mode 100644 .changeset/media-usage-progress-locale.md diff --git a/.changeset/media-usage-progress-locale.md b/.changeset/media-usage-progress-locale.md new file mode 100644 index 0000000000..222a609883 --- /dev/null +++ b/.changeset/media-usage-progress-locale.md @@ -0,0 +1,5 @@ +--- +"@emdash-cms/admin": patch +--- + +Fixes Media Usage progress text displaying untranslated plural syntax. diff --git a/packages/admin/src/locales/en/messages.po b/packages/admin/src/locales/en/messages.po index d25ed2ad35..56cf968aba 100644 --- a/packages/admin/src/locales/en/messages.po +++ b/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 diff --git a/packages/admin/src/locales/loadMessages.ts b/packages/admin/src/locales/loadMessages.ts index aa16230b53..1398cb56dd 100644 --- a/packages/admin/src/locales/loadMessages.ts +++ b/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 { 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 }; } diff --git a/packages/admin/tests/lib/locales.test.ts b/packages/admin/tests/lib/locales.test.ts index 25ff7241ed..7af1756a37 100644 --- a/packages/admin/tests/lib/locales.test.ts +++ b/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", () => {