diff --git a/.changeset/calm-files-reconcile.md b/.changeset/calm-files-reconcile.md new file mode 100644 index 0000000000..f1b97eb9c7 --- /dev/null +++ b/.changeset/calm-files-reconcile.md @@ -0,0 +1,6 @@ +--- +"emdash": patch +"@emdash-cms/cloudflare": patch +--- + +Adds automatic, resumable background indexing so Media Usage can safely catch up on existing content without processing the whole site at once. diff --git a/demos/cloudflare/wrangler.jsonc b/demos/cloudflare/wrangler.jsonc index 9e10c0f6bc..71bbfb825c 100644 --- a/demos/cloudflare/wrangler.jsonc +++ b/demos/cloudflare/wrangler.jsonc @@ -36,7 +36,7 @@ ], // Cron trigger drives the AI Search reindex queue flush. "triggers": { - "crons": ["* * * * *"], + "crons": ["* * * * *", "*/2 * * * *"], }, // Observability "observability": { diff --git a/docs/src/content/docs/deployment/cloudflare.mdx b/docs/src/content/docs/deployment/cloudflare.mdx index d3b9b0bb1b..f08ac5fb4f 100644 --- a/docs/src/content/docs/deployment/cloudflare.mdx +++ b/docs/src/content/docs/deployment/cloudflare.mdx @@ -74,24 +74,36 @@ To change the schema or content model of a site that is already deployed, see [E ## Scheduled Publishing -On Cloudflare Workers, scheduled publishing, plugin cron, and maintenance tasks run from a Worker Cron Trigger. New Cloudflare templates include this setup automatically. If you are adding updating an existing project, export the EmDash Worker entry from `@emdash-cms/cloudflare/worker`: +On Cloudflare Workers, scheduled publishing, plugin cron, and maintenance tasks run from Worker Cron Triggers. New Cloudflare templates include both required schedules automatically. When updating an existing project, configure distinct general and Media Usage lanes: ```ts title="src/worker.ts" -export { default, PluginBridge } from "@emdash-cms/cloudflare/worker"; +import handler, { + createScheduledHandler, + PluginBridge, +} from "@emdash-cms/cloudflare/worker"; + +export { PluginBridge }; + +export default { + ...handler, + scheduled: createScheduledHandler(), +} satisfies ExportedHandler; ``` -Then add a Cron Trigger to `wrangler.jsonc`: +By default, `*/2 * * * *` runs Media Usage maintenance and every other expression runs general maintenance. Then add both Cron Triggers to `wrangler.jsonc`: ```jsonc title="wrangler.jsonc" { "triggers": { - "crons": ["* * * * *"], + "crons": ["* * * * *", "*/2 * * * *"], }, } ``` +To use different schedules, set the corresponding `generalCron` or `mediaUsageCron` option in `createScheduledHandler()` and use the same expression in `wrangler.jsonc`. + ## Deploy diff --git a/packages/cloudflare/src/worker.ts b/packages/cloudflare/src/worker.ts index 12b258d6e9..d2272d38a3 100644 --- a/packages/cloudflare/src/worker.ts +++ b/packages/cloudflare/src/worker.ts @@ -2,17 +2,9 @@ * 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": ["* * * * *"] } + * Cron Triggers drive general maintenance and the separately bounded Media + * Usage lane without request side effects. Re-exports the `PluginBridge` + * Durable Object so the sandbox binding resolves against the entry module. * * The `@astrojs/cloudflare/entrypoints/server` import is resolved by the * consuming app's Astro build (it pulls the build-time `virtual:astro:app` @@ -22,7 +14,7 @@ // @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"; +import { runScheduledMediaUsageTasks, runScheduledTasks } from "emdash/middleware"; export { PluginBridge } from "./sandbox/index.js"; @@ -48,13 +40,43 @@ async function invalidatePublishedTags( } /** - * 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. + * Build a Worker `scheduled()` handler. By default the every-two-minutes + * expression runs Media Usage maintenance and every other expression runs + * general maintenance. Configuring a general expression changes that lane + * from catch-all to exact. */ -export function createScheduledHandler(): ExportedHandlerScheduledHandler { - return (_controller, _env, ctx) => { +export interface ScheduledHandlerOptions { + generalCron?: string; + mediaUsageCron?: string; +} + +const DEFAULT_MEDIA_USAGE_CRON = "*/2 * * * *"; + +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) { + throw new Error("Configured scheduled-handler expressions must be non-empty"); + } + if (generalCron === mediaUsageCron) { + throw new Error("General and Media Usage Cron expressions must differ"); + } + + return (controller, _env, ctx) => { + if (controller.cron === mediaUsageCron) { + ctx.waitUntil( + runScheduledMediaUsageTasks().catch((error: unknown) => { + console.error("[scheduled] Media Usage maintenance failed:", error); + }), + ); + return; + } + if (generalCron !== undefined && controller.cron !== generalCron) { + console.warn(`[scheduled] Ignoring unexpected Cron expression: ${controller.cron}`); + return; + } ctx.waitUntil( // Invalidate incrementally as each collection batch publishes, so a // scheduled() invocation killed mid-sweep (CPU/wall-clock limits on a diff --git a/packages/cloudflare/tests/worker-scheduled.test.ts b/packages/cloudflare/tests/worker-scheduled.test.ts new file mode 100644 index 0000000000..034844f207 --- /dev/null +++ b/packages/cloudflare/tests/worker-scheduled.test.ts @@ -0,0 +1,107 @@ +import { beforeEach, expect, it, vi } from "vitest"; + +const scheduled = vi.hoisted(() => ({ + general: vi.fn(async () => ({ published: [] })), + mediaUsage: vi.fn(async () => ({ outcome: "inactive", taskClass: null, turn: null })), +})); + +vi.mock("@astrojs/cloudflare/entrypoints/server", () => ({ default: { fetch: vi.fn() } })); +vi.mock("astro/app/entrypoint", () => ({ + createApp: () => ({ pipeline: { getCacheProvider: async () => null } }), +})); +vi.mock("emdash/middleware", () => ({ + runScheduledTasks: scheduled.general, + runScheduledMediaUsageTasks: scheduled.mediaUsage, +})); +vi.mock("../src/sandbox/index.js", () => ({ PluginBridge: vi.fn() })); + +import { createScheduledHandler } from "../src/worker.js"; + +beforeEach(() => { + scheduled.general.mockClear(); + scheduled.mediaUsage.mockClear(); +}); + +it("uses the default Media Usage expression and treats every other expression as general", async () => { + const handler = createScheduledHandler(); + + await invoke(handler, "custom expression"); + expect(scheduled.general).toHaveBeenCalledOnce(); + expect(scheduled.mediaUsage).not.toHaveBeenCalled(); + + scheduled.general.mockClear(); + await invoke(handler, "*/2 * * * *"); + expect(scheduled.general).not.toHaveBeenCalled(); + expect(scheduled.mediaUsage).toHaveBeenCalledOnce(); +}); + +it("dispatches distinct configured cron expressions to exactly one lane", async () => { + const handler = createScheduledHandler({ + generalCron: "* * * * *", + mediaUsageCron: "*/2 * * * *", + }); + + await invoke(handler, "* * * * *"); + expect(scheduled.general).toHaveBeenCalledOnce(); + expect(scheduled.mediaUsage).not.toHaveBeenCalled(); + + scheduled.general.mockClear(); + await invoke(handler, "*/2 * * * *"); + expect(scheduled.general).not.toHaveBeenCalled(); + expect(scheduled.mediaUsage).toHaveBeenCalledOnce(); + + scheduled.mediaUsage.mockClear(); + await invoke(handler, "0 0 * * *"); + expect(scheduled.general).not.toHaveBeenCalled(); + expect(scheduled.mediaUsage).not.toHaveBeenCalled(); +}); + +it("allows either default expression to be overridden independently", async () => { + const customMedia = createScheduledHandler({ mediaUsageCron: "*/5 * * * *" }); + await invoke(customMedia, "*/5 * * * *"); + expect(scheduled.mediaUsage).toHaveBeenCalledOnce(); + expect(scheduled.general).not.toHaveBeenCalled(); + + scheduled.mediaUsage.mockClear(); + await invoke(customMedia, "15 * * * *"); + expect(scheduled.mediaUsage).not.toHaveBeenCalled(); + expect(scheduled.general).toHaveBeenCalledOnce(); + + scheduled.general.mockClear(); + const customGeneral = createScheduledHandler({ generalCron: "0 * * * *" }); + await invoke(customGeneral, "*/2 * * * *"); + expect(scheduled.mediaUsage).toHaveBeenCalledOnce(); + expect(scheduled.general).not.toHaveBeenCalled(); + + scheduled.mediaUsage.mockClear(); + await invoke(customGeneral, "0 * * * *"); + expect(scheduled.mediaUsage).not.toHaveBeenCalled(); + expect(scheduled.general).toHaveBeenCalledOnce(); + + scheduled.general.mockClear(); + await invoke(customGeneral, "15 * * * *"); + expect(scheduled.mediaUsage).not.toHaveBeenCalled(); + expect(scheduled.general).not.toHaveBeenCalled(); +}); + +it("rejects empty or aliased configured expressions", () => { + expect(() => + createScheduledHandler({ generalCron: "* * * * *", mediaUsageCron: "* * * * *" }), + ).toThrow(/must differ/i); + expect(() => createScheduledHandler({ generalCron: "", mediaUsageCron: "*/2 * * * *" })).toThrow( + /non-empty/i, + ); + expect(() => createScheduledHandler({ mediaUsageCron: " " })).toThrow(/non-empty/i); + expect(() => createScheduledHandler({ generalCron: " */2 * * * * " })).toThrow(/must differ/i); +}); + +async function invoke(handler: ExportedHandlerScheduledHandler, cron: string): Promise { + const pending: Promise[] = []; + const context = { + waitUntil(promise: Promise) { + pending.push(promise); + }, + }; + Reflect.apply(handler, undefined, [{ cron }, {}, context]); + await Promise.all(pending); +} diff --git a/packages/core/src/astro/middleware.ts b/packages/core/src/astro/middleware.ts index 49bd8e259a..eab7aa1245 100644 --- a/packages/core/src/astro/middleware.ts +++ b/packages/core/src/astro/middleware.ts @@ -39,9 +39,11 @@ import { flushRecorder, isInstrumentationEnabled, } from "../database/instrumentation.js"; +import { createDeferredTaskTracker } from "../deferred-tasks.js"; import { DB_INIT_DEADLINE_MS, EmDashRuntime, + type MediaUsageMaintenanceResult, type RuntimeDependencies, type SandboxedPluginEntry, type MediaProviderEntry, @@ -287,6 +289,12 @@ export async function runScheduledTasks( return runOutsideRequest(config, (runtime) => runtime.runScheduledTasks(options)); } +export async function runScheduledMediaUsageTasks(): Promise { + const config = getConfig(); + if (!config) return { outcome: "inactive", taskClass: null, turn: null }; + return runOutsideRequest(config, (runtime) => runtime.runScheduledMediaUsageTasks()); +} + /** * Run a callback against the EmDash runtime outside any HTTP request — from a * Cloudflare Queue consumer, a `scheduled()` handler, or any other @@ -352,8 +360,35 @@ async function runOutsideRequest( config: EmDashConfig, fn: (runtime: EmDashRuntime) => Promise, ): Promise { - const runtime = await getRuntime(config); + if (getRequestContext()) { + const runtime = await getRuntime(config); + return runOutsideRequestWithRuntime(config, runtime, fn); + } + const deferredTasks = createDeferredTaskTracker(() => {}); + const context = { + editMode: false, + metrics: createRequestMetrics(performance.now()), + deferredTasks, + }; + return runWithContext(context, async () => { + const runtime = await (async () => { + try { + return await getRuntime(config); + } finally { + deferredTasks.settle(); + await deferredTasks.settled; + } + })(); + return runOutsideRequestWithRuntime(config, runtime, fn); + }); +} + +async function runOutsideRequestWithRuntime( + config: EmDashConfig, + runtime: EmDashRuntime, + fn: (runtime: EmDashRuntime) => Promise, +): Promise { const scoped = createRequestScopedDb({ config: config.database?.config, isAuthenticated: false, diff --git a/packages/core/src/database/migrations/066_media_usage_reconciliation.ts b/packages/core/src/database/migrations/066_media_usage_reconciliation.ts new file mode 100644 index 0000000000..c97e409a35 --- /dev/null +++ b/packages/core/src/database/migrations/066_media_usage_reconciliation.ts @@ -0,0 +1,115 @@ +import { sql, type Kysely, type RawBuilder } from "kysely"; + +import { columnExists, isPostgres, tableExists } from "../dialect-helpers.js"; + +const ACTIVATION_KEY = "incremental_capture"; + +export async function up(db: Kysely): Promise { + await db.schema + .createTable("_emdash_media_usage_reconciliations") + .ifNotExists() + .addColumn("collection_id", "text", (column) => column.notNull().primaryKey()) + .addColumn("collection_slug", "text", (column) => column.notNull()) + .addColumn("run_token", "text", (column) => column.notNull()) + .addColumn("target_epoch", "bigint") + .addColumn("field_fingerprint", "text") + .addColumn("state", "text", (column) => column.notNull().defaultTo("pending")) + .addColumn("phase", "text", (column) => column.notNull().defaultTo("scan")) + .addColumn("scan_cursor", "text") + .addColumn("scan_upper_id", "text") + .addColumn("source_cursor", "text") + .addColumn("source_upper_key", "text") + .addColumn("attempt_count", "integer", (column) => column.notNull().defaultTo(0)) + .addColumn("next_attempt_at", "text", (column) => column.notNull()) + .addColumn("lease_token", "text") + .addColumn("lease_expires_at", "text") + .addColumn("last_error_code", "text") + .addColumn("created_at", "text", (column) => + column.notNull().defaultTo(sortableUtcTimestamp(db)), + ) + .addColumn("updated_at", "text", (column) => + column.notNull().defaultTo(sortableUtcTimestamp(db)), + ) + .execute(); + + await db.schema + .createIndex("idx__emdash_media_usage_reconciliations_due") + .ifNotExists() + .on("_emdash_media_usage_reconciliations") + .columns(["state", "next_attempt_at", "updated_at", "collection_id"]) + .execute(); + await db.schema + .createIndex("idx__emdash_media_usage_reconciliations_lease") + .ifNotExists() + .on("_emdash_media_usage_reconciliations") + .columns(["state", "lease_expires_at", "updated_at", "collection_id"]) + .execute(); + await db.schema + .createIndex("idx__emdash_media_usage_reconciliations_failed") + .ifNotExists() + .on("_emdash_media_usage_reconciliations") + .columns(["state", "updated_at", "collection_id"]) + .execute(); + await db.schema + .createIndex("idx__emdash_media_usage_status_reconciliation") + .ifNotExists() + .on("_emdash_media_usage_index_status") + .columns([ + "adapter_id", + "scope_type", + "capture_state", + "reconciliation_required", + "collection_id", + ]) + .execute(); + + if (!(await columnExists(db, "_emdash_media_usage_activation", "media_usage_maintenance_turn"))) { + await db.schema + .alterTable("_emdash_media_usage_activation") + .addColumn("media_usage_maintenance_turn", "integer", (column) => + column.notNull().defaultTo(2), + ) + .execute(); + } +} + +export async function down(db: Kysely): Promise { + if (await tableExists(db, "_emdash_media_usage_reconciliations")) { + const evidence = await sql<{ present: number }>` + SELECT 1 AS present + FROM _emdash_media_usage_reconciliations + LIMIT 1 + `.execute(db); + if (evidence.rows.length > 0) { + throw new Error("Cannot roll back while durable reconciliation evidence exists"); + } + } + + const activation = await sql<{ state: string }>` + SELECT state + FROM _emdash_media_usage_activation + WHERE task_key = ${ACTIVATION_KEY} + `.execute(db); + if (activation.rows[0]?.state !== "expanded") { + throw new Error("Cannot roll back media usage reconciliation after activation has started"); + } + + await db.schema.dropIndex("idx__emdash_media_usage_status_reconciliation").ifExists().execute(); + await db.schema.dropIndex("idx__emdash_media_usage_reconciliations_failed").ifExists().execute(); + await db.schema.dropIndex("idx__emdash_media_usage_reconciliations_lease").ifExists().execute(); + await db.schema.dropIndex("idx__emdash_media_usage_reconciliations_due").ifExists().execute(); + await db.schema.dropTable("_emdash_media_usage_reconciliations").ifExists().execute(); + if (await columnExists(db, "_emdash_media_usage_activation", "media_usage_maintenance_turn")) { + await db.schema + .alterTable("_emdash_media_usage_activation") + .dropColumn("media_usage_maintenance_turn") + .execute(); + } +} + +function sortableUtcTimestamp(db: Kysely): RawBuilder { + if (isPostgres(db)) { + return sql`to_char(clock_timestamp() AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.MS"Z"')`; + } + return sql`(strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))`; +} diff --git a/packages/core/src/database/migrations/runner.ts b/packages/core/src/database/migrations/runner.ts index 640b19d2f0..5fc302a5b2 100644 --- a/packages/core/src/database/migrations/runner.ts +++ b/packages/core/src/database/migrations/runner.ts @@ -68,6 +68,7 @@ import * as m062 from "./062_media_usage_cleanup_fence.js"; import * as m063 from "./063_media_usage_incremental_work.js"; import * as m064 from "./064_fts_plain_text.js"; import * as m065 from "./065_media_usage_collection_deletion.js"; +import * as m066 from "./066_media_usage_reconciliation.js"; const MIGRATIONS: Readonly> = Object.freeze({ "001_initial": m001, @@ -134,6 +135,7 @@ const MIGRATIONS: Readonly> = Object.freeze({ "063_media_usage_incremental_work": m063, "064_fts_plain_text": m064, "065_media_usage_collection_deletion": m065, + "066_media_usage_reconciliation": m066, }); /** Total number of registered migrations. Exported for use in tests. */ diff --git a/packages/core/src/database/repositories/media-usage-work.ts b/packages/core/src/database/repositories/media-usage-work.ts index ffa48e1474..115967a4ab 100644 --- a/packages/core/src/database/repositories/media-usage-work.ts +++ b/packages/core/src/database/repositories/media-usage-work.ts @@ -15,6 +15,7 @@ const STABLE_ERROR_CODE_PATTERN = /^[A-Z][A-Z0-9_]{0,63}$/; export const MEDIA_USAGE_WORK_OPERATOR_DEFAULT_LIMIT = 50; export const MEDIA_USAGE_WORK_OPERATOR_MAX_LIMIT = 100; +export const MEDIA_USAGE_RECONCILIATION_PAGE_LIMIT = 50; const MEDIA_USAGE_WORK_STATES = ["pending", "retry", "leased", "failed"] as const; export interface MediaUsageWorkIdentity { @@ -63,6 +64,92 @@ export type MediaUsageOperatorRetryResult = export class MediaUsageWorkRepository { constructor(private db: Kysely) {} + async enqueueReconciliationPage(input: { + collectionId: string; + collectionSlug: string; + runToken: string; + leaseToken: string; + changeEpoch: number | string; + phase: "scan" | "sources"; + contentIds: readonly string[]; + }): Promise { + if (!input.collectionId || !input.collectionSlug || !input.runToken || !input.leaseToken) { + throw new Error("Reconciliation enqueue requires exact collection and lease identity"); + } + assertNonNegativeDecimal(input.changeEpoch, "change epoch"); + const contentIds = [...new Set(input.contentIds)]; + if (contentIds.length < 1 || contentIds.length > MEDIA_USAGE_RECONCILIATION_PAGE_LIMIT) { + throw new Error("Reconciliation enqueue requires from 1 to 50 content IDs"); + } + if (contentIds.some((contentId) => !contentId)) { + throw new Error("Reconciliation enqueue content IDs must not be empty"); + } + const now = this.timestampOffset(0); + const pageValues = sql.join( + contentIds.map((contentId) => sql`(${contentId})`), + sql`, `, + ); + await sql` + WITH page(content_id) AS (VALUES ${pageValues}) + INSERT INTO _emdash_media_usage_work ( + collection_id, collection_slug, content_id, change_epoch, work_version, + state, attempt_count, next_attempt_at, lease_token, lease_expires_at, + last_attempted_at, last_error_code, created_at, updated_at + ) + SELECT + ${input.collectionId}, ${input.collectionSlug}, page.content_id, ${input.changeEpoch}, 1, + 'pending', 0, ${now}, NULL, NULL, NULL, NULL, ${now}, ${now} + FROM page + WHERE EXISTS ( + SELECT 1 + FROM _emdash_media_usage_reconciliations AS reconciliation + INNER JOIN _emdash_media_usage_index_status AS status + ON status.collection_id = reconciliation.collection_id + AND status.scope_key = reconciliation.collection_slug + INNER JOIN _emdash_collections AS collection + ON collection.id = reconciliation.collection_id + AND collection.slug = reconciliation.collection_slug + WHERE reconciliation.collection_id = ${input.collectionId} + AND reconciliation.collection_slug = ${input.collectionSlug} + AND reconciliation.run_token = ${input.runToken} + AND reconciliation.target_epoch = ${input.changeEpoch} + AND reconciliation.state = 'leased' + AND reconciliation.phase = ${input.phase} + AND reconciliation.lease_token = ${input.leaseToken} + AND ${this.qualifiedLeaseIsLive("reconciliation.lease_expires_at")} + AND status.adapter_id = 'content-media' + AND status.scope_type = 'collection' + AND status.capture_state = 'active' + AND status.reconciliation_required = 1 + AND status.status = 'running' + AND status.cursor = ${input.runToken} + AND status.change_epoch = ${input.changeEpoch} + AND EXISTS ( + SELECT 1 FROM _emdash_media_usage_activation AS activation + WHERE activation.task_key = 'incremental_capture' + AND activation.state = 'active' + ) + AND NOT EXISTS ( + SELECT 1 FROM _emdash_media_usage_collection_deletions AS deletion + WHERE deletion.collection_id = reconciliation.collection_id + ) + ) + ON CONFLICT (collection_id, content_id) DO UPDATE SET + collection_slug = excluded.collection_slug, + change_epoch = excluded.change_epoch, + work_version = _emdash_media_usage_work.work_version + 1, + state = 'pending', + attempt_count = 0, + next_attempt_at = excluded.next_attempt_at, + lease_token = NULL, + lease_expires_at = NULL, + last_attempted_at = NULL, + last_error_code = NULL, + updated_at = excluded.updated_at + WHERE _emdash_media_usage_work.change_epoch < excluded.change_epoch + `.execute(this.db); + } + async findOperatorPage(options: { collectionSlug: string; state?: MediaUsageWorkState; @@ -604,6 +691,13 @@ export class MediaUsageWorkRepository { : sql`lease_expires_at > strftime('%Y-%m-%dT%H:%M:%fZ', 'now')`; } + private qualifiedLeaseIsLive(column: string): RawBuilder { + const expiry = sql.ref(column); + return isPostgres(this.db) + ? sql`${expiry} > to_char(statement_timestamp() AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.MS"Z"')` + : sql`${expiry} > strftime('%Y-%m-%dT%H:%M:%fZ', 'now')`; + } + private timestampIsDue( column: "next_attempt_at" | "lease_expires_at" | "work.lease_expires_at", ): RawBuilder { diff --git a/packages/core/src/database/repositories/media-usage.ts b/packages/core/src/database/repositories/media-usage.ts index 3561a9dd28..49ccb3f6f4 100644 --- a/packages/core/src/database/repositories/media-usage.ts +++ b/packages/core/src/database/repositories/media-usage.ts @@ -1723,19 +1723,31 @@ export class MediaUsageRepository { }, ): Promise { const now = this.sortableUtcTimestamp(); + const automaticRunOwnsCoverage = sql`EXISTS ( + SELECT 1 + FROM _emdash_media_usage_reconciliations AS reconciliation + WHERE reconciliation.collection_id = ${input.collectionId} + AND reconciliation.run_token = cursor + )`; const result = await this.db .updateTable("_emdash_media_usage_index_status") .set({ status: sql`CASE + WHEN ${automaticRunOwnsCoverage} THEN status WHEN reconciliation_required = 0 THEN 'partial' WHEN status = 'running' THEN 'stale' ELSE status END`, completed_at: sql`CASE + WHEN ${automaticRunOwnsCoverage} THEN completed_at WHEN reconciliation_required = 0 OR status = 'running' THEN NULL ELSE completed_at END`, - cursor: sql`CASE WHEN status = 'running' THEN NULL ELSE cursor END`, + cursor: sql`CASE + WHEN ${automaticRunOwnsCoverage} THEN cursor + WHEN status = 'running' THEN NULL + ELSE cursor + END`, last_error_code: input.errorCode, updated_at: now, }) @@ -2335,6 +2347,7 @@ export class MediaUsageRepository { AND ${this.generationWriteLeaseExpiryIsInFuture("expires_at")} ) AND ${this.currentCollectionExists(row.collection_id, row.collection_slug)} + AND ${this.currentCanonicalContentExists(row)} ${conflict} `.execute(db); return Number(result.numAffectedRows ?? 0) > 0; @@ -2436,6 +2449,7 @@ export class MediaUsageRepository { .where("current_generation", "=", expectedCurrentGeneration) .where(this.generationWriteLeaseExpression(row, leaseToken)) .where(this.currentCollectionExists(row.collection_id, row.collection_slug)) + .where(this.currentCanonicalContentExists(row)) .executeTakeFirst(); return Number(result.numUpdatedRows ?? 0) > 0; } @@ -2453,6 +2467,7 @@ export class MediaUsageRepository { .where(this.sourceMatchExpression(expectedSource)) .where(this.generationWriteLeaseExpression(row, leaseToken)) .where(this.currentCollectionExists(row.collection_id, row.collection_slug)) + .where(this.currentCanonicalContentExists(row)) .executeTakeFirst(); return Number(result.numUpdatedRows ?? 0) > 0; } @@ -2469,6 +2484,7 @@ export class MediaUsageRepository { .where("source_key", "=", row.source_key) .where(this.sourceMatchExpression(expectedSource)) .where(this.currentCollectionExists(row.collection_id, row.collection_slug)) + .where(this.currentCanonicalContentExists(row)) .executeTakeFirst(); return Number(result.numUpdatedRows ?? 0) > 0; } @@ -2557,6 +2573,47 @@ export class MediaUsageRepository { )`; } + private currentCanonicalContentExists( + row: + | ReturnType + | ReturnType, + ): RawBuilder { + if (row.collection_id === null || row.identity_version !== 1 || row.source_type !== "content") { + return sql`1 = 1`; + } + if ( + !row.collection_slug || + !row.content_id || + row.source_version === null || + row.source_updated_at === null + ) { + return sql`1 = 0`; + } + validateIdentifier(row.collection_slug, "collection slug"); + const tableName = `ec_${row.collection_slug}`; + validateIdentifier(tableName, "content table"); + const revisionColumn = + row.source_variant === "columns" + ? "live_revision_id" + : row.source_variant === "draft_overlay" + ? "draft_revision_id" + : null; + if (!revisionColumn) return sql`1 = 0`; + const revision = sql.ref(`content.${revisionColumn}`); + const revisionMatches = + row.revision_id === null + ? sql`${revision} IS NULL` + : sql`${revision} = ${row.revision_id}`; + return sql`EXISTS ( + SELECT 1 + FROM ${sql.ref(tableName)} AS content + WHERE content.id = ${row.content_id} + AND content.version = ${row.source_version} + AND content.updated_at = ${row.source_updated_at} + AND ${revisionMatches} + )`; + } + private nullableNumberExpression( eb: ExpressionBuilder, column: "source_version" | "identity_version", diff --git a/packages/core/src/database/types.ts b/packages/core/src/database/types.ts index 9511f11e89..71f18fbf97 100644 --- a/packages/core/src/database/types.ts +++ b/packages/core/src/database/types.ts @@ -216,6 +216,7 @@ export interface MediaUsageActivationTable { activated_at: Generated; created_at: Generated; updated_at: Generated; + media_usage_maintenance_turn: Generated; } export interface MediaUsageWorkTable { @@ -253,6 +254,27 @@ export interface MediaUsageCollectionDeletionTable { updated_at: Generated; } +export interface MediaUsageReconciliationTable { + collection_id: string; + collection_slug: string; + run_token: string; + target_epoch: Generated; + field_fingerprint: Generated; + state: Generated; + phase: Generated; + scan_cursor: Generated; + scan_upper_id: Generated; + source_cursor: Generated; + source_upper_key: Generated; + attempt_count: Generated; + next_attempt_at: string; + lease_token: Generated; + lease_expires_at: Generated; + last_error_code: Generated; + created_at: Generated; + updated_at: Generated; +} + export interface UserTable { id: string; email: string; @@ -623,6 +645,7 @@ export interface Database { _emdash_media_usage_activation: MediaUsageActivationTable; _emdash_media_usage_work: MediaUsageWorkTable; _emdash_media_usage_collection_deletions: MediaUsageCollectionDeletionTable; + _emdash_media_usage_reconciliations: MediaUsageReconciliationTable; users: UserTable; credentials: CredentialTable; auth_tokens: AuthTokenTable; diff --git a/packages/core/src/emdash-runtime.ts b/packages/core/src/emdash-runtime.ts index db67f93563..5d02c4386a 100644 --- a/packages/core/src/emdash-runtime.ts +++ b/packages/core/src/emdash-runtime.ts @@ -9,7 +9,7 @@ import { Permissions } from "@emdash-cms/auth"; import type { Element } from "@emdash-cms/blocks"; -import { Kysely, type Dialect } from "kysely"; +import { Kysely, sql, type Dialect } from "kysely"; import virtualConfig from "virtual:emdash/config"; import { z } from "zod"; @@ -42,7 +42,10 @@ import { getI18nConfig } from "./i18n/config.js"; import { repairLocaleCasing } from "./i18n/repair-locale-casing.js"; import { normalizeMediaValue } from "./media/normalize.js"; import type { MediaProvider, MediaProviderCapabilities } from "./media/types.js"; -import { processDueMediaUsageCollectionDeletions } from "./media/usage/collection-deletion-processor.js"; +import { + MEDIA_USAGE_COLLECTION_DELETION_LIMITS, + processDueMediaUsageCollectionDeletions, +} from "./media/usage/collection-deletion-processor.js"; import { deleteContentMediaUsage, findNonTranslatableSiblingContentIds, @@ -50,6 +53,11 @@ import { refreshContentMediaUsageAfterWrite, } from "./media/usage/content-refresh.js"; import { + MEDIA_USAGE_RECONCILIATION_LIMITS, + processDueMediaUsageReconciliation, +} from "./media/usage/reconciliation-processor.js"; +import { + MEDIA_USAGE_WORK_PROCESSING_LIMITS, processDueMediaUsageWork, processMediaUsageWorkAfterWrite, } from "./media/usage/work-processor.js"; @@ -532,23 +540,60 @@ const marketplaceManifestCache = new Map< const sandboxedRouteMetaCache = new Map>(); let sandboxRunner: SandboxRunner | null = null; -async function runScheduledMediaUsageWork(db: Kysely): Promise { - try { - const result = await processDueMediaUsageWork(db); - if (result.candidateCount > 0) { - console.info("[media-usage:work] Scheduled processing", result); - } - } catch (error) { - console.error("[media-usage:work] Scheduled processing failed:", error); +export const MEDIA_USAGE_MAINTENANCE_QUERY_RESERVATIONS = Object.freeze({ + entryWork: MEDIA_USAGE_WORK_PROCESSING_LIMITS.ordinaryStatementsPerJob, + collectionDeletion: MEDIA_USAGE_COLLECTION_DELETION_LIMITS.maxQueriesPerTick, + reconciliation: MEDIA_USAGE_RECONCILIATION_LIMITS.maxQueriesPerTick, + maxClassQueries: Math.max( + MEDIA_USAGE_WORK_PROCESSING_LIMITS.ordinaryStatementsPerJob, + MEDIA_USAGE_COLLECTION_DELETION_LIMITS.maxQueriesPerTick, + MEDIA_USAGE_RECONCILIATION_LIMITS.maxQueriesPerTick, + ), + eventCeiling: 40, +}); + +export type MediaUsageMaintenanceTaskClass = + | "entry_work" + | "collection_deletion" + | "reconciliation"; + +export type MediaUsageMaintenanceResult = + | { outcome: "inactive" | "admission_closed"; taskClass: null; turn: null } + | { outcome: "processed"; taskClass: MediaUsageMaintenanceTaskClass; turn: number }; + +async function runScheduledMediaUsageLane( + db: Kysely, +): Promise { + const queriesAlreadySpent = getRequestContext()?.metrics?.dbCount ?? 0; + if ( + queriesAlreadySpent + 1 + MEDIA_USAGE_MAINTENANCE_QUERY_RESERVATIONS.maxClassQueries > + MEDIA_USAGE_MAINTENANCE_QUERY_RESERVATIONS.eventCeiling + ) { + return { outcome: "admission_closed", taskClass: null, turn: null }; } - try { - const result = await processDueMediaUsageCollectionDeletions(db); - if (result.candidateCount > 0) { - console.info("[media-usage:collection-deletion] Scheduled processing", result); - } - } catch (error) { - console.error("[media-usage:collection-deletion] Scheduled processing failed:", error); + + 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 { outcome: "inactive", taskClass: null, turn: null }; + + const turn = activation.media_usage_maintenance_turn; + if (turn === 0) { + await processDueMediaUsageWork(db); + return { outcome: "processed", taskClass: "entry_work", turn }; } + if (turn === 1) { + await processDueMediaUsageCollectionDeletions(db); + return { outcome: "processed", taskClass: "collection_deletion", turn }; + } + await processDueMediaUsageReconciliation(db); + return { outcome: "processed", taskClass: "reconciliation", turn }; } /** @@ -743,7 +788,6 @@ export class EmDashRuntime { console.error("[cleanup] System cleanup failed:", error); } - await runScheduledMediaUsageWork(this.db); try { await this.syncPluginStorageIndexesOnce(); } catch (error) { @@ -756,6 +800,10 @@ export class EmDashRuntime { return { published }; } + async runScheduledMediaUsageTasks(): Promise { + return runScheduledMediaUsageLane(this.db); + } + /** * Materialize plugin-declared storage indexes, once per process. * @@ -1675,6 +1723,14 @@ export class EmDashRuntime { if (deps.createScheduler) { const scheduler = deps.createScheduler(cronExecutor); cronScheduler = scheduler; + const runMediaUsageMaintenance = async () => { + const runtime = runtimeRef.current; + if (runtime) { + await runtime.runScheduledMediaUsageTasks(); + } else { + await runScheduledMediaUsageLane(db); + } + }; // Run scheduled publishing and system cleanup alongside each tick. // Pass storage so cleanupPendingUploads can delete orphaned files. @@ -1700,7 +1756,6 @@ export class EmDashRuntime { // by runSystemCleanup. This catches unexpected errors. console.error("[cleanup] System cleanup failed:", error); } - await runScheduledMediaUsageWork(db); try { await runtimeRef.current?.syncPluginStorageIndexesOnce(); } catch (error) { @@ -1708,7 +1763,15 @@ export class EmDashRuntime { } // Never throws; no-op unless scheduled backups are enabled and due. await maybeRunScheduledBackup(db, storage ?? undefined); + if (!scheduler.setMediaUsageMaintenance) { + try { + await runMediaUsageMaintenance(); + } catch (error) { + console.error("[media-usage] Scheduled maintenance failed:", error); + } + } }); + 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/media/usage/collection-deletion-processor.ts b/packages/core/src/media/usage/collection-deletion-processor.ts index 99749ee5ce..f0adb7ce98 100644 --- a/packages/core/src/media/usage/collection-deletion-processor.ts +++ b/packages/core/src/media/usage/collection-deletion-processor.ts @@ -18,6 +18,7 @@ export const MEDIA_USAGE_COLLECTION_DELETION_LIMITS = Object.freeze({ maxAttempts: 5, retryBaseSeconds: 30, retryMaxSeconds: 15 * 60, + maxQueriesPerTick: 30, }); export interface MediaUsageCollectionDeletionTickResult { @@ -221,6 +222,12 @@ async function processStatus( if (await exactCleanupRowsRemain(trx, claim, false)) { throw new Error("Collection deletion cleanup is incomplete"); } + await trx + .deleteFrom("_emdash_media_usage_reconciliations") + .where("collection_id", "=", claim.collectionId) + .where("collection_slug", "=", claim.collectionSlug) + .where(liveLeaseGuard(trx, claim)) + .execute(); await trx .deleteFrom("_emdash_media_usage_index_status") .where("adapter_id", "=", "content-media") diff --git a/packages/core/src/media/usage/content-fields.ts b/packages/core/src/media/usage/content-fields.ts index 9dc5146a74..82afa3d5c2 100644 --- a/packages/core/src/media/usage/content-fields.ts +++ b/packages/core/src/media/usage/content-fields.ts @@ -2,7 +2,9 @@ import type { Kysely } from "kysely"; import type { Database } from "../../database/types.js"; import { validateIdentifier } from "../../database/validate.js"; +import { buildCanonicalSha256Fingerprint } from "./projection-fingerprint.js"; import type { MediaUsageExtractionField, MediaUsageExtractionSubField } from "./types.js"; +import { CONTENT_SOURCE_SCHEMA_VERSION } from "./types.js"; export type ContentMediaUsageField = MediaUsageExtractionField; @@ -11,6 +13,31 @@ export interface ContentMediaUsageFieldDiscovery { displayFieldSlugs: string[]; } +export async function buildContentMediaUsageFieldFingerprint( + discovery: ContentMediaUsageFieldDiscovery, +): Promise { + const extractionFields = discovery.extractionFields + .map((field) => ({ + slug: field.slug, + type: field.type, + ...(field.type === "repeater" + ? { + subFields: (field.validation?.subFields ?? []) + .map((subField) => ({ slug: subField.slug, type: subField.type })) + .toSorted(compareFieldIdentity), + } + : {}), + })) + .toSorted(compareFieldIdentity); + const result = await buildCanonicalSha256Fingerprint("media-usage-fields:v1:sha256:", { + fingerprintVersion: 1, + contentSourceSchemaVersion: CONTENT_SOURCE_SCHEMA_VERSION, + extractionFields, + displayFieldSlugs: discovery.displayFieldSlugs.toSorted(compareStrings), + }); + return result.fingerprint; +} + export class MediaUsageFieldDiscoveryError extends Error { constructor( message: string, @@ -119,3 +146,11 @@ function isSupportedTopLevelType(value: string): value is SupportedTopLevelType function isRecord(value: unknown): value is Record { return typeof value === "object" && value !== null && !Array.isArray(value); } + +function compareFieldIdentity(a: { slug: string }, b: { slug: string }): number { + return compareStrings(a.slug, b.slug); +} + +function compareStrings(a: string, b: string): number { + return a < b ? -1 : a > b ? 1 : 0; +} diff --git a/packages/core/src/media/usage/content-snapshots.ts b/packages/core/src/media/usage/content-snapshots.ts index 0011ab2bde..6757a3375c 100644 --- a/packages/core/src/media/usage/content-snapshots.ts +++ b/packages/core/src/media/usage/content-snapshots.ts @@ -17,8 +17,9 @@ import { buildContentMediaUsageSourceKey, type MediaUsageContentSourceVariant, } from "./source-key.js"; +import { CONTENT_SOURCE_SCHEMA_VERSION } from "./types.js"; -export const CONTENT_SOURCE_SCHEMA_VERSION = 1; +export { CONTENT_SOURCE_SCHEMA_VERSION } from "./types.js"; const CONTENT_COLLECTION_ID_RESULT = "__emdash_media_usage_collection_id"; const CONTENT_SYSTEM_COLUMNS = [ diff --git a/packages/core/src/media/usage/projection-fingerprint.ts b/packages/core/src/media/usage/projection-fingerprint.ts index bb40063e9b..d3eb16e1a8 100644 --- a/packages/core/src/media/usage/projection-fingerprint.ts +++ b/packages/core/src/media/usage/projection-fingerprint.ts @@ -41,7 +41,7 @@ export async function buildMediaUsageProjectionFingerprint( .map((occurrence) => ({ occurrence, key: canonicalJson(occurrence) })) .toSorted((a, b) => compareCanonicalStrings(a.key, b.key)) .map(({ occurrence }) => occurrence); - const payload = canonicalJson({ + return buildCanonicalSha256Fingerprint(FINGERPRINT_PREFIX, { fingerprintVersion: MEDIA_USAGE_PROJECTION_FINGERPRINT_VERSION, collectionId: input.collectionId, extractionSchema: normalizeExtractionFields(input.extractionFields), @@ -64,13 +64,19 @@ export async function buildMediaUsageProjectionFingerprint( }, occurrences: canonicalOccurrences, }); - const encodedPayload = new TextEncoder().encode(payload); +} + +export async function buildCanonicalSha256Fingerprint( + prefix: string, + payload: unknown, +): Promise { + const encodedPayload = new TextEncoder().encode(canonicalJson(payload)); const digest = await crypto.subtle.digest("SHA-256", encodedPayload); const hex = Array.from(new Uint8Array(digest), (byte) => byte.toString(16).padStart(2, "0")).join( "", ); return { - fingerprint: `${FINGERPRINT_PREFIX}${hex}`, + fingerprint: `${prefix}${hex}`, byteLength: encodedPayload.byteLength, }; } diff --git a/packages/core/src/media/usage/reconciliation-processor.ts b/packages/core/src/media/usage/reconciliation-processor.ts new file mode 100644 index 0000000000..3bf944f437 --- /dev/null +++ b/packages/core/src/media/usage/reconciliation-processor.ts @@ -0,0 +1,372 @@ +import type { Kysely } from "kysely"; + +import { MediaUsageWorkRepository } from "../../database/repositories/media-usage-work.js"; +import type { Database } from "../../database/types.js"; +import { + buildContentMediaUsageFieldFingerprint, + loadContentMediaUsageFields, +} from "./content-fields.js"; +import { + MediaUsageReconciliationRepository, + type MediaUsageReconciliationClaim, + type MediaUsageReconciliationRecord, +} from "./reconciliation.js"; +import { CONTENT_SOURCE_SCHEMA_VERSION } from "./types.js"; + +export const MEDIA_USAGE_RECONCILIATION_LIMITS = Object.freeze({ + candidatesPerTick: 4, + pageSize: 50, + leaseDurationSeconds: 60, + maxAttempts: 5, + retryBaseSeconds: 30, + retryMaxSeconds: 15 * 60, + retryJitterRatio: 0.25, + maxQueriesPerTick: 20, +}); + +export type MediaUsageReconciliationOutcome = + | "inactive" + | "not_due" + | "claim_lost" + | "advanced" + | "deferred" + | "completed" + | "retry" + | "failed"; + +export type MediaUsageReconciliationScanOutcome = + | "advanced" + | "exhausted" + | "deferred" + | "restart_required"; + +export async function processDueMediaUsageReconciliation( + 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"; + + const reconciliation = new MediaUsageReconciliationRepository(db); + if (await reconciliation.deleteOneObsolete()) return "completed"; + const [failed] = await reconciliation.findFailed(1); + if (failed) { + if (await reconciliation.finishFailedCoverage(failed.collectionId, failed.runToken)) { + return "failed"; + } + if (await reconciliation.resetFailedForNewEpoch(failed)) return "advanced"; + } + + await reconciliation.seedNextCandidate(); + const candidates = await reconciliation.findDue( + MEDIA_USAGE_RECONCILIATION_LIMITS.candidatesPerTick, + ); + let claim: MediaUsageReconciliationClaim | null = null; + for (const candidate of candidates) { + claim = await reconciliation.claim({ + collectionId: candidate.collectionId, + runToken: candidate.runToken, + leaseDurationSeconds: MEDIA_USAGE_RECONCILIATION_LIMITS.leaseDurationSeconds, + }); + if (claim) break; + } + if (!claim) return candidates.length === 0 ? "not_due" : "claim_lost"; + + try { + return await processClaimedReconciliation(db, claim); + } catch (error) { + const terminal = claim.attemptCount + 1 >= MEDIA_USAGE_RECONCILIATION_LIMITS.maxAttempts; + const recorded = await reconciliation.recordFailure({ + collectionId: claim.collectionId, + runToken: claim.runToken, + leaseToken: claim.leaseToken, + errorCode: "MEDIA_USAGE_RECONCILIATION_FAILED", + retryDelaySeconds: retryDelaySeconds(claim.attemptCount), + terminal, + }); + if (!recorded) return "claim_lost"; + if (terminal) await reconciliation.finishFailedCoverage(claim.collectionId, claim.runToken); + console.error("[media-usage:reconciliation] Processing failed:", error); + return terminal ? "failed" : "retry"; + } +} + +export async function processClaimedMediaUsageReconciliationScan( + db: Kysely, + claim: MediaUsageReconciliationClaim, + options: { releaseOnExhausted?: boolean } = {}, +): Promise { + const reconciliation = new MediaUsageReconciliationRepository(db); + let current = await reconciliation.findByIdentity(claim.collectionId, claim.runToken); + if (!current || current.leaseToken !== claim.leaseToken || current.phase !== "scan") { + return "deferred"; + } + + let fields; + let fieldFingerprint: string; + if (current.targetEpoch === null) { + const targetEpoch = await reconciliation.beginRun(claim); + if (targetEpoch === null) { + await reconciliation.release({ ...claim, delaySeconds: 30 }); + return "deferred"; + } + fields = await loadContentMediaUsageFields(db, claim.collectionSlug, claim.collectionId); + fieldFingerprint = await buildContentMediaUsageFieldFingerprint(fields); + const scanUpperId = + fields.extractionFields.length === 0 ? null : await reconciliation.findScanUpperId(claim); + if ( + !(await reconciliation.initializeScan({ + claim, + targetEpoch, + fieldFingerprint, + scanUpperId, + })) + ) { + return "deferred"; + } + current = await reconciliation.findByIdentity(claim.collectionId, claim.runToken); + if (!current) return "deferred"; + } else { + fields = await loadContentMediaUsageFields(db, claim.collectionSlug, claim.collectionId); + fieldFingerprint = await buildContentMediaUsageFieldFingerprint(fields); + } + + if (current.fieldFingerprint !== fieldFingerprint || current.targetEpoch === null) { + return "restart_required"; + } + const contentIds = await reconciliation.findScanPage(current, 50); + if (contentIds.length === 0) { + if (options.releaseOnExhausted ?? true) { + await reconciliation.release({ ...claim, delaySeconds: 30 }); + } + return "exhausted"; + } + + const work = new MediaUsageWorkRepository(db); + await work.enqueueReconciliationPage({ + collectionId: claim.collectionId, + collectionSlug: claim.collectionSlug, + runToken: claim.runToken, + leaseToken: claim.leaseToken, + changeEpoch: current.targetEpoch, + phase: "scan", + contentIds, + }); + const nextCursor = contentIds.at(-1)!; + if ( + !(await reconciliation.checkpointScan({ + claim, + targetEpoch: current.targetEpoch, + previousCursor: current.scanCursor, + nextCursor, + })) + ) { + return "deferred"; + } + if (!(await reconciliation.release({ ...claim, delaySeconds: 0 }))) return "deferred"; + return "advanced"; +} + +async function processClaimedReconciliation( + db: Kysely, + claim: MediaUsageReconciliationClaim, +): Promise { + const reconciliation = new MediaUsageReconciliationRepository(db); + let current = await reconciliation.findByIdentity(claim.collectionId, claim.runToken); + if (!current || current.leaseToken !== claim.leaseToken) return "claim_lost"; + if (current.targetEpoch !== null && !(await reconciliation.ownsRun(claim, current.targetEpoch))) { + return restartReconciliation(db, reconciliation, claim, current); + } + + if (current.phase === "scan") { + const outcome = await processClaimedMediaUsageReconciliationScan(db, claim, { + releaseOnExhausted: false, + }); + if (outcome === "restart_required") { + current = + (await reconciliation.findByIdentity(claim.collectionId, claim.runToken)) ?? current; + return restartReconciliation(db, reconciliation, claim, current); + } + if (outcome !== "exhausted") return outcome; + current = (await reconciliation.findByIdentity(claim.collectionId, claim.runToken)) ?? current; + const barrier = await reconciliation.findWorkBarrier(claim.collectionId); + if (barrier.state === "failed") { + return failReconciliation(reconciliation, claim, barrier.errorCode, true); + } + if (barrier.state === "pending") { + await reconciliation.release({ ...claim, delaySeconds: 30 }); + return "deferred"; + } + if (current.targetEpoch === null || current.fieldFingerprint === null) return "claim_lost"; + const sourceUpperKey = await reconciliation.findSourceUpperKey(claim, current.targetEpoch); + if ( + !(await reconciliation.transitionToSources({ + claim, + targetEpoch: current.targetEpoch, + fieldFingerprint: current.fieldFingerprint, + sourceUpperKey, + })) + ) { + return "claim_lost"; + } + if (!(await reconciliation.release({ ...claim, delaySeconds: 0 }))) return "claim_lost"; + return "advanced"; + } + + return processSourcePhase(db, reconciliation, claim, current); +} + +async function processSourcePhase( + db: Kysely, + reconciliation: MediaUsageReconciliationRepository, + claim: MediaUsageReconciliationClaim, + current: MediaUsageReconciliationRecord, +): Promise { + if (current.targetEpoch === null || current.fieldFingerprint === null) return "claim_lost"; + const fields = await loadContentMediaUsageFields(db, claim.collectionSlug, claim.collectionId); + const fieldFingerprint = await buildContentMediaUsageFieldFingerprint(fields); + if (fieldFingerprint !== current.fieldFingerprint) { + return restartReconciliation(db, reconciliation, claim, current, fields, fieldFingerprint); + } + + const page = await reconciliation.findSourcePage( + current, + MEDIA_USAGE_RECONCILIATION_LIMITS.pageSize, + ); + if (page.length > 0) { + const malformed = page.some( + (source) => + !source.contentId || + (source.sourceVariant !== "columns" && source.sourceVariant !== "draft_overlay"), + ); + if (malformed) { + return failReconciliation( + reconciliation, + claim, + "MEDIA_USAGE_RECONCILIATION_INVALID_SOURCE", + false, + ); + } + const contentIds = [...new Set(page.map((source) => source.contentId!))]; + const enqueueIds = + fields.extractionFields.length === 0 + ? contentIds + : await reconciliation.findMissingContentIds(claim.collectionSlug, contentIds); + if (enqueueIds.length > 0) { + await new MediaUsageWorkRepository(db).enqueueReconciliationPage({ + collectionId: claim.collectionId, + collectionSlug: claim.collectionSlug, + runToken: claim.runToken, + leaseToken: claim.leaseToken, + changeEpoch: current.targetEpoch, + phase: "sources", + contentIds: enqueueIds, + }); + } + if ( + !(await reconciliation.checkpointSources({ + claim, + targetEpoch: current.targetEpoch, + previousCursor: current.sourceCursor, + nextCursor: page.at(-1)!.sourceKey, + })) + ) { + return "claim_lost"; + } + if (!(await reconciliation.release({ ...claim, delaySeconds: 0 }))) return "claim_lost"; + return "advanced"; + } + + const barrier = await reconciliation.findWorkBarrier(claim.collectionId); + if (barrier.state === "failed") { + return failReconciliation(reconciliation, claim, barrier.errorCode, true); + } + if (barrier.state === "pending") { + await reconciliation.release({ ...claim, delaySeconds: 30 }); + return "deferred"; + } + if ( + !(await reconciliation.finalizeCoverage({ + claim, + targetEpoch: current.targetEpoch, + fieldFingerprint, + schemaVersion: CONTENT_SOURCE_SCHEMA_VERSION, + })) + ) { + return "claim_lost"; + } + if (!(await reconciliation.deleteFinalized(claim))) return "claim_lost"; + return "completed"; +} + +async function restartReconciliation( + db: Kysely, + reconciliation: MediaUsageReconciliationRepository, + claim: MediaUsageReconciliationClaim, + current: MediaUsageReconciliationRecord, + fields?: Awaited>, + fieldFingerprint?: string, +): Promise { + if (current.targetEpoch === null) return "claim_lost"; + const discoveredFields = + fields ?? (await loadContentMediaUsageFields(db, claim.collectionSlug, claim.collectionId)); + const fingerprint = + fieldFingerprint ?? (await buildContentMediaUsageFieldFingerprint(discoveredFields)); + const targetEpoch = await reconciliation.restartRun(claim, current.targetEpoch); + if (targetEpoch === null) { + await reconciliation.release({ ...claim, delaySeconds: 30 }); + return "deferred"; + } + const scanUpperId = + discoveredFields.extractionFields.length === 0 + ? null + : await reconciliation.findScanUpperId(claim); + if ( + !(await reconciliation.restartScan({ + claim, + previousEpoch: current.targetEpoch, + targetEpoch, + fieldFingerprint: fingerprint, + scanUpperId, + })) + ) { + return "claim_lost"; + } + if (!(await reconciliation.release({ ...claim, delaySeconds: 0 }))) return "claim_lost"; + return "advanced"; +} + +async function failReconciliation( + reconciliation: MediaUsageReconciliationRepository, + claim: MediaUsageReconciliationClaim, + errorCode: string, + entryFailure: boolean, +): Promise { + const recorded = entryFailure + ? await reconciliation.recordEntryFailure(claim) + : await reconciliation.recordFailure({ + collectionId: claim.collectionId, + runToken: claim.runToken, + leaseToken: claim.leaseToken, + errorCode, + retryDelaySeconds: 0, + terminal: true, + }); + if (!recorded) return "claim_lost"; + await reconciliation.finishFailedCoverage(claim.collectionId, claim.runToken); + return "failed"; +} + +function retryDelaySeconds(attemptCount: number): number { + const exponential = Math.min( + MEDIA_USAGE_RECONCILIATION_LIMITS.retryMaxSeconds, + MEDIA_USAGE_RECONCILIATION_LIMITS.retryBaseSeconds * 2 ** attemptCount, + ); + const jitter = Math.floor( + exponential * MEDIA_USAGE_RECONCILIATION_LIMITS.retryJitterRatio * Math.random(), + ); + return Math.min(MEDIA_USAGE_RECONCILIATION_LIMITS.retryMaxSeconds, exponential + jitter); +} diff --git a/packages/core/src/media/usage/reconciliation.ts b/packages/core/src/media/usage/reconciliation.ts new file mode 100644 index 0000000000..2fc7bcddcc --- /dev/null +++ b/packages/core/src/media/usage/reconciliation.ts @@ -0,0 +1,1117 @@ +import { sql, type Kysely, type RawBuilder, type Selectable } from "kysely"; +import { ulid } from "ulidx"; + +import { isPostgres } from "../../database/dialect-helpers.js"; +import type { Database, MediaUsageReconciliationTable } from "../../database/types.js"; +import { validateIdentifier } from "../../database/validate.js"; + +const ACTIVATION_KEY = "incremental_capture"; +const CONTENT_ADAPTER_ID = "content-media"; +const COLLECTION_SCOPE = "collection"; +const MAX_CANDIDATES = 100; +const MAX_PORTABLE_DURATION_SECONDS = 365 * 24 * 60 * 60; +const STABLE_ERROR_CODE_PATTERN = /^[A-Z][A-Z0-9_]{0,63}$/; + +export type MediaUsageReconciliationState = "pending" | "retry" | "leased" | "failed"; +export type MediaUsageReconciliationPhase = "scan" | "sources"; + +export interface MediaUsageReconciliationRecord { + collectionId: string; + collectionSlug: string; + runToken: string; + targetEpoch: number | string | null; + fieldFingerprint: string | null; + state: MediaUsageReconciliationState; + phase: MediaUsageReconciliationPhase; + scanCursor: string | null; + scanUpperId: string | null; + sourceCursor: string | null; + sourceUpperKey: string | null; + attemptCount: number; + nextAttemptAt: string; + leaseToken: string | null; + leaseExpiresAt: string | null; + lastErrorCode: string | null; + createdAt: string; + updatedAt: string; +} + +export interface MediaUsageReconciliationClaim extends MediaUsageReconciliationRecord { + leaseToken: string; +} + +export interface MediaUsageReconciliationSourceCandidate { + sourceKey: string; + contentId: string | null; + sourceVariant: string; +} + +export type MediaUsageReconciliationWorkBarrier = + | { state: "empty" } + | { state: "pending" } + | { state: "failed"; errorCode: string }; + +export class MediaUsageReconciliationRepository { + constructor(private db: Kysely) {} + + async findByIdentity( + collectionId: string, + runToken: string, + ): Promise { + assertIdentity({ collectionId, runToken }); + const row = await this.db + .selectFrom("_emdash_media_usage_reconciliations") + .selectAll() + .where("collection_id", "=", collectionId) + .where("run_token", "=", runToken) + .executeTakeFirst(); + return row ? rowToRecord(row) : null; + } + + async beginRun(claim: MediaUsageReconciliationClaim): Promise { + const now = timestampOffset(this.db, 0); + const sameRun = sql`status = 'running' AND cursor = ${claim.runToken}`; + const row = await this.db + .updateTable("_emdash_media_usage_index_status as status") + .set({ + status: "running", + started_at: sql`CASE WHEN ${sameRun} THEN started_at ELSE ${now} END`, + completed_at: null, + cursor: claim.runToken, + indexed_source_count: 0, + failed_source_count: 0, + last_error_code: null, + change_epoch: sql`CASE WHEN ${sameRun} THEN change_epoch ELSE change_epoch + 1 END`, + reconciliation_required: 1, + updated_at: now, + }) + .where("status.adapter_id", "=", CONTENT_ADAPTER_ID) + .where("status.scope_type", "=", COLLECTION_SCOPE) + .where("status.collection_id", "=", claim.collectionId) + .where("status.scope_key", "=", claim.collectionSlug) + .where("status.capture_state", "=", "active") + .where("status.reconciliation_required", "=", 1) + .where((eb) => + eb.or([eb("status.status", "!=", "running"), eb("status.cursor", "=", claim.runToken)]), + ) + .where(this.liveClaimExists(claim)) + .returning("change_epoch") + .executeTakeFirst(); + return row?.change_epoch ?? null; + } + + async findScanUpperId(claim: MediaUsageReconciliationClaim): Promise { + const tableName = contentTableName(claim.collectionSlug); + const result = await sql<{ id: string }>` + SELECT content.id + FROM ${sql.ref(tableName)} AS content + WHERE ${this.liveClaimExistsSql(claim)} + ORDER BY content.id DESC + LIMIT 1 + `.execute(this.db); + return result.rows[0]?.id ?? null; + } + + async initializeScan(input: { + claim: MediaUsageReconciliationClaim; + targetEpoch: number | string; + fieldFingerprint: string; + scanUpperId: string | null; + }): Promise { + const result = await this.db + .updateTable("_emdash_media_usage_reconciliations as reconciliation") + .set({ + target_epoch: input.targetEpoch, + field_fingerprint: input.fieldFingerprint, + phase: "scan", + scan_cursor: null, + scan_upper_id: input.scanUpperId, + source_cursor: null, + source_upper_key: null, + attempt_count: 0, + last_error_code: null, + updated_at: timestampOffset(this.db, 0), + }) + .where("reconciliation.collection_id", "=", input.claim.collectionId) + .where("reconciliation.run_token", "=", input.claim.runToken) + .where("reconciliation.target_epoch", "is", null) + .where("reconciliation.state", "=", "leased") + .where("reconciliation.lease_token", "=", input.claim.leaseToken) + .where(liveLease(this.db)) + .where(this.statusOwnsRun(input.claim, input.targetEpoch)) + .executeTakeFirst(); + return Number(result.numUpdatedRows ?? 0) === 1; + } + + async findScanPage( + reconciliation: MediaUsageReconciliationRecord, + limit: number, + ): Promise { + if (!Number.isSafeInteger(limit) || limit < 1 || limit > 50) { + throw new Error("Reconciliation scan page limit must be from 1 to 50"); + } + if (!reconciliation.leaseToken || reconciliation.targetEpoch === null) return []; + const tableName = contentTableName(reconciliation.collectionSlug); + const lowerBound = reconciliation.scanCursor + ? sql`AND content.id > ${reconciliation.scanCursor}` + : sql``; + const upperBound = reconciliation.scanUpperId + ? sql`AND content.id <= ${reconciliation.scanUpperId}` + : sql`AND 1 = 0`; + const result = await sql<{ id: string }>` + SELECT content.id + FROM ${sql.ref(tableName)} AS content + WHERE 1 = 1 + ${lowerBound} + ${upperBound} + AND ${this.liveClaimExistsSql(reconciliation)} + ORDER BY content.id ASC + LIMIT ${limit} + `.execute(this.db); + return result.rows.map((row) => row.id); + } + + async checkpointScan(input: { + claim: MediaUsageReconciliationClaim; + targetEpoch: number | string; + previousCursor: string | null; + nextCursor: string; + }): Promise { + let query = this.db + .updateTable("_emdash_media_usage_reconciliations as reconciliation") + .set({ + scan_cursor: input.nextCursor, + attempt_count: 0, + last_error_code: null, + updated_at: timestampOffset(this.db, 0), + }) + .where("reconciliation.collection_id", "=", input.claim.collectionId) + .where("reconciliation.run_token", "=", input.claim.runToken) + .where("reconciliation.target_epoch", "=", input.targetEpoch) + .where("reconciliation.state", "=", "leased") + .where("reconciliation.phase", "=", "scan") + .where("reconciliation.lease_token", "=", input.claim.leaseToken) + .where(liveLease(this.db)) + .where(this.statusOwnsRun(input.claim, input.targetEpoch)); + query = input.previousCursor + ? query.where("reconciliation.scan_cursor", "=", input.previousCursor) + : query.where("reconciliation.scan_cursor", "is", null); + const result = await query.executeTakeFirst(); + return Number(result.numUpdatedRows ?? 0) === 1; + } + + async ownsRun( + claim: MediaUsageReconciliationClaim, + targetEpoch: number | string, + ): Promise { + const result = await sql<{ owned: boolean | number }>` + SELECT ${this.statusOwnsRun(claim, targetEpoch)} AS owned + `.execute(this.db); + return Boolean(result.rows[0]?.owned); + } + + async restartRun( + claim: MediaUsageReconciliationClaim, + previousEpoch: number | string, + ): Promise { + const now = timestampOffset(this.db, 0); + const interruptedRestart = sql`status = 'running' + AND cursor = ${claim.runToken} + AND change_epoch > ${previousEpoch}`; + const row = await this.db + .updateTable("_emdash_media_usage_index_status as status") + .set({ + status: "running", + started_at: sql< + string | null + >`CASE WHEN ${interruptedRestart} THEN started_at ELSE ${now} END`, + completed_at: null, + cursor: claim.runToken, + last_error_code: null, + change_epoch: sql`CASE WHEN ${interruptedRestart} THEN change_epoch ELSE change_epoch + 1 END`, + reconciliation_required: 1, + updated_at: now, + }) + .where("status.adapter_id", "=", CONTENT_ADAPTER_ID) + .where("status.scope_type", "=", COLLECTION_SCOPE) + .where("status.collection_id", "=", claim.collectionId) + .where("status.scope_key", "=", claim.collectionSlug) + .where("status.capture_state", "=", "active") + .where("status.reconciliation_required", "=", 1) + .where((eb) => + eb.or([eb("status.status", "!=", "running"), eb("status.cursor", "=", claim.runToken)]), + ) + .where(this.liveClaimExists(claim)) + .returning("change_epoch") + .executeTakeFirst(); + return row?.change_epoch ?? null; + } + + async restartScan(input: { + claim: MediaUsageReconciliationClaim; + previousEpoch: number | string; + targetEpoch: number | string; + fieldFingerprint: string; + scanUpperId: string | null; + }): Promise { + const result = await this.db + .updateTable("_emdash_media_usage_reconciliations as reconciliation") + .set({ + target_epoch: input.targetEpoch, + field_fingerprint: input.fieldFingerprint, + phase: "scan", + scan_cursor: null, + scan_upper_id: input.scanUpperId, + source_cursor: null, + source_upper_key: null, + attempt_count: 0, + next_attempt_at: timestampOffset(this.db, 0), + last_error_code: null, + updated_at: timestampOffset(this.db, 0), + }) + .where("reconciliation.collection_id", "=", input.claim.collectionId) + .where("reconciliation.run_token", "=", input.claim.runToken) + .where("reconciliation.target_epoch", "=", input.previousEpoch) + .where("reconciliation.state", "=", "leased") + .where("reconciliation.lease_token", "=", input.claim.leaseToken) + .where(liveLease(this.db)) + .where(this.statusOwnsRun(input.claim, input.targetEpoch)) + .executeTakeFirst(); + return Number(result.numUpdatedRows ?? 0) === 1; + } + + async findWorkBarrier(collectionId: string): Promise { + if (!collectionId) throw new Error("Reconciliation work barrier requires a collection ID"); + const failed = await this.db + .selectFrom("_emdash_media_usage_work") + .select("last_error_code") + .where("collection_id", "=", collectionId) + .where("state", "=", "failed") + .orderBy("content_id") + .limit(1) + .executeTakeFirst(); + if (failed) { + return { + state: "failed", + errorCode: failed.last_error_code ?? "MEDIA_USAGE_PROCESSING_FAILED", + }; + } + const pending = await this.db + .selectFrom("_emdash_media_usage_work") + .select("content_id") + .where("collection_id", "=", collectionId) + .limit(1) + .executeTakeFirst(); + return pending ? { state: "pending" } : { state: "empty" }; + } + + async findSourceUpperKey( + claim: MediaUsageReconciliationClaim, + targetEpoch: number | string, + ): Promise { + const row = await this.db + .selectFrom("_emdash_media_usage_sources as source") + .select("source.source_key") + .where("source.source_type", "=", "content") + .where("source.collection_id", "=", claim.collectionId) + .where("source.identity_version", "=", 1) + .where(this.liveClaimExists(claim)) + .where(this.statusOwnsRun(claim, targetEpoch)) + .orderBy("source.source_key", "desc") + .limit(1) + .executeTakeFirst(); + return row?.source_key ?? null; + } + + async transitionToSources(input: { + claim: MediaUsageReconciliationClaim; + targetEpoch: number | string; + fieldFingerprint: string; + sourceUpperKey: string | null; + }): Promise { + const result = await this.db + .updateTable("_emdash_media_usage_reconciliations as reconciliation") + .set({ + phase: "sources", + source_cursor: null, + source_upper_key: input.sourceUpperKey, + attempt_count: 0, + last_error_code: null, + updated_at: timestampOffset(this.db, 0), + }) + .where("reconciliation.collection_id", "=", input.claim.collectionId) + .where("reconciliation.run_token", "=", input.claim.runToken) + .where("reconciliation.target_epoch", "=", input.targetEpoch) + .where("reconciliation.field_fingerprint", "=", input.fieldFingerprint) + .where("reconciliation.state", "=", "leased") + .where("reconciliation.phase", "=", "scan") + .where("reconciliation.lease_token", "=", input.claim.leaseToken) + .where(liveLease(this.db)) + .where(this.statusOwnsRun(input.claim, input.targetEpoch)) + .where((eb) => + eb.not( + eb.exists( + eb + .selectFrom("_emdash_media_usage_work as work") + .select("work.content_id") + .where("work.collection_id", "=", input.claim.collectionId), + ), + ), + ) + .executeTakeFirst(); + return Number(result.numUpdatedRows ?? 0) === 1; + } + + async findSourcePage( + reconciliation: MediaUsageReconciliationRecord, + limit: number, + ): Promise { + if (!Number.isSafeInteger(limit) || limit < 1 || limit > 50) { + throw new Error("Reconciliation source page limit must be from 1 to 50"); + } + if (!reconciliation.leaseToken || reconciliation.targetEpoch === null) return []; + let query = this.db + .selectFrom("_emdash_media_usage_sources as source") + .select(["source.source_key", "source.content_id", "source.source_variant"]) + .where("source.source_type", "=", "content") + .where("source.collection_id", "=", reconciliation.collectionId) + .where("source.identity_version", "=", 1) + .where(this.liveClaimExists(reconciliation)) + .where(this.statusOwnsRun(reconciliation, reconciliation.targetEpoch)); + if (reconciliation.sourceCursor) { + query = query.where("source.source_key", ">", reconciliation.sourceCursor); + } + if (reconciliation.sourceUpperKey) { + query = query.where("source.source_key", "<=", reconciliation.sourceUpperKey); + } else { + query = query.where(sql`1 = 0`); + } + const rows = await query.orderBy("source.source_key").limit(limit).execute(); + return rows.map((row) => ({ + sourceKey: row.source_key, + contentId: row.content_id, + sourceVariant: row.source_variant, + })); + } + + async findMissingContentIds( + collectionSlug: string, + contentIds: readonly string[], + ): Promise { + const unique = [...new Set(contentIds)]; + if (unique.length === 0) return []; + if (unique.length > 50 || unique.some((contentId) => !contentId)) { + throw new Error("Reconciliation source page has invalid content identity"); + } + const tableName = contentTableName(collectionSlug); + const existing = await sql<{ id: string }>` + SELECT id FROM ${sql.ref(tableName)} WHERE id IN (${sql.join(unique)}) + `.execute(this.db); + const present = new Set(existing.rows.map((row) => row.id)); + return unique.filter((contentId) => !present.has(contentId)); + } + + async checkpointSources(input: { + claim: MediaUsageReconciliationClaim; + targetEpoch: number | string; + previousCursor: string | null; + nextCursor: string; + }): Promise { + let query = this.db + .updateTable("_emdash_media_usage_reconciliations as reconciliation") + .set({ + source_cursor: input.nextCursor, + attempt_count: 0, + last_error_code: null, + updated_at: timestampOffset(this.db, 0), + }) + .where("reconciliation.collection_id", "=", input.claim.collectionId) + .where("reconciliation.run_token", "=", input.claim.runToken) + .where("reconciliation.target_epoch", "=", input.targetEpoch) + .where("reconciliation.state", "=", "leased") + .where("reconciliation.phase", "=", "sources") + .where("reconciliation.lease_token", "=", input.claim.leaseToken) + .where(liveLease(this.db)) + .where(this.statusOwnsRun(input.claim, input.targetEpoch)); + query = input.previousCursor + ? query.where("reconciliation.source_cursor", "=", input.previousCursor) + : query.where("reconciliation.source_cursor", "is", null); + const result = await query.executeTakeFirst(); + return Number(result.numUpdatedRows ?? 0) === 1; + } + + async finishFailedCoverage(collectionId: string, runToken: string): Promise { + assertIdentity({ collectionId, runToken }); + const now = timestampOffset(this.db, 0); + const result = await this.db + .updateTable("_emdash_media_usage_index_status as status") + .set({ + status: sql`CASE WHEN EXISTS ( + SELECT 1 FROM _emdash_media_usage_sources AS source + WHERE source.source_type = 'content' + AND source.collection_id = ${collectionId} + AND source.identity_version = 1 + LIMIT 1 + ) THEN 'partial' ELSE 'failed' END`, + completed_at: null, + cursor: null, + last_error_code: sql`CASE WHEN ( + SELECT reconciliation.last_error_code + FROM _emdash_media_usage_reconciliations AS reconciliation + WHERE reconciliation.collection_id = ${collectionId} + AND reconciliation.run_token = ${runToken} + ) = 'MEDIA_USAGE_RECONCILIATION_ENTRY_FAILED' + THEN COALESCE( + (SELECT work.last_error_code + FROM _emdash_media_usage_work AS work + WHERE work.collection_id = ${collectionId} AND work.state = 'failed' + ORDER BY work.content_id LIMIT 1), + 'MEDIA_USAGE_RECONCILIATION_ENTRY_FAILED' + ) + ELSE ( + SELECT reconciliation.last_error_code + FROM _emdash_media_usage_reconciliations AS reconciliation + WHERE reconciliation.collection_id = ${collectionId} + AND reconciliation.run_token = ${runToken} + ) END`, + reconciliation_required: 1, + updated_at: now, + }) + .where("status.adapter_id", "=", CONTENT_ADAPTER_ID) + .where("status.scope_type", "=", COLLECTION_SCOPE) + .where("status.collection_id", "=", collectionId) + .where("status.status", "=", "running") + .where("status.cursor", "=", runToken) + .where((eb) => + eb.exists( + eb + .selectFrom("_emdash_media_usage_reconciliations as reconciliation") + .select("reconciliation.collection_id") + .where("reconciliation.collection_id", "=", collectionId) + .where("reconciliation.run_token", "=", runToken) + .where("reconciliation.state", "=", "failed"), + ), + ) + .executeTakeFirst(); + return Number(result.numUpdatedRows ?? 0) === 1; + } + + async finalizeCoverage(input: { + claim: MediaUsageReconciliationClaim; + targetEpoch: number | string; + fieldFingerprint: string; + schemaVersion: number; + }): Promise { + const now = timestampOffset(this.db, 0); + const result = await this.db + .updateTable("_emdash_media_usage_index_status as status") + .set({ + status: "complete", + schema_version: input.schemaVersion, + completed_at: now, + cursor: null, + last_error_code: null, + reconciliation_required: 0, + updated_at: now, + }) + .where("status.adapter_id", "=", CONTENT_ADAPTER_ID) + .where("status.scope_type", "=", COLLECTION_SCOPE) + .where("status.collection_id", "=", input.claim.collectionId) + .where("status.scope_key", "=", input.claim.collectionSlug) + .where("status.capture_state", "=", "active") + .where("status.reconciliation_required", "=", 1) + .where("status.status", "=", "running") + .where("status.cursor", "=", input.claim.runToken) + .where("status.change_epoch", "=", input.targetEpoch) + .where((eb) => + eb.not( + eb.exists( + eb + .selectFrom("_emdash_media_usage_work as work") + .select("work.content_id") + .where("work.collection_id", "=", input.claim.collectionId), + ), + ), + ) + .where((eb) => + eb.exists( + eb + .selectFrom("_emdash_media_usage_reconciliations as reconciliation") + .innerJoin("_emdash_collections as collection", (join) => + join + .onRef("collection.id", "=", "reconciliation.collection_id") + .onRef("collection.slug", "=", "reconciliation.collection_slug"), + ) + .select("reconciliation.collection_id") + .where("reconciliation.collection_id", "=", input.claim.collectionId) + .where("reconciliation.run_token", "=", input.claim.runToken) + .where("reconciliation.target_epoch", "=", input.targetEpoch) + .where("reconciliation.field_fingerprint", "=", input.fieldFingerprint) + .where("reconciliation.state", "=", "leased") + .where("reconciliation.phase", "=", "sources") + .where("reconciliation.lease_token", "=", input.claim.leaseToken) + .where(liveLease(this.db, "reconciliation.lease_expires_at")) + .where((inner) => + inner.not( + inner.exists( + inner + .selectFrom("_emdash_media_usage_collection_deletions as deletion") + .select("deletion.collection_id") + .where("deletion.collection_id", "=", input.claim.collectionId), + ), + ), + ), + ), + ) + .where((eb) => + eb.exists( + eb + .selectFrom("_emdash_media_usage_activation as activation") + .select("activation.task_key") + .where("activation.task_key", "=", ACTIVATION_KEY) + .where("activation.state", "=", "active"), + ), + ) + .executeTakeFirst(); + return Number(result.numUpdatedRows ?? 0) === 1; + } + + async deleteFinalized(claim: MediaUsageReconciliationClaim): Promise { + const result = await this.db + .deleteFrom("_emdash_media_usage_reconciliations as reconciliation") + .where("reconciliation.collection_id", "=", claim.collectionId) + .where("reconciliation.run_token", "=", claim.runToken) + .where("reconciliation.state", "=", "leased") + .where("reconciliation.lease_token", "=", claim.leaseToken) + .where((eb) => + eb.exists( + eb + .selectFrom("_emdash_media_usage_index_status as status") + .select("status.collection_id") + .where("status.collection_id", "=", claim.collectionId) + .where("status.scope_key", "=", claim.collectionSlug) + .where("status.status", "=", "complete") + .where("status.reconciliation_required", "=", 0), + ), + ) + .executeTakeFirst(); + return Number(result.numDeletedRows ?? 0) === 1; + } + + async deleteOneObsolete(): Promise { + const result = await sql<{ collection_id: string }>` + DELETE FROM _emdash_media_usage_reconciliations + WHERE (collection_id, run_token) IN ( + SELECT reconciliation.collection_id, reconciliation.run_token + FROM _emdash_media_usage_reconciliations AS reconciliation + INNER JOIN _emdash_media_usage_index_status AS status + ON status.collection_id = reconciliation.collection_id + AND status.scope_key = reconciliation.collection_slug + WHERE status.adapter_id = ${CONTENT_ADAPTER_ID} + AND status.scope_type = ${COLLECTION_SCOPE} + AND status.reconciliation_required = 0 + ORDER BY reconciliation.updated_at, reconciliation.collection_id + LIMIT 1 + ) + RETURNING collection_id + `.execute(this.db); + return result.rows.length === 1; + } + + private liveClaimExists( + claim: Pick< + MediaUsageReconciliationRecord, + "collectionId" | "collectionSlug" | "runToken" | "leaseToken" + >, + ): RawBuilder { + return this.liveClaimExistsSql(claim); + } + + private liveClaimExistsSql( + claim: Pick< + MediaUsageReconciliationRecord, + "collectionId" | "collectionSlug" | "runToken" | "leaseToken" + >, + ): RawBuilder { + return sql`EXISTS ( + SELECT 1 + FROM _emdash_media_usage_reconciliations AS reconciliation + INNER JOIN _emdash_collections AS collection + ON collection.id = reconciliation.collection_id + AND collection.slug = reconciliation.collection_slug + WHERE reconciliation.collection_id = ${claim.collectionId} + AND reconciliation.collection_slug = ${claim.collectionSlug} + AND reconciliation.run_token = ${claim.runToken} + AND reconciliation.state = 'leased' + AND reconciliation.lease_token = ${claim.leaseToken} + AND ${liveLease(this.db, "reconciliation.lease_expires_at")} + AND EXISTS ( + SELECT 1 FROM _emdash_media_usage_activation AS activation + WHERE activation.task_key = ${ACTIVATION_KEY} + AND activation.state = 'active' + ) + AND NOT EXISTS ( + SELECT 1 FROM _emdash_media_usage_collection_deletions AS deletion + WHERE deletion.collection_id = reconciliation.collection_id + ) + )`; + } + + private statusOwnsRun( + claim: Pick, + targetEpoch: number | string, + ): RawBuilder { + return sql`EXISTS ( + SELECT 1 FROM _emdash_media_usage_index_status AS status + WHERE status.adapter_id = ${CONTENT_ADAPTER_ID} + AND status.scope_type = ${COLLECTION_SCOPE} + AND status.collection_id = ${claim.collectionId} + AND status.scope_key = ${claim.collectionSlug} + AND status.capture_state = 'active' + AND status.reconciliation_required = 1 + AND status.status = 'running' + AND status.cursor = ${claim.runToken} + AND status.change_epoch = ${targetEpoch} + )`; + } + + async seedNextCandidate(): Promise { + const runToken = ulid(); + const now = timestampOffset(this.db, 0); + const result = await sql<{ collection_id: string }>` + INSERT INTO _emdash_media_usage_reconciliations ( + collection_id, + collection_slug, + run_token, + next_attempt_at, + updated_at + ) + SELECT status.collection_id, status.scope_key, ${runToken}, ${now}, ${now} + FROM _emdash_media_usage_index_status AS status + INNER JOIN _emdash_collections AS collection + ON collection.id = status.collection_id + AND collection.slug = status.scope_key + WHERE status.adapter_id = ${CONTENT_ADAPTER_ID} + AND status.scope_type = ${COLLECTION_SCOPE} + AND status.capture_state = 'active' + AND status.reconciliation_required = 1 + AND status.collection_id IS NOT NULL + AND EXISTS ( + SELECT 1 FROM _emdash_media_usage_activation AS activation + WHERE activation.task_key = ${ACTIVATION_KEY} + AND activation.state = 'active' + ) + AND NOT EXISTS ( + SELECT 1 FROM _emdash_media_usage_reconciliations AS existing + WHERE existing.collection_id = status.collection_id + ) + AND NOT EXISTS ( + SELECT 1 FROM _emdash_media_usage_collection_deletions AS deletion + WHERE deletion.collection_id = status.collection_id + ) + ORDER BY status.collection_id + LIMIT 1 + ON CONFLICT (collection_id) DO NOTHING + RETURNING collection_id + `.execute(this.db); + return result.rows.length === 1; + } + + async findDue(limit: number): Promise { + assertLimit(limit); + const nextAttemptIsDue = timestampIsDue(this.db, "next_attempt_at"); + const leaseIsDue = timestampIsDue(this.db, "lease_expires_at"); + const result = await sql>` + WITH pending_candidates AS ( + SELECT * FROM _emdash_media_usage_reconciliations + WHERE state = 'pending' AND ${nextAttemptIsDue} + ORDER BY next_attempt_at, updated_at, collection_id + LIMIT ${limit} + ), retry_candidates AS ( + SELECT * FROM _emdash_media_usage_reconciliations + WHERE state = 'retry' AND ${nextAttemptIsDue} + ORDER BY next_attempt_at, updated_at, collection_id + LIMIT ${limit} + ), leased_candidates AS ( + SELECT * FROM _emdash_media_usage_reconciliations + WHERE state = 'leased' AND ${leaseIsDue} + ORDER BY lease_expires_at, updated_at, collection_id + LIMIT ${limit} + ), candidates AS ( + SELECT * FROM pending_candidates + UNION ALL SELECT * FROM retry_candidates + UNION ALL SELECT * FROM leased_candidates + ) + SELECT * FROM candidates + ORDER BY CASE WHEN state = 'leased' THEN lease_expires_at ELSE next_attempt_at END, + updated_at, + collection_id + LIMIT ${limit} + `.execute(this.db); + return result.rows.map(rowToRecord); + } + + async findFailed(limit: number): Promise { + assertLimit(limit); + const rows = await this.db + .selectFrom("_emdash_media_usage_reconciliations as reconciliation") + .innerJoin("_emdash_media_usage_index_status as status", (join) => + join + .onRef("status.collection_id", "=", "reconciliation.collection_id") + .onRef("status.scope_key", "=", "reconciliation.collection_slug"), + ) + .selectAll("reconciliation") + .where("reconciliation.state", "=", "failed") + .where("status.adapter_id", "=", CONTENT_ADAPTER_ID) + .where("status.scope_type", "=", COLLECTION_SCOPE) + .where((eb) => + eb.or([ + eb("status.reconciliation_required", "=", 0), + eb.and([ + eb("status.status", "=", "running"), + eb("status.cursor", "=", eb.ref("reconciliation.run_token")), + ]), + eb.and([ + eb("status.cursor", "is", null), + eb("status.change_epoch", ">", eb.ref("reconciliation.target_epoch")), + ]), + ]), + ) + .orderBy("reconciliation.updated_at") + .orderBy("reconciliation.collection_id") + .limit(limit) + .execute(); + return rows.map(rowToRecord); + } + + async claim(input: { + collectionId: string; + runToken: string; + leaseDurationSeconds: number; + }): Promise { + assertIdentity(input); + assertDuration(input.leaseDurationSeconds, "lease duration"); + const leaseToken = ulid(); + const row = await this.db + .updateTable("_emdash_media_usage_reconciliations as reconciliation") + .set({ + state: "leased", + lease_token: leaseToken, + lease_expires_at: timestampOffset(this.db, input.leaseDurationSeconds), + updated_at: timestampOffset(this.db, 0), + }) + .where("reconciliation.collection_id", "=", input.collectionId) + .where("reconciliation.run_token", "=", input.runToken) + .where((eb) => + eb.or([ + eb.and([ + eb("reconciliation.state", "in", ["pending", "retry"]), + timestampIsDue(this.db, "reconciliation.next_attempt_at"), + ]), + eb.and([ + eb("reconciliation.state", "=", "leased"), + timestampIsDue(this.db, "reconciliation.lease_expires_at"), + ]), + ]), + ) + .where((eb) => + eb.exists( + eb + .selectFrom("_emdash_media_usage_activation as activation") + .select("activation.task_key") + .where("activation.task_key", "=", ACTIVATION_KEY) + .where("activation.state", "=", "active"), + ), + ) + .where((eb) => + eb.exists( + eb + .selectFrom("_emdash_media_usage_index_status as status") + .innerJoin("_emdash_collections as collection", (join) => + join + .onRef("collection.id", "=", "status.collection_id") + .onRef("collection.slug", "=", "status.scope_key"), + ) + .select("status.collection_id") + .whereRef("status.collection_id", "=", "reconciliation.collection_id") + .whereRef("status.scope_key", "=", "reconciliation.collection_slug") + .where("status.adapter_id", "=", CONTENT_ADAPTER_ID) + .where("status.scope_type", "=", COLLECTION_SCOPE) + .where("status.capture_state", "=", "active") + .where("status.reconciliation_required", "=", 1), + ), + ) + .where((eb) => + eb.not( + eb.exists( + eb + .selectFrom("_emdash_media_usage_collection_deletions as deletion") + .select("deletion.collection_id") + .whereRef("deletion.collection_id", "=", "reconciliation.collection_id"), + ), + ), + ) + .returningAll() + .executeTakeFirst(); + return row + ? ({ ...rowToRecord(row), leaseToken } satisfies MediaUsageReconciliationClaim) + : null; + } + + async release(input: { + collectionId: string; + runToken: string; + leaseToken: string; + delaySeconds: number; + }): Promise { + assertLeaseIdentity(input); + assertDuration(input.delaySeconds, "release delay", true); + const result = await this.db + .updateTable("_emdash_media_usage_reconciliations") + .set({ + state: "pending", + next_attempt_at: timestampOffset(this.db, input.delaySeconds), + lease_token: null, + lease_expires_at: null, + updated_at: timestampOffset(this.db, 0), + }) + .where("collection_id", "=", input.collectionId) + .where("run_token", "=", input.runToken) + .where("state", "=", "leased") + .where("lease_token", "=", input.leaseToken) + .where(liveLease(this.db)) + .executeTakeFirst(); + return Number(result.numUpdatedRows ?? 0) === 1; + } + + async recordFailure(input: { + collectionId: string; + runToken: string; + leaseToken: string; + errorCode: string; + retryDelaySeconds: number; + terminal: boolean; + }): Promise { + assertLeaseIdentity(input); + if (!STABLE_ERROR_CODE_PATTERN.test(input.errorCode)) { + throw new Error("Reconciliation failure requires a stable error code"); + } + assertDuration(input.retryDelaySeconds, "retry delay", true); + const result = await this.db + .updateTable("_emdash_media_usage_reconciliations") + .set({ + state: input.terminal + ? "failed" + : sql`CASE WHEN attempt_count >= 4 THEN 'failed' ELSE 'retry' END`, + ...(input.terminal + ? { + target_epoch: sql`COALESCE( + target_epoch, + (SELECT status.change_epoch + FROM _emdash_media_usage_index_status AS status + WHERE status.adapter_id = ${CONTENT_ADAPTER_ID} + AND status.scope_type = ${COLLECTION_SCOPE} + AND status.collection_id = ${input.collectionId} + AND status.status = 'running' + AND status.cursor = ${input.runToken}) + )`, + } + : {}), + attempt_count: sql`attempt_count + 1`, + next_attempt_at: timestampOffset(this.db, input.retryDelaySeconds), + lease_token: null, + lease_expires_at: null, + last_error_code: input.errorCode, + updated_at: timestampOffset(this.db, 0), + }) + .where("collection_id", "=", input.collectionId) + .where("run_token", "=", input.runToken) + .where("state", "=", "leased") + .where("lease_token", "=", input.leaseToken) + .where(liveLease(this.db)) + .executeTakeFirst(); + return Number(result.numUpdatedRows ?? 0) === 1; + } + + async recordEntryFailure(claim: MediaUsageReconciliationClaim): Promise { + const result = await this.db + .updateTable("_emdash_media_usage_reconciliations as reconciliation") + .set({ + state: "failed", + attempt_count: sql`attempt_count + 1`, + next_attempt_at: timestampOffset(this.db, 0), + lease_token: null, + lease_expires_at: null, + last_error_code: "MEDIA_USAGE_RECONCILIATION_ENTRY_FAILED", + updated_at: timestampOffset(this.db, 0), + }) + .where("reconciliation.collection_id", "=", claim.collectionId) + .where("reconciliation.run_token", "=", claim.runToken) + .where("reconciliation.state", "=", "leased") + .where("reconciliation.lease_token", "=", claim.leaseToken) + .where(liveLease(this.db)) + .where((eb) => + eb.exists( + eb + .selectFrom("_emdash_media_usage_work as work") + .select("work.content_id") + .where("work.collection_id", "=", claim.collectionId) + .where("work.state", "=", "failed"), + ), + ) + .executeTakeFirst(); + return Number(result.numUpdatedRows ?? 0) === 1; + } + + async resetFailedForNewEpoch( + observed: Selectable | MediaUsageReconciliationRecord, + ): Promise { + const collectionId = + "collection_id" in observed ? observed.collection_id : observed.collectionId; + const runToken = "run_token" in observed ? observed.run_token : observed.runToken; + const targetEpoch = "target_epoch" in observed ? observed.target_epoch : observed.targetEpoch; + if (targetEpoch === null) return false; + const now = timestampOffset(this.db, 0); + const result = await this.db + .updateTable("_emdash_media_usage_reconciliations as reconciliation") + .set({ + state: "pending", + phase: "scan", + target_epoch: null, + field_fingerprint: null, + scan_cursor: null, + scan_upper_id: null, + source_cursor: null, + source_upper_key: null, + attempt_count: 0, + next_attempt_at: now, + lease_token: null, + lease_expires_at: null, + last_error_code: null, + updated_at: now, + }) + .where("reconciliation.collection_id", "=", collectionId) + .where("reconciliation.run_token", "=", runToken) + .where("reconciliation.state", "=", "failed") + .where("reconciliation.target_epoch", "=", targetEpoch) + .where((eb) => + eb.exists( + eb + .selectFrom("_emdash_media_usage_index_status as status") + .select("status.collection_id") + .whereRef("status.collection_id", "=", "reconciliation.collection_id") + .whereRef("status.scope_key", "=", "reconciliation.collection_slug") + .where("status.adapter_id", "=", CONTENT_ADAPTER_ID) + .where("status.scope_type", "=", COLLECTION_SCOPE) + .where("status.capture_state", "=", "active") + .where("status.reconciliation_required", "=", 1) + .where("status.cursor", "is", null) + .where("status.change_epoch", ">", targetEpoch), + ), + ) + .executeTakeFirst(); + return Number(result.numUpdatedRows ?? 0) === 1; + } +} + +function rowToRecord( + row: Selectable, +): MediaUsageReconciliationRecord { + if (!isState(row.state) || !isPhase(row.phase) || !Number.isSafeInteger(row.attempt_count)) { + throw new Error("Invalid media usage reconciliation lifecycle"); + } + return { + collectionId: row.collection_id, + collectionSlug: row.collection_slug, + runToken: row.run_token, + targetEpoch: row.target_epoch, + fieldFingerprint: row.field_fingerprint, + state: row.state, + phase: row.phase, + scanCursor: row.scan_cursor, + scanUpperId: row.scan_upper_id, + sourceCursor: row.source_cursor, + sourceUpperKey: row.source_upper_key, + attemptCount: row.attempt_count, + nextAttemptAt: row.next_attempt_at, + leaseToken: row.lease_token, + leaseExpiresAt: row.lease_expires_at, + lastErrorCode: row.last_error_code, + createdAt: row.created_at, + updatedAt: row.updated_at, + }; +} + +function isState(state: string): state is MediaUsageReconciliationState { + return state === "pending" || state === "retry" || state === "leased" || state === "failed"; +} + +function isPhase(phase: string): phase is MediaUsageReconciliationPhase { + return phase === "scan" || phase === "sources"; +} + +function assertLimit(limit: number): void { + if (!Number.isSafeInteger(limit) || limit < 1 || limit > MAX_CANDIDATES) { + throw new Error("Reconciliation candidate limit must be from 1 to 100"); + } +} + +function assertIdentity(input: { collectionId: string; runToken: string }): void { + if (!input.collectionId || !input.runToken) { + throw new Error("Reconciliation requires an exact collection and run token"); + } +} + +function assertLeaseIdentity(input: { + collectionId: string; + runToken: string; + leaseToken: string; +}): void { + assertIdentity(input); + if (!input.leaseToken) throw new Error("Reconciliation requires a lease token"); +} + +function assertDuration(value: number, label: string, allowZero = false): void { + if ( + !Number.isSafeInteger(value) || + value < (allowZero ? 0 : 1) || + value > MAX_PORTABLE_DURATION_SECONDS + ) { + throw new Error(`Reconciliation ${label} is outside the portable range`); + } +} + +function liveLease(db: Kysely, column = "lease_expires_at"): RawBuilder { + const expiry = sql.ref(column); + return isPostgres(db) + ? sql`${expiry} > to_char(statement_timestamp() AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.MS"Z"')` + : sql`${expiry} > strftime('%Y-%m-%dT%H:%M:%fZ', 'now')`; +} + +function timestampIsDue(db: Kysely, column: string): RawBuilder { + const value = sql.ref(column); + return isPostgres(db) + ? sql`${value} <= to_char(statement_timestamp() AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.MS"Z"')` + : sql`${value} <= strftime('%Y-%m-%dT%H:%M:%fZ', 'now')`; +} + +function timestampOffset(db: Kysely, offsetSeconds: number): RawBuilder { + if (isPostgres(db)) { + return sql`to_char( + (clock_timestamp() AT TIME ZONE 'UTC') + (${offsetSeconds} * INTERVAL '1 second'), + 'YYYY-MM-DD"T"HH24:MI:SS.MS"Z"' + )`; + } + return sql`strftime( + '%Y-%m-%dT%H:%M:%fZ', + 'now', + ${`${offsetSeconds >= 0 ? "+" : ""}${offsetSeconds} seconds`} + )`; +} + +function contentTableName(collectionSlug: string): string { + validateIdentifier(collectionSlug, "collection slug"); + const tableName = `ec_${collectionSlug}`; + validateIdentifier(tableName, "content table"); + return tableName; +} diff --git a/packages/core/src/media/usage/types.ts b/packages/core/src/media/usage/types.ts index a76281c7db..d4fc944058 100644 --- a/packages/core/src/media/usage/types.ts +++ b/packages/core/src/media/usage/types.ts @@ -1,5 +1,7 @@ import type { FieldType } from "../../schema/types.js"; +export const CONTENT_SOURCE_SCHEMA_VERSION = 1; + export type MediaKind = | "image" | "video" diff --git a/packages/core/src/plugins/scheduler/node.ts b/packages/core/src/plugins/scheduler/node.ts index c4486c4b3d..32e63d9ded 100644 --- a/packages/core/src/plugins/scheduler/node.ts +++ b/packages/core/src/plugins/scheduler/node.ts @@ -29,6 +29,7 @@ export class NodeCronScheduler implements CronScheduler { private timer: ReturnType | null = null; private running = false; private systemCleanup: SystemCleanupFn | null = null; + private mediaUsageMaintenance: SystemCleanupFn | null = null; constructor(private executor: CronExecutor) {} @@ -36,6 +37,10 @@ export class NodeCronScheduler implements CronScheduler { this.systemCleanup = fn; } + setMediaUsageMaintenance(fn: SystemCleanupFn): void { + this.mediaUsageMaintenance = fn; + } + start(): void { this.running = true; this.arm(); @@ -112,12 +117,19 @@ export class NodeCronScheduler implements CronScheduler { } void Promise.allSettled(tasks) - .then((results) => { + .then(async (results) => { for (const r of results) { if (r.status === "rejected") { console.error("[cron:node] Tick task failed:", r.reason); } } + if (this.mediaUsageMaintenance) { + try { + await this.mediaUsageMaintenance(); + } catch (error) { + console.error("[cron:node] Media Usage maintenance failed:", error); + } + } return undefined; }) .finally(() => { diff --git a/packages/core/src/plugins/scheduler/types.ts b/packages/core/src/plugins/scheduler/types.ts index 511f9b8fcc..f4ee58afbb 100644 --- a/packages/core/src/plugins/scheduler/types.ts +++ b/packages/core/src/plugins/scheduler/types.ts @@ -18,6 +18,8 @@ export interface CronScheduler { reschedule(): void; /** Register a system cleanup function to run alongside each tick. */ setSystemCleanup(fn: SystemCleanupFn): void; + /** Register bounded Media Usage maintenance to run after the general tick settles. */ + setMediaUsageMaintenance?(fn: SystemCleanupFn): void; } /** diff --git a/packages/core/tests/integration/database/media-usage-collection-deletion-lifecycle.test.ts b/packages/core/tests/integration/database/media-usage-collection-deletion-lifecycle.test.ts index 5e032093fa..285ad93c6c 100644 --- a/packages/core/tests/integration/database/media-usage-collection-deletion-lifecycle.test.ts +++ b/packages/core/tests/integration/database/media-usage-collection-deletion-lifecycle.test.ts @@ -339,6 +339,11 @@ describeEachDialect("media usage activated collection deletion", (dialect) => { slug: "projecting", label: "Projecting", }); + const sourceUpdatedAt = "2026-08-12T10:00:00.000Z"; + await sql` + INSERT INTO ${sql.ref("ec_projecting")} (id, slug, version, updated_at) + VALUES ('entry-1', 'entry-1', 1, ${sourceUpdatedAt}) + `.execute(ctx.db); const advisoryKey = 8642031; await sql .raw(` @@ -386,6 +391,9 @@ describeEachDialect("media usage activated collection deletion", (dialect) => { collectionSlug: collection.slug, contentId: "entry-1", sourceVariant: "columns", + revisionId: null, + sourceVersion: 1, + sourceUpdatedAt, identityVersion: 1, }, [ diff --git a/packages/core/tests/integration/database/media-usage-content-fields.test.ts b/packages/core/tests/integration/database/media-usage-content-fields.test.ts index eee9d89044..50c3d4b2bf 100644 --- a/packages/core/tests/integration/database/media-usage-content-fields.test.ts +++ b/packages/core/tests/integration/database/media-usage-content-fields.test.ts @@ -2,6 +2,7 @@ import { afterEach, beforeEach, expect, it } from "vitest"; import { IdentifierError } from "../../../src/database/validate.js"; import { + buildContentMediaUsageFieldFingerprint, loadContentMediaUsageFields, MediaUsageFieldDiscoveryError, } from "../../../src/media/usage/content-fields.js"; @@ -152,4 +153,46 @@ describeEachDialect("content media usage field discovery", (dialect) => { await expect(loadContentMediaUsageFields(ctx.db, "posts")).rejects.toThrow(IdentifierError); }); + + it("fingerprints exact extraction and display-field definitions independent of row order", async () => { + const first = await buildContentMediaUsageFieldFingerprint({ + extractionFields: [ + { + slug: "sections", + type: "repeater", + validation: { + subFields: [ + { slug: "secondary", type: "image" }, + { slug: "primary", type: "image" }, + ], + }, + }, + { slug: "hero", type: "image" }, + ], + displayFieldSlugs: ["title", "name"], + }); + const reordered = await buildContentMediaUsageFieldFingerprint({ + extractionFields: [ + { slug: "hero", type: "image" }, + { + slug: "sections", + type: "repeater", + validation: { + subFields: [ + { slug: "primary", type: "image" }, + { slug: "secondary", type: "image" }, + ], + }, + }, + ], + displayFieldSlugs: ["name", "title"], + }); + const changed = await buildContentMediaUsageFieldFingerprint({ + extractionFields: [{ slug: "hero", type: "file" }], + displayFieldSlugs: ["name", "title"], + }); + + expect(first).toBe(reordered); + expect(changed).not.toBe(first); + }); }); diff --git a/packages/core/tests/integration/database/media-usage-incremental-work-migration.test.ts b/packages/core/tests/integration/database/media-usage-incremental-work-migration.test.ts index d27ec993e5..0ba28c2207 100644 --- a/packages/core/tests/integration/database/media-usage-incremental-work-migration.test.ts +++ b/packages/core/tests/integration/database/media-usage-incremental-work-migration.test.ts @@ -49,6 +49,9 @@ describeEachDialect("media usage incremental work migration", (dialect) => { }); it("keeps V1 collection deletion available after rolling back incremental capture", async () => { + const reconciliationMigration = + await import("../../../src/database/migrations/066_media_usage_reconciliation.js"); + await reconciliationMigration.down(ctx.db); const collectionDeletionMigration = await import("../../../src/database/migrations/065_media_usage_collection_deletion.js"); await collectionDeletionMigration.down(ctx.db); @@ -64,6 +67,9 @@ describeEachDialect("media usage incremental work migration", (dialect) => { }); it("upgrades and reruns without rewriting legacy evidence or inventing work", async () => { + const reconciliationMigration = + await import("../../../src/database/migrations/066_media_usage_reconciliation.js"); + await reconciliationMigration.down(ctx.db); const collectionDeletionMigration = await import("../../../src/database/migrations/065_media_usage_collection_deletion.js"); await collectionDeletionMigration.down(ctx.db); @@ -209,6 +215,9 @@ describeEachDialect("media usage incremental work migration", (dialect) => { }); it("purges a partially bound status if its collection is deleted or recreated before retry", async () => { + const reconciliationMigration = + await import("../../../src/database/migrations/066_media_usage_reconciliation.js"); + await reconciliationMigration.down(ctx.db); const collectionDeletionMigration = await import("../../../src/database/migrations/065_media_usage_collection_deletion.js"); await collectionDeletionMigration.down(ctx.db); diff --git a/packages/core/tests/integration/database/media-usage-projection-admission.test.ts b/packages/core/tests/integration/database/media-usage-projection-admission.test.ts index c5fc2f5e38..e98b744ff0 100644 --- a/packages/core/tests/integration/database/media-usage-projection-admission.test.ts +++ b/packages/core/tests/integration/database/media-usage-projection-admission.test.ts @@ -1,3 +1,4 @@ +import { sql } from "kysely"; import { afterEach, beforeEach, expect, it } from "vitest"; import { @@ -38,6 +39,15 @@ describeEachDialect("content media usage projection admission", (dialect) => { .insertInto("_emdash_collections") .values({ id: COLLECTION_ID, slug: COLLECTION_SLUG, label: "Admission" }) .execute(); + await sql` + CREATE TABLE ${sql.ref("ec_admission")} ( + id TEXT PRIMARY KEY, + version INTEGER NOT NULL, + updated_at TEXT NOT NULL, + live_revision_id TEXT, + draft_revision_id TEXT + ) + `.execute(ctx.db); }); afterEach(async () => { @@ -60,6 +70,7 @@ describeEachDialect("content media usage projection admission", (dialect) => { it("recomputes an oversized mixed plan after proving its replacement is unchanged", async () => { const contentId = "mixed-no-op-delete"; + await insertContentIdentity(ctx, contentId, true); const unchangedColumns = await snapshot(contentId, "columns", 0, "é".repeat(270_000)); const absentDraft = await snapshot(contentId, "draft_overlay", 1, "Small draft"); await repo.replaceSource(unchangedColumns.source, unchangedColumns.occurrences); @@ -89,6 +100,7 @@ describeEachDialect("content media usage projection admission", (dialect) => { it("rejects deletion when the stored source row alone exceeds the byte limit", async () => { const contentId = "oversized-source-delete"; + await insertContentIdentity(ctx, contentId, true); const absentDraft = await snapshot(contentId, "draft_overlay", 0, "é".repeat(300_000)); await repo.replaceSource(absentDraft.source, []); const observedSources = await repo.findSources(canonicalSourceKeys(contentId)); @@ -151,6 +163,24 @@ describeEachDialect("content media usage projection admission", (dialect) => { } }); +async function insertContentIdentity( + ctx: DialectTestContext, + contentId: string, + withDraft: boolean, +): Promise { + await sql` + INSERT INTO ${sql.ref("ec_admission")} ( + id, version, updated_at, live_revision_id, draft_revision_id + ) VALUES ( + ${contentId}, + 1, + '2026-08-11T00:00:00.000Z', + NULL, + ${withDraft ? `revision-${contentId}` : null} + ) + `.execute(ctx.db); +} + async function snapshot( contentId: string, sourceVariant: MediaUsageContentSourceVariant, diff --git a/packages/core/tests/integration/database/media-usage-read-repository.test.ts b/packages/core/tests/integration/database/media-usage-read-repository.test.ts index 7112a1834f..51503d836b 100644 --- a/packages/core/tests/integration/database/media-usage-read-repository.test.ts +++ b/packages/core/tests/integration/database/media-usage-read-repository.test.ts @@ -1,3 +1,4 @@ +import { sql } from "kysely"; import { afterEach, beforeEach, expect, it } from "vitest"; import { MediaUsageRepository } from "../../../src/database/repositories/media-usage.js"; @@ -190,6 +191,7 @@ describeEachDialect("MediaUsageRepository reads", (dialect) => { .insertInto("_emdash_collections") .values({ id: "collection-posts-old", slug: "posts", label: "Old posts" }) .execute(); + await installCanonicalContentFixture(ctx, "posts", "old-only", "rev-old-only-columns"); await repo.replaceSource( contentSource("old-only", "columns", { sourceKey: buildContentMediaUsageSourceKey({ @@ -200,6 +202,8 @@ describeEachDialect("MediaUsageRepository reads", (dialect) => { }), collectionId: "collection-posts-old", identityVersion: 1, + sourceVersion: 1, + sourceUpdatedAt: "2026-08-12T00:00:00.000Z", }), [occurrence("hero", "media-shared")], ); @@ -239,6 +243,7 @@ describeEachDialect("MediaUsageRepository reads", (dialect) => { }), [occurrence("unversioned", "media-shared")], ); + await installCanonicalContentFixture(ctx, "posts", "current", "rev-current-columns"); await repo.replaceSource( contentSource("current", "columns", { sourceKey: buildContentMediaUsageSourceKey({ @@ -250,6 +255,8 @@ describeEachDialect("MediaUsageRepository reads", (dialect) => { collectionId: "collection-posts", contentStatus: "draft", identityVersion: 1, + sourceVersion: 1, + sourceUpdatedAt: "2026-08-12T00:00:00.000Z", }), [occurrence("canonical", "media-shared")], ); @@ -492,6 +499,31 @@ async function registerCollection(ctx: DialectTestContext, slug: string): Promis .execute(); } +async function installCanonicalContentFixture( + ctx: DialectTestContext, + collectionSlug: string, + contentId: string, + liveRevisionId: string, +): Promise { + const tableName = `ec_${collectionSlug}`; + await sql` + CREATE TABLE IF NOT EXISTS ${sql.ref(tableName)} ( + id TEXT PRIMARY KEY, + version INTEGER NOT NULL, + updated_at TEXT NOT NULL, + live_revision_id TEXT, + draft_revision_id TEXT + ) + `.execute(ctx.db); + await sql` + INSERT INTO ${sql.ref(tableName)} ( + id, version, updated_at, live_revision_id, draft_revision_id + ) VALUES ( + ${contentId}, 1, '2026-08-12T00:00:00.000Z', ${liveRevisionId}, NULL + ) + `.execute(ctx.db); +} + function entryIdentity(entry: { collectionSlug: string; contentId: string }): [string, string] { return [entry.collectionSlug, entry.contentId]; } 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 new file mode 100644 index 0000000000..c3ae649d77 --- /dev/null +++ b/packages/core/tests/integration/database/media-usage-reconciliation-finalization.test.ts @@ -0,0 +1,369 @@ +import { sql } from "kysely"; +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 { 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"; +import { SchemaRegistry } from "../../../src/schema/registry.js"; +import { + describeEachDialect, + setupForDialect, + teardownForDialect, + type DialectTestContext, +} from "../../utils/test-db.js"; + +describeEachDialect("media usage reconciliation finalization", (dialect) => { + let ctx: DialectTestContext; + + beforeEach(async () => { + ctx = await setupForDialect(dialect); + }); + + afterEach(async () => { + await teardownForDialect(ctx); + }); + + it("reconciles historical missing content through existing entry work", async () => { + const collection = await createCollection(ctx, "articles", true); + await sql` + INSERT INTO ${sql.ref("ec_articles")} (id, slug) + VALUES ('historical', 'historical') + `.execute(ctx.db); + const snapshots = await loadContentMediaUsageSnapshots( + ctx.db, + collection.slug, + "historical", + undefined, + { collectionId: collection.id, identityVersion: 1 }, + ); + if (!snapshots.success) throw new Error(snapshots.error); + const usage = new MediaUsageRepository(ctx.db); + for (const snapshot of snapshots.snapshots) { + await usage.replaceSourceIfMatching(snapshot.source, snapshot.occurrences, null); + } + const quarantinedSourceKey = buildContentMediaUsageSourceKey({ + collectionId: collection.id, + collectionSlug: collection.slug, + contentId: "quarantined", + sourceVariant: "columns", + }); + await ctx.db + .insertInto("_emdash_media_usage_sources") + .values({ + source_key: quarantinedSourceKey, + source_type: "content", + collection_id: collection.id, + collection_slug: collection.slug, + content_id: "quarantined", + source_variant: "columns", + current_generation: "legacy-generation", + identity_version: null, + }) + .execute(); + await sql`DELETE FROM ${sql.ref("ec_articles")} WHERE id = 'historical'`.execute(ctx.db); + await activateCollection(ctx, collection); + + await expect(processDueMediaUsageReconciliation(ctx.db)).resolves.toBe("advanced"); + await expect(processDueMediaUsageReconciliation(ctx.db)).resolves.toBe("advanced"); + expect( + await ctx.db + .selectFrom("_emdash_media_usage_work") + .select(["content_id", "state"]) + .where("collection_id", "=", collection.id) + .execute(), + ).toEqual([{ content_id: "historical", state: "pending" }]); + + await expect(processDueMediaUsageWork(ctx.db)).resolves.toMatchObject({ completedCount: 1 }); + expect(await usage.findSource(snapshots.snapshots[0]!.source.sourceKey)).toBeNull(); + await expect(processDueMediaUsageReconciliation(ctx.db)).resolves.toBe("completed"); + + expect( + await ctx.db + .selectFrom("_emdash_media_usage_index_status") + .select(["status", "reconciliation_required", "cursor", "last_error_code"]) + .where("collection_id", "=", collection.id) + .executeTakeFirstOrThrow(), + ).toEqual({ + status: "complete", + reconciliation_required: 0, + cursor: null, + last_error_code: null, + }); + expect( + await ctx.db + .selectFrom("_emdash_media_usage_reconciliations") + .select("collection_id") + .where("collection_id", "=", collection.id) + .executeTakeFirst(), + ).toBeUndefined(); + expect(await usage.findSource(quarantinedSourceKey)).not.toBeNull(); + }); + + it("preserves automatic ownership until failed work reaches terminal coverage", async () => { + const collection = await createCollection(ctx, "articles", true); + await sql` + INSERT INTO ${sql.ref("ec_articles")} (id, slug) + VALUES ('entry-1', 'entry-1') + `.execute(ctx.db); + await activateCollection(ctx, collection); + await expect(processDueMediaUsageReconciliation(ctx.db)).resolves.toBe("advanced"); + const coordinator = await ctx.db + .selectFrom("_emdash_media_usage_reconciliations") + .select(["run_token", "target_epoch"]) + .where("collection_id", "=", collection.id) + .executeTakeFirstOrThrow(); + const work = await ctx.db + .updateTable("_emdash_media_usage_work") + .set({ state: "failed", last_error_code: "MEDIA_USAGE_RESOURCE_LIMIT" }) + .where("collection_id", "=", collection.id) + .returning(["content_id", "work_version"]) + .executeTakeFirstOrThrow(); + await new MediaUsageRepository(ctx.db).recordIncrementalFailure({ + collectionId: collection.id, + collectionSlug: collection.slug, + contentId: work.content_id, + workVersion: work.work_version, + errorCode: "MEDIA_USAGE_RESOURCE_LIMIT", + }); + + expect( + await ctx.db + .selectFrom("_emdash_media_usage_index_status") + .select(["status", "cursor", "last_error_code"]) + .where("collection_id", "=", collection.id) + .executeTakeFirstOrThrow(), + ).toEqual({ + status: "running", + cursor: coordinator.run_token, + last_error_code: "MEDIA_USAGE_RESOURCE_LIMIT", + }); + await expect(processDueMediaUsageReconciliation(ctx.db)).resolves.toBe("failed"); + + expect( + await ctx.db + .selectFrom("_emdash_media_usage_reconciliations") + .select(["state", "last_error_code", "target_epoch"]) + .where("collection_id", "=", collection.id) + .executeTakeFirstOrThrow(), + ).toEqual({ + state: "failed", + last_error_code: "MEDIA_USAGE_RECONCILIATION_ENTRY_FAILED", + target_epoch: coordinator.target_epoch, + }); + expect( + await ctx.db + .selectFrom("_emdash_media_usage_index_status") + .select(["status", "reconciliation_required", "cursor", "last_error_code"]) + .where("collection_id", "=", collection.id) + .executeTakeFirstOrThrow(), + ).toEqual({ + status: "failed", + reconciliation_required: 1, + cursor: null, + last_error_code: "MEDIA_USAGE_RESOURCE_LIMIT", + }); + }); + + it("collection deletion removes the exact reconciliation row before status", async () => { + const collection = await createCollection(ctx, "articles", false); + await activateCollection(ctx, collection); + await processDueMediaUsageReconciliation(ctx.db); + await ctx.db + .insertInto("_emdash_media_usage_collection_deletions") + .values({ + collection_id: collection.id, + collection_slug: collection.slug, + force_delete: 1, + state: "pending", + phase: "status", + next_attempt_at: "2000-01-01T00:00:00.000Z", + }) + .execute(); + + await expect(processDueMediaUsageCollectionDeletions(ctx.db)).resolves.toMatchObject({ + outcome: "progress", + }); + expect( + await ctx.db + .selectFrom("_emdash_media_usage_reconciliations") + .select("collection_id") + .where("collection_id", "=", collection.id) + .executeTakeFirst(), + ).toBeUndefined(); + expect( + await ctx.db + .selectFrom("_emdash_media_usage_index_status") + .select("collection_id") + .where("collection_id", "=", collection.id) + .executeTakeFirst(), + ).toBeUndefined(); + }); + + it("does not persist terminal entry failure after work has been reopened", async () => { + const collection = await createCollection(ctx, "articles", true); + await sql` + INSERT INTO ${sql.ref("ec_articles")} (id, slug) + VALUES ('entry-1', 'entry-1') + `.execute(ctx.db); + await activateCollection(ctx, collection); + await processDueMediaUsageReconciliation(ctx.db); + const repository = new MediaUsageReconciliationRepository(ctx.db); + const [candidate] = await repository.findDue(4); + if (!candidate) throw new Error("Expected reconciliation work"); + const claim = await repository.claim({ + collectionId: candidate.collectionId, + runToken: candidate.runToken, + leaseDurationSeconds: 60, + }); + if (!claim) throw new Error("Expected reconciliation claim"); + + await expect(repository.recordEntryFailure(claim)).resolves.toBe(false); + expect( + await ctx.db + .selectFrom("_emdash_media_usage_reconciliations") + .select("state") + .where("collection_id", "=", collection.id) + .executeTakeFirstOrThrow(), + ).toEqual({ state: "leased" }); + }); + + it("restarts from a new epoch after a schema fingerprint changes", async () => { + const collection = await createCollection(ctx, "articles", true); + await sql` + INSERT INTO ${sql.ref("ec_articles")} (id, slug) + VALUES ('entry-1', 'entry-1') + `.execute(ctx.db); + await activateCollection(ctx, collection); + await processDueMediaUsageReconciliation(ctx.db); + await processDueMediaUsageWork(ctx.db); + await processDueMediaUsageReconciliation(ctx.db); + const before = await ctx.db + .selectFrom("_emdash_media_usage_reconciliations") + .select(["target_epoch", "field_fingerprint", "phase"]) + .where("collection_id", "=", collection.id) + .executeTakeFirstOrThrow(); + + await new SchemaRegistry(ctx.db).createField("articles", { + slug: "attachment", + label: "Attachment", + type: "file", + }); + await expect(processDueMediaUsageReconciliation(ctx.db)).resolves.toBe("advanced"); + const after = await ctx.db + .selectFrom("_emdash_media_usage_reconciliations") + .select(["target_epoch", "field_fingerprint", "phase", "scan_cursor", "source_cursor"]) + .where("collection_id", "=", collection.id) + .executeTakeFirstOrThrow(); + + expect(Number(after.target_epoch)).toBeGreaterThan(Number(before.target_epoch)); + expect(after.field_fingerprint).not.toBe(before.field_fingerprint); + expect(after).toMatchObject({ phase: "scan", scan_cursor: null, source_cursor: null }); + }); + + it("defers without taking coverage from a manual repair owner", async () => { + const collection = await createCollection(ctx, "articles", true); + await sql` + INSERT INTO ${sql.ref("ec_articles")} (id, slug) + VALUES ('entry-1', 'entry-1') + `.execute(ctx.db); + await activateCollection(ctx, collection); + await processDueMediaUsageReconciliation(ctx.db); + await ctx.db + .updateTable("_emdash_media_usage_index_status") + .set({ status: "running", cursor: "manual-repair" }) + .where("collection_id", "=", collection.id) + .execute(); + + await expect(processDueMediaUsageReconciliation(ctx.db)).resolves.toBe("deferred"); + expect( + await ctx.db + .selectFrom("_emdash_media_usage_index_status") + .select(["status", "cursor"]) + .where("collection_id", "=", collection.id) + .executeTakeFirstOrThrow(), + ).toEqual({ status: "running", cursor: "manual-repair" }); + }); + + it("retains the claimed epoch when initialization fails terminally", async () => { + const collection = await createCollection(ctx, "articles", true); + await activateCollection(ctx, collection); + const repository = new MediaUsageReconciliationRepository(ctx.db); + await repository.seedNextCandidate(); + const [candidate] = await repository.findDue(4); + if (!candidate) throw new Error("Expected reconciliation candidate"); + const claim = await repository.claim({ + collectionId: candidate.collectionId, + runToken: candidate.runToken, + leaseDurationSeconds: 60, + }); + if (!claim) throw new Error("Expected reconciliation claim"); + const epoch = await repository.beginRun(claim); + if (epoch === null) throw new Error("Expected a claimed coverage epoch"); + + await repository.recordFailure({ + collectionId: claim.collectionId, + runToken: claim.runToken, + leaseToken: claim.leaseToken, + errorCode: "MEDIA_USAGE_RECONCILIATION_FAILED", + retryDelaySeconds: 0, + terminal: true, + }); + expect( + await ctx.db + .selectFrom("_emdash_media_usage_reconciliations") + .select(["state", "target_epoch"]) + .where("collection_id", "=", collection.id) + .executeTakeFirstOrThrow(), + ).toEqual({ state: "failed", target_epoch: epoch }); + }); +}); + +async function createCollection( + ctx: DialectTestContext, + slug: string, + withMediaField: boolean, +): Promise<{ id: string; slug: string }> { + const registry = new SchemaRegistry(ctx.db); + await registry.createCollection({ slug, label: slug }); + if (withMediaField) { + await registry.createField(slug, { slug: "hero", label: "Hero", type: "image" }); + } + const collection = await registry.getCollection(slug); + if (!collection) throw new Error(`Expected ${slug} collection`); + return { id: collection.id, slug: collection.slug }; +} + +async function activateCollection( + ctx: DialectTestContext, + collection: { id: string; slug: string }, +): Promise { + await ctx.db + .insertInto("_emdash_media_usage_index_status") + .values({ + adapter_id: "content-media", + scope_type: "collection", + scope_key: collection.slug, + collection_id: collection.id, + status: "stale", + capture_state: "active", + reconciliation_required: 1, + }) + .onConflict((conflict) => + conflict.columns(["adapter_id", "scope_type", "scope_key"]).doUpdateSet({ + collection_id: collection.id, + status: "stale", + capture_state: "active", + reconciliation_required: 1, + }), + ) + .execute(); + await ctx.db + .updateTable("_emdash_media_usage_activation") + .set({ state: "active" }) + .where("task_key", "=", "incremental_capture") + .execute(); +} diff --git a/packages/core/tests/integration/database/media-usage-reconciliation-foundation.test.ts b/packages/core/tests/integration/database/media-usage-reconciliation-foundation.test.ts new file mode 100644 index 0000000000..8366209922 --- /dev/null +++ b/packages/core/tests/integration/database/media-usage-reconciliation-foundation.test.ts @@ -0,0 +1,296 @@ +import { afterEach, beforeEach, expect, it } from "vitest"; + +import { indexExists, tableExists } from "../../../src/database/dialect-helpers.js"; +import { MediaUsageReconciliationRepository } from "../../../src/media/usage/reconciliation.js"; +import { SchemaRegistry } from "../../../src/schema/registry.js"; +import { + describeEachDialect, + setupForDialect, + teardownForDialect, + type DialectTestContext, +} from "../../utils/test-db.js"; + +describeEachDialect("media usage reconciliation foundation", (dialect) => { + let ctx: DialectTestContext; + + beforeEach(async () => { + ctx = await setupForDialect(dialect); + }); + + afterEach(async () => { + await teardownForDialect(ctx); + }); + + it("installs the durable coordinator schema with dormant scheduling", async () => { + expect(await tableExists(ctx.db, "_emdash_media_usage_reconciliations")).toBe(true); + expect(await indexExists(ctx.db, "idx__emdash_media_usage_reconciliations_due")).toBe(true); + expect(await indexExists(ctx.db, "idx__emdash_media_usage_reconciliations_lease")).toBe(true); + expect(await indexExists(ctx.db, "idx__emdash_media_usage_reconciliations_failed")).toBe(true); + expect(await indexExists(ctx.db, "idx__emdash_media_usage_status_reconciliation")).toBe(true); + + const activation = await ctx.db + .selectFrom("_emdash_media_usage_activation") + .select(["state", "media_usage_maintenance_turn"]) + .where("task_key", "=", "incremental_capture") + .executeTakeFirstOrThrow(); + expect(activation).toEqual({ state: "expanded", media_usage_maintenance_turn: 2 }); + }); + + it("seeds at most one exact active collection per invocation", async () => { + const first = await createActiveCollection(ctx, "articles"); + const second = await createActiveCollection(ctx, "pages"); + const repository = new MediaUsageReconciliationRepository(ctx.db); + + await expect(repository.seedNextCandidate()).resolves.toBe(false); + await ctx.db + .updateTable("_emdash_media_usage_activation") + .set({ state: "active" }) + .where("task_key", "=", "incremental_capture") + .execute(); + + await expect(repository.seedNextCandidate()).resolves.toBe(true); + expect(await reconciliationIdentities(ctx)).toHaveLength(1); + await expect(repository.seedNextCandidate()).resolves.toBe(true); + expect(await reconciliationIdentities(ctx)).toEqual( + [first, second].toSorted((left, right) => left.id.localeCompare(right.id)), + ); + await expect(repository.seedNextCandidate()).resolves.toBe(false); + }); + + it("excludes inactive, complete, and deleting collections from discovery", async () => { + await createActiveCollection(ctx, "inactive", { captureState: "installing" }); + await createActiveCollection(ctx, "complete", { reconciliationRequired: 0 }); + await createActiveCollection(ctx, "deleting", { captureState: "deleting" }); + await ctx.db + .updateTable("_emdash_media_usage_activation") + .set({ state: "active" }) + .where("task_key", "=", "incremental_capture") + .execute(); + + const repository = new MediaUsageReconciliationRepository(ctx.db); + await expect(repository.seedNextCandidate()).resolves.toBe(false); + expect(await reconciliationIdentities(ctx)).toEqual([]); + }); + + it("claims due work once and fences stale lease owners", async () => { + const collection = await createActiveCollection(ctx, "articles"); + await activateAndSeed(ctx); + const repository = new MediaUsageReconciliationRepository(ctx.db); + const [candidate] = await repository.findDue(4); + if (!candidate) throw new Error("Expected a due reconciliation candidate"); + + const claim = await repository.claim({ + collectionId: collection.id, + runToken: candidate.runToken, + leaseDurationSeconds: 60, + }); + expect(claim).toMatchObject({ state: "leased", collectionId: collection.id }); + await expect( + repository.claim({ + collectionId: collection.id, + runToken: candidate.runToken, + leaseDurationSeconds: 60, + }), + ).resolves.toBeNull(); + + await expect( + repository.release({ + collectionId: collection.id, + runToken: candidate.runToken, + leaseToken: "stale-owner", + delaySeconds: 30, + }), + ).resolves.toBe(false); + await expect( + repository.release({ + collectionId: collection.id, + runToken: candidate.runToken, + leaseToken: claim!.leaseToken, + delaySeconds: 30, + }), + ).resolves.toBe(true); + }); + + it("does not claim a coordinator after coverage becomes complete", async () => { + const collection = await createActiveCollection(ctx, "articles"); + await activateAndSeed(ctx); + const repository = new MediaUsageReconciliationRepository(ctx.db); + const [candidate] = await repository.findDue(4); + if (!candidate) throw new Error("Expected a due reconciliation candidate"); + await ctx.db + .updateTable("_emdash_media_usage_index_status") + .set({ status: "complete", reconciliation_required: 0 }) + .where("collection_id", "=", collection.id) + .execute(); + + await expect( + repository.claim({ + collectionId: collection.id, + runToken: candidate.runToken, + leaseDurationSeconds: 60, + }), + ).resolves.toBeNull(); + }); + + it("persists bounded retry and terminal failure without hot-looping", async () => { + await createActiveCollection(ctx, "articles"); + await activateAndSeed(ctx); + const repository = new MediaUsageReconciliationRepository(ctx.db); + const [candidate] = await repository.findDue(4); + if (!candidate) throw new Error("Expected a due reconciliation candidate"); + const firstClaim = await repository.claim({ + collectionId: candidate.collectionId, + runToken: candidate.runToken, + leaseDurationSeconds: 60, + }); + if (!firstClaim) throw new Error("Expected the first reconciliation claim"); + + await expect( + repository.recordFailure({ + collectionId: candidate.collectionId, + runToken: candidate.runToken, + leaseToken: firstClaim.leaseToken, + errorCode: "MEDIA_USAGE_RECONCILIATION_FAILED", + retryDelaySeconds: 30, + terminal: false, + }), + ).resolves.toBe(true); + expect(await repository.findDue(4)).toEqual([]); + + await ctx.db + .updateTable("_emdash_media_usage_reconciliations") + .set({ next_attempt_at: "2000-01-01T00:00:00.000Z" }) + .where("collection_id", "=", candidate.collectionId) + .execute(); + const retryClaim = await repository.claim({ + collectionId: candidate.collectionId, + runToken: candidate.runToken, + leaseDurationSeconds: 60, + }); + if (!retryClaim) throw new Error("Expected the retry reconciliation claim"); + await expect( + repository.recordFailure({ + collectionId: candidate.collectionId, + runToken: candidate.runToken, + leaseToken: retryClaim.leaseToken, + errorCode: "MEDIA_USAGE_RECONCILIATION_INVALID_SOURCE", + retryDelaySeconds: 0, + terminal: true, + }), + ).resolves.toBe(true); + + expect(await repository.findDue(4)).toEqual([]); + expect(await repository.findFailed(4)).toEqual([]); + expect( + await ctx.db + .selectFrom("_emdash_media_usage_reconciliations") + .select(["collection_id", "state", "attempt_count", "last_error_code"]) + .where("collection_id", "=", candidate.collectionId) + .executeTakeFirstOrThrow(), + ).toEqual({ + collection_id: candidate.collectionId, + state: "failed", + attempt_count: 2, + last_error_code: "MEDIA_USAGE_RECONCILIATION_INVALID_SOURCE", + }); + }); + + it("reopens failed work only after a newer coverage epoch", async () => { + const collection = await createActiveCollection(ctx, "articles"); + await activateAndSeed(ctx); + const repository = new MediaUsageReconciliationRepository(ctx.db); + const row = await ctx.db + .updateTable("_emdash_media_usage_reconciliations") + .set({ + state: "failed", + target_epoch: 4, + attempt_count: 5, + last_error_code: "MEDIA_USAGE_RECONCILIATION_FAILED", + }) + .where("collection_id", "=", collection.id) + .returningAll() + .executeTakeFirstOrThrow(); + + await ctx.db + .updateTable("_emdash_media_usage_index_status") + .set({ change_epoch: 4, cursor: null }) + .where("collection_id", "=", collection.id) + .execute(); + await expect(repository.resetFailedForNewEpoch(row)).resolves.toBe(false); + + await ctx.db + .updateTable("_emdash_media_usage_index_status") + .set({ change_epoch: 5 }) + .where("collection_id", "=", collection.id) + .execute(); + await expect(repository.resetFailedForNewEpoch(row)).resolves.toBe(true); + expect( + await ctx.db + .selectFrom("_emdash_media_usage_reconciliations") + .select(["state", "target_epoch", "attempt_count", "last_error_code"]) + .where("collection_id", "=", collection.id) + .executeTakeFirstOrThrow(), + ).toEqual({ state: "pending", target_epoch: null, attempt_count: 0, last_error_code: null }); + }); + + it("refuses rollback while reconciliation evidence exists", async () => { + const migration = + await import("../../../src/database/migrations/066_media_usage_reconciliation.js"); + await ctx.db + .insertInto("_emdash_media_usage_reconciliations") + .values({ + collection_id: "collection-1", + collection_slug: "articles", + run_token: "run-1", + next_attempt_at: "2000-01-01T00:00:00.000Z", + }) + .execute(); + + await expect(migration.down(ctx.db)).rejects.toThrow(/reconciliation evidence/i); + }); +}); + +async function createActiveCollection( + ctx: DialectTestContext, + slug: string, + options: { captureState?: string; reconciliationRequired?: number } = {}, +): Promise<{ id: string; slug: string }> { + const registry = new SchemaRegistry(ctx.db); + await registry.createCollection({ slug, label: slug }); + const collection = await registry.getCollection(slug); + if (!collection) throw new Error(`Expected ${slug} collection`); + await ctx.db + .insertInto("_emdash_media_usage_index_status") + .values({ + adapter_id: "content-media", + scope_type: "collection", + scope_key: collection.slug, + collection_id: collection.id, + status: "stale", + capture_state: options.captureState ?? "active", + reconciliation_required: options.reconciliationRequired ?? 1, + }) + .execute(); + return { id: collection.id, slug: collection.slug }; +} + +async function activateAndSeed(ctx: DialectTestContext): Promise { + await ctx.db + .updateTable("_emdash_media_usage_activation") + .set({ state: "active" }) + .where("task_key", "=", "incremental_capture") + .execute(); + const repository = new MediaUsageReconciliationRepository(ctx.db); + await expect(repository.seedNextCandidate()).resolves.toBe(true); +} + +async function reconciliationIdentities( + ctx: DialectTestContext, +): Promise<{ id: string; slug: string }[]> { + const rows = await ctx.db + .selectFrom("_emdash_media_usage_reconciliations") + .select(["collection_id", "collection_slug"]) + .orderBy("collection_id") + .execute(); + return rows.map((row) => ({ id: row.collection_id, slug: row.collection_slug })); +} diff --git a/packages/core/tests/integration/database/media-usage-reconciliation-scan.test.ts b/packages/core/tests/integration/database/media-usage-reconciliation-scan.test.ts new file mode 100644 index 0000000000..f5b98de0cf --- /dev/null +++ b/packages/core/tests/integration/database/media-usage-reconciliation-scan.test.ts @@ -0,0 +1,324 @@ +import { sql } from "kysely"; +import { afterEach, beforeEach, expect, it } from "vitest"; + +import { MediaUsageRepository } from "../../../src/database/repositories/media-usage.js"; +import { processClaimedMediaUsageReconciliationScan } from "../../../src/media/usage/reconciliation-processor.js"; +import { MediaUsageReconciliationRepository } from "../../../src/media/usage/reconciliation.js"; +import { SchemaRegistry } from "../../../src/schema/registry.js"; +import { + describeEachDialect, + setupForDialect, + teardownForDialect, + type DialectTestContext, +} from "../../utils/test-db.js"; + +describeEachDialect("media usage reconciliation scan", (dialect) => { + let ctx: DialectTestContext; + + beforeEach(async () => { + ctx = await setupForDialect(dialect); + }); + + afterEach(async () => { + await teardownForDialect(ctx); + }); + + it("pages through the captured bound without resetting equal or newer work", async () => { + const collection = await createScanCollection(ctx, 51); + await ctx.db + .insertInto("_emdash_media_usage_work") + .values([ + { + collection_id: collection.id, + collection_slug: collection.slug, + content_id: "entry-001", + change_epoch: 0, + state: "failed", + next_attempt_at: "2000-01-01T00:00:00.000Z", + last_error_code: "OLD_FAILURE", + }, + { + collection_id: collection.id, + collection_slug: collection.slug, + content_id: "entry-002", + change_epoch: 99, + state: "failed", + next_attempt_at: "2000-01-01T00:00:00.000Z", + last_error_code: "NEWER_FAILURE", + }, + ]) + .execute(); + + await expect(claimAndScan(ctx)).resolves.toBe("advanced"); + const coordinator = await ctx.db + .selectFrom("_emdash_media_usage_reconciliations") + .select(["target_epoch", "field_fingerprint", "scan_upper_id", "scan_cursor"]) + .where("collection_id", "=", collection.id) + .executeTakeFirstOrThrow(); + expect(Number(coordinator.target_epoch)).toBe(1); + expect(coordinator).toMatchObject({ + scan_upper_id: "entry-050", + scan_cursor: "entry-049", + }); + expect(coordinator.field_fingerprint).toMatch(/^media-usage-fields:v1:sha256:[a-f0-9]{64}$/); + const workCount = await ctx.db + .selectFrom("_emdash_media_usage_work") + .select((eb) => eb.fn.countAll().as("count")) + .where("collection_id", "=", collection.id) + .executeTakeFirstOrThrow(); + expect(Number(workCount.count)).toBe(50); + expect(await workState(ctx, collection.id, "entry-001")).toMatchObject({ + change_epoch: 1, + work_version: 2, + state: "pending", + last_error_code: null, + }); + expect(await workState(ctx, collection.id, "entry-002")).toMatchObject({ + change_epoch: 99, + work_version: 1, + state: "failed", + last_error_code: "NEWER_FAILURE", + }); + + const versionsBeforeReplay = await workVersions(ctx, collection.id); + await ctx.db + .updateTable("_emdash_media_usage_reconciliations") + .set({ scan_cursor: null, state: "pending", next_attempt_at: "2000-01-01T00:00:00.000Z" }) + .where("collection_id", "=", collection.id) + .execute(); + await expect(claimAndScan(ctx)).resolves.toBe("advanced"); + expect(await workVersions(ctx, collection.id)).toEqual(versionsBeforeReplay); + + await expect(claimAndScan(ctx)).resolves.toBe("advanced"); + expect( + await ctx.db + .selectFrom("_emdash_media_usage_reconciliations") + .select("scan_cursor") + .where("collection_id", "=", collection.id) + .executeTakeFirstOrThrow(), + ).toEqual({ scan_cursor: "entry-050" }); + expect(await claimAndScan(ctx)).toBe("exhausted"); + }); + + it("invalidates stale canonical publication after deletion or a newer version", async () => { + const registry = new SchemaRegistry(ctx.db); + await registry.createCollection({ slug: "guarded", label: "Guarded" }); + const collection = await registry.getCollection("guarded"); + if (!collection) throw new Error("Expected guarded collection"); + await sql` + INSERT INTO ${sql.ref("ec_guarded")} (id, slug, version, updated_at) + VALUES ('entry-1', 'entry-1', 1, '2026-08-12T10:00:00.000Z') + `.execute(ctx.db); + const repository = new MediaUsageRepository(ctx.db); + const source = canonicalSource(collection.id, 1, "2026-08-12T10:00:00.000Z"); + + await sql`DELETE FROM ${sql.ref("ec_guarded")} WHERE id = 'entry-1'`.execute(ctx.db); + await expect(repository.replaceSourceIfMatching(source, [], null)).resolves.toMatchObject({ + replaced: false, + }); + expect(await repository.findSource(source.sourceKey)).toBeNull(); + + await sql` + INSERT INTO ${sql.ref("ec_guarded")} (id, slug, version, updated_at) + VALUES ('entry-1', 'entry-1', 2, '2026-08-12T10:05:00.000Z') + `.execute(ctx.db); + await expect(repository.replaceSourceIfMatching(source, [], null)).resolves.toMatchObject({ + replaced: false, + }); + expect(await repository.findSource(source.sourceKey)).toBeNull(); + + await ctx.db + .insertInto("revisions") + .values({ + id: "live-2", + collection: "guarded", + entry_id: "entry-1", + data: "{}", + author_id: null, + }) + .execute(); + await sql` + UPDATE ${sql.ref("ec_guarded")} + SET live_revision_id = 'live-2' + WHERE id = 'entry-1' + `.execute(ctx.db); + const staleRevision = canonicalSource( + collection.id, + 2, + "2026-08-12T10:05:00.000Z", + "guarded", + "live-1", + ); + await expect( + repository.replaceSourceIfMatching(staleRevision, [], null), + ).resolves.toMatchObject({ replaced: false }); + + const current = canonicalSource( + collection.id, + 2, + "2026-08-12T10:05:00.000Z", + "guarded", + "live-2", + ); + await expect(repository.replaceSourceIfMatching(current, [], null)).resolves.toMatchObject({ + replaced: true, + }); + }); + + it("skips content work when the collection has no media-bearing fields", async () => { + const registry = new SchemaRegistry(ctx.db); + await registry.createCollection({ slug: "plain", label: "Plain" }); + const collection = await registry.getCollection("plain"); + if (!collection) throw new Error("Expected plain collection"); + await sql`INSERT INTO ${sql.ref("ec_plain")} (id, slug) VALUES ('entry-1', 'entry-1')`.execute( + ctx.db, + ); + await activateCollection(ctx, collection); + + await expect(claimAndScan(ctx)).resolves.toBe("exhausted"); + expect( + await ctx.db + .selectFrom("_emdash_media_usage_work") + .select("content_id") + .where("collection_id", "=", collection.id) + .execute(), + ).toEqual([]); + expect( + await ctx.db + .selectFrom("_emdash_media_usage_reconciliations") + .select("scan_upper_id") + .where("collection_id", "=", collection.id) + .executeTakeFirstOrThrow(), + ).toEqual({ scan_upper_id: null }); + }); + + it("applies the content identity guard to attempted canonical sources", async () => { + const registry = new SchemaRegistry(ctx.db); + await registry.createCollection({ slug: "attempted", label: "Attempted" }); + const collection = await registry.getCollection("attempted"); + if (!collection) throw new Error("Expected attempted collection"); + await sql` + INSERT INTO ${sql.ref("ec_attempted")} (id, slug, version, updated_at) + VALUES ('entry-1', 'entry-1', 2, '2026-08-12T10:05:00.000Z') + `.execute(ctx.db); + const repository = new MediaUsageRepository(ctx.db); + const stale = { + ...canonicalSource(collection.id, 1, "2026-08-12T10:00:00.000Z", "attempted"), + lastErrorCode: "DRAFT_REVISION_NOT_FOUND", + }; + + await expect(repository.markSourceAttemptedIfMatching(stale, null)).resolves.toMatchObject({ + attempted: false, + }); + expect(await repository.findSource(stale.sourceKey)).toBeNull(); + }); +}); + +async function createScanCollection( + ctx: DialectTestContext, + entryCount: number, +): Promise<{ id: string; slug: string }> { + const registry = new SchemaRegistry(ctx.db); + await registry.createCollection({ slug: "articles", label: "Articles" }); + await registry.createField("articles", { slug: "hero", label: "Hero", type: "image" }); + const collection = await registry.getCollection("articles"); + if (!collection) throw new Error("Expected articles collection"); + for (let index = 0; index < entryCount; index++) { + const id = `entry-${String(index).padStart(3, "0")}`; + await sql`INSERT INTO ${sql.ref("ec_articles")} (id, slug) VALUES (${id}, ${id})`.execute( + ctx.db, + ); + } + await activateCollection(ctx, collection); + return { id: collection.id, slug: collection.slug }; +} + +async function activateCollection( + ctx: DialectTestContext, + collection: { id: string; slug: string }, +): Promise { + await ctx.db + .insertInto("_emdash_media_usage_index_status") + .values({ + adapter_id: "content-media", + scope_type: "collection", + scope_key: collection.slug, + collection_id: collection.id, + status: "stale", + capture_state: "active", + reconciliation_required: 1, + }) + .onConflict((conflict) => + conflict.columns(["adapter_id", "scope_type", "scope_key"]).doUpdateSet({ + collection_id: collection.id, + status: "stale", + capture_state: "active", + reconciliation_required: 1, + }), + ) + .execute(); + await ctx.db + .updateTable("_emdash_media_usage_activation") + .set({ state: "active" }) + .where("task_key", "=", "incremental_capture") + .execute(); + const repository = new MediaUsageReconciliationRepository(ctx.db); + await repository.seedNextCandidate(); +} + +async function claimAndScan(ctx: DialectTestContext) { + const repository = new MediaUsageReconciliationRepository(ctx.db); + const [candidate] = await repository.findDue(4); + if (!candidate) throw new Error("Expected a due reconciliation candidate"); + const claim = await repository.claim({ + collectionId: candidate.collectionId, + runToken: candidate.runToken, + leaseDurationSeconds: 60, + }); + if (!claim) throw new Error("Expected a reconciliation claim"); + return processClaimedMediaUsageReconciliationScan(ctx.db, claim); +} + +function canonicalSource( + collectionId: string, + version: number, + updatedAt: string, + collectionSlug = "guarded", + revisionId: string | null = null, +) { + return { + sourceKey: `content-id:${collectionId}:entry-1:columns`, + sourceType: "content", + collectionId, + collectionSlug, + contentId: "entry-1", + sourceVariant: "columns" as const, + revisionId, + sourceVersion: version, + sourceUpdatedAt: updatedAt, + identityVersion: 1, + }; +} + +async function workState(ctx: DialectTestContext, collectionId: string, contentId: string) { + const row = await ctx.db + .selectFrom("_emdash_media_usage_work") + .select(["change_epoch", "work_version", "state", "last_error_code"]) + .where("collection_id", "=", collectionId) + .where("content_id", "=", contentId) + .executeTakeFirstOrThrow(); + return { + ...row, + change_epoch: Number(row.change_epoch), + work_version: Number(row.work_version), + }; +} + +async function workVersions(ctx: DialectTestContext, collectionId: string) { + return ctx.db + .selectFrom("_emdash_media_usage_work") + .select(["content_id", "work_version"]) + .where("collection_id", "=", collectionId) + .orderBy("content_id") + .execute(); +} diff --git a/packages/core/tests/integration/database/media-usage-repository.test.ts b/packages/core/tests/integration/database/media-usage-repository.test.ts index 038f81635d..146cb2fe87 100644 --- a/packages/core/tests/integration/database/media-usage-repository.test.ts +++ b/packages/core/tests/integration/database/media-usage-repository.test.ts @@ -98,6 +98,7 @@ describeEachDialect("MediaUsageRepository", (dialect) => { .insertInto("_emdash_collections") .values({ id: collectionId, slug: "posts", label: "Posts" }) .execute(); + await installCanonicalContentFixture(ctx, "posts", "entry1", "rev-entry1-columns"); const sourceKey = buildContentMediaUsageSourceKey({ collectionId, collectionSlug: "posts", @@ -110,6 +111,8 @@ describeEachDialect("MediaUsageRepository", (dialect) => { sourceKey, collectionId, identityVersion: 1, + sourceVersion: 1, + sourceUpdatedAt: "2026-08-12T00:00:00.000Z", }), [occurrence("hero", "media-old")], ); @@ -131,6 +134,7 @@ describeEachDialect("MediaUsageRepository", (dialect) => { .insertInto("_emdash_collections") .values({ id: collectionId, slug: "posts", label: "Posts" }) .execute(); + await installCanonicalContentFixture(ctx, "posts", "entry1", "rev-entry1-columns"); const source = contentSource("entry1", "columns", { sourceKey: buildContentMediaUsageSourceKey({ collectionId, @@ -140,6 +144,8 @@ describeEachDialect("MediaUsageRepository", (dialect) => { }), collectionId, identityVersion: 1, + sourceVersion: 1, + sourceUpdatedAt: "2026-08-12T00:00:00.000Z", }); const observed = await repo.replaceSource(source, [occurrence("hero", "media-old")]); await ctx.db.deleteFrom("_emdash_collections").where("id", "=", collectionId).execute(); @@ -1581,6 +1587,31 @@ function occurrence( }; } +async function installCanonicalContentFixture( + ctx: DialectTestContext, + collectionSlug: string, + contentId: string, + liveRevisionId: string, +): Promise { + const tableName = `ec_${collectionSlug}`; + await sql` + CREATE TABLE IF NOT EXISTS ${sql.ref(tableName)} ( + id TEXT PRIMARY KEY, + version INTEGER NOT NULL, + updated_at TEXT NOT NULL, + live_revision_id TEXT, + draft_revision_id TEXT + ) + `.execute(ctx.db); + await sql` + INSERT INTO ${sql.ref(tableName)} ( + id, version, updated_at, live_revision_id, draft_revision_id + ) VALUES ( + ${contentId}, 1, '2026-08-12T00:00:00.000Z', ${liveRevisionId}, NULL + ) + `.execute(ctx.db); +} + async function insertOccurrenceGeneration( ctx: DialectTestContext, sourceKey: string, diff --git a/packages/core/tests/integration/database/migrations.test.ts b/packages/core/tests/integration/database/migrations.test.ts index 239bafe558..852f0c4177 100644 --- a/packages/core/tests/integration/database/migrations.test.ts +++ b/packages/core/tests/integration/database/migrations.test.ts @@ -156,6 +156,7 @@ describe("Database Migrations (Integration)", () => { "063_media_usage_incremental_work", "064_fts_plain_text", "065_media_usage_collection_deletion", + "066_media_usage_reconciliation", ]; await db.deleteFrom("_emdash_migrations").where("name", "in", trailing).execute(); 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 fdf2b50032..e6f124ba0b 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 @@ -5,9 +5,14 @@ import { sql, SqliteDialect } from "kysely"; import { afterEach, describe, expect, it } from "vitest"; import { MediaUsageRepository } from "../../../src/database/repositories/media-usage.js"; -import { EmDashRuntime, type RuntimeDependencies } from "../../../src/emdash-runtime.js"; +import { + EmDashRuntime, + MEDIA_USAGE_MAINTENANCE_QUERY_RESERVATIONS, + type RuntimeDependencies, +} from "../../../src/emdash-runtime.js"; import { installMediaUsageCaptureTriggers } from "../../../src/media/usage/capture-triggers.js"; import type { CronScheduler, SystemCleanupFn } from "../../../src/plugins/scheduler/types.js"; +import { createRequestMetrics, runWithContext } from "../../../src/request-context.js"; describe("media usage scheduled drivers", () => { let runtime: EmDashRuntime | null = null; @@ -17,12 +22,17 @@ describe("media usage scheduled drivers", () => { runtime = null; }); - it("drains bounded work from the Cloudflare scheduled entry point", async () => { + it("keeps general maintenance unchanged and drains work from the dedicated lane", async () => { runtime = await EmDashRuntime.create(createDeps(null)); const fixture = await activateCollection(runtime, "cloudflare_posts"); await insertEntry(runtime, fixture.tableName, "entry-1"); await runtime.runScheduledTasks(); + expect(await countWork(runtime)).toBe(1); + await expect(runtime.runScheduledMediaUsageTasks()).resolves.toMatchObject({ + outcome: "processed", + taskClass: "entry_work", + }); expect(await countWork(runtime)).toBe(0); expect( @@ -48,12 +58,28 @@ describe("media usage scheduled drivers", () => { ).not.toBeNull(); }); + it("drains bounded work through a legacy Node scheduler cleanup callback", async () => { + const scheduler = new LegacyCapturingScheduler(); + runtime = await EmDashRuntime.create(createDeps(() => scheduler)); + const fixture = await activateCollection(runtime, "legacy_node_posts"); + await insertEntry(runtime, fixture.tableName, "entry-1"); + + await scheduler.runMaintenance(); + + expect(await countWork(runtime)).toBe(0); + expect( + await new MediaUsageRepository(runtime.db).findSource( + canonicalSourceKey(fixture.collectionId, "entry-1"), + ), + ).not.toBeNull(); + }); + it("advances bounded collection deletion from the Cloudflare scheduled entry point", async () => { runtime = await EmDashRuntime.create(createDeps(null)); const fixture = await activateCollection(runtime, "cloudflare_delete"); await runtime.schemaRegistry.deleteCollection("cloudflare_delete", { force: true }); - - await runtime.runScheduledTasks(); + await runtime.runScheduledMediaUsageTasks(); + await runtime.runScheduledMediaUsageTasks(); expect(await deletionPhase(runtime, fixture.collectionId)).toBe("sources"); }); @@ -64,6 +90,7 @@ describe("media usage scheduled drivers", () => { const fixture = await activateCollection(runtime, "node_delete"); await runtime.schemaRegistry.deleteCollection("node_delete", { force: true }); + await scheduler.runMaintenance(); await scheduler.runMaintenance(); expect(await deletionPhase(runtime, fixture.collectionId)).toBe("sources"); @@ -89,10 +116,119 @@ describe("media usage scheduled drivers", () => { ), ).not.toBeNull(); }); + + it("persists a fair entry, deletion, reconciliation turn sequence", async () => { + runtime = await EmDashRuntime.create(createDeps(null)); + const fixture = await activateCollection(runtime, "fair_posts"); + await runtime.db + .updateTable("_emdash_media_usage_index_status") + .set({ status: "stale", reconciliation_required: 1 }) + .where("collection_id", "=", fixture.collectionId) + .execute(); + + const classes = []; + for (let index = 0; index < 3; index++) { + const result = await runtime.runScheduledMediaUsageTasks(); + classes.push(result.taskClass); + } + expect(classes).toEqual(["entry_work", "collection_deletion", "reconciliation"]); + expect( + await runtime.db + .selectFrom("_emdash_media_usage_reconciliations") + .select("collection_id") + .where("collection_id", "=", fixture.collectionId) + .executeTakeFirst(), + ).toBeDefined(); + }); + + it("does not advance the turn or spend class queries before activation", async () => { + runtime = await EmDashRuntime.create(createDeps(null)); + const before = await maintenanceTurn(runtime); + await expect(runtime.runScheduledMediaUsageTasks()).resolves.toEqual({ + outcome: "inactive", + taskClass: null, + turn: null, + }); + expect(await maintenanceTurn(runtime)).toBe(before); + }); + + it("reserves ten queries of headroom before changing the persisted turn", async () => { + runtime = await EmDashRuntime.create(createDeps(null)); + await runtime.db + .updateTable("_emdash_media_usage_activation") + .set({ state: "active" }) + .where("task_key", "=", "incremental_capture") + .execute(); + const before = await maintenanceTurn(runtime); + const metrics = createRequestMetrics(performance.now()); + metrics.dbCount = + MEDIA_USAGE_MAINTENANCE_QUERY_RESERVATIONS.eventCeiling - + MEDIA_USAGE_MAINTENANCE_QUERY_RESERVATIONS.maxClassQueries; + + const result = await runWithContext({ editMode: false, metrics }, () => + runtime!.runScheduledMediaUsageTasks(), + ); + expect(result).toEqual({ outcome: "admission_closed", taskClass: null, turn: null }); + expect(await maintenanceTurn(runtime)).toBe(before); + }); + + it("measures every mutation class within its exported event reservation", async () => { + runtime = await EmDashRuntime.create(createDeps(null)); + const work = await activateCollection(runtime, "measure_work"); + await insertEntry(runtime, work.tableName, "entry-1"); + await activateCollection(runtime, "measure_delete"); + await runtime.schemaRegistry.deleteCollection("measure_delete", { force: true }); + const reconciliation = await activateCollection(runtime, "measure_reconcile"); + await runtime.db + .updateTable("_emdash_media_usage_index_status") + .set({ status: "stale", reconciliation_required: 1 }) + .where("collection_id", "=", reconciliation.collectionId) + .execute(); + + const expected = [ + ["entry_work", MEDIA_USAGE_MAINTENANCE_QUERY_RESERVATIONS.entryWork], + ["collection_deletion", MEDIA_USAGE_MAINTENANCE_QUERY_RESERVATIONS.collectionDeletion], + ["reconciliation", MEDIA_USAGE_MAINTENANCE_QUERY_RESERVATIONS.reconciliation], + ] as const; + for (const [taskClass, reservation] of expected) { + const metrics = createRequestMetrics(performance.now()); + const result = await runWithContext({ editMode: false, metrics }, () => + runtime!.runScheduledMediaUsageTasks(), + ); + expect(result.taskClass).toBe(taskClass); + expect(metrics.dbCount).toBeLessThanOrEqual(1 + reservation); + expect(metrics.dbCount).toBeLessThanOrEqual( + MEDIA_USAGE_MAINTENANCE_QUERY_RESERVATIONS.eventCeiling, + ); + } + }); }); class CapturingScheduler implements CronScheduler { private maintenance: SystemCleanupFn | null = null; + private mediaUsageMaintenance: SystemCleanupFn | null = null; + + setSystemCleanup(fn: SystemCleanupFn): void { + this.maintenance = fn; + } + setMediaUsageMaintenance(fn: SystemCleanupFn): void { + this.mediaUsageMaintenance = fn; + } + + start(): void {} + stop(): void {} + reschedule(): void {} + + async runMaintenance(): Promise { + if (!this.maintenance) throw new Error("Expected Node maintenance callback"); + await this.maintenance(); + if (!this.mediaUsageMaintenance) throw new Error("Expected Media Usage maintenance callback"); + await this.mediaUsageMaintenance(); + } +} + +class LegacyCapturingScheduler implements CronScheduler { + private maintenance: SystemCleanupFn | null = null; setSystemCleanup(fn: SystemCleanupFn): void { this.maintenance = fn; @@ -195,6 +331,15 @@ async function deletionPhase(runtime: EmDashRuntime, collectionId: string): Prom return row?.phase ?? null; } +async function maintenanceTurn(runtime: EmDashRuntime): Promise { + const row = await runtime.db + .selectFrom("_emdash_media_usage_activation") + .select("media_usage_maintenance_turn") + .where("task_key", "=", "incremental_capture") + .executeTakeFirstOrThrow(); + return row.media_usage_maintenance_turn; +} + function canonicalSourceKey(collectionId: string, contentId: string): string { return `content:${collectionId}:${contentId}:columns`; } diff --git a/templates/blog-cloudflare/src/worker.ts b/templates/blog-cloudflare/src/worker.ts index df373a0194..75fe3e2743 100644 --- a/templates/blog-cloudflare/src/worker.ts +++ b/templates/blog-cloudflare/src/worker.ts @@ -1,4 +1,8 @@ -// Worker entry: Astro's fetch handler plus EmDash's scheduled() handler, which -// the Cron Trigger in wrangler.jsonc drives. PluginBridge is the sandbox -// Durable Object, re-exported here so its binding resolves. -export { default, PluginBridge } from "@emdash-cms/cloudflare/worker"; +import handler, { createScheduledHandler, PluginBridge } from "@emdash-cms/cloudflare/worker"; + +export { PluginBridge }; + +export default { + ...handler, + scheduled: createScheduledHandler(), +}; diff --git a/templates/blog-cloudflare/wrangler.jsonc b/templates/blog-cloudflare/wrangler.jsonc index f6baeeca71..23632b0e98 100644 --- a/templates/blog-cloudflare/wrangler.jsonc +++ b/templates/blog-cloudflare/wrangler.jsonc @@ -22,8 +22,8 @@ "binding": "LOADER", }, ], - // Drives scheduled publishing, plugin cron, and maintenance (see src/worker.ts) + // General maintenance plus the bounded Media Usage lane (see src/worker.ts) "triggers": { - "crons": ["* * * * *"], + "crons": ["* * * * *", "*/2 * * * *"], }, } diff --git a/templates/marketing-cloudflare/src/worker.ts b/templates/marketing-cloudflare/src/worker.ts index df373a0194..75fe3e2743 100644 --- a/templates/marketing-cloudflare/src/worker.ts +++ b/templates/marketing-cloudflare/src/worker.ts @@ -1,4 +1,8 @@ -// Worker entry: Astro's fetch handler plus EmDash's scheduled() handler, which -// the Cron Trigger in wrangler.jsonc drives. PluginBridge is the sandbox -// Durable Object, re-exported here so its binding resolves. -export { default, PluginBridge } from "@emdash-cms/cloudflare/worker"; +import handler, { createScheduledHandler, PluginBridge } from "@emdash-cms/cloudflare/worker"; + +export { PluginBridge }; + +export default { + ...handler, + scheduled: createScheduledHandler(), +}; diff --git a/templates/marketing-cloudflare/wrangler.jsonc b/templates/marketing-cloudflare/wrangler.jsonc index c020df0791..e5e23e764d 100644 --- a/templates/marketing-cloudflare/wrangler.jsonc +++ b/templates/marketing-cloudflare/wrangler.jsonc @@ -22,8 +22,8 @@ "binding": "LOADER", }, ], - // Drives scheduled publishing, plugin cron, and maintenance (see src/worker.ts) + // General maintenance plus the bounded Media Usage lane (see src/worker.ts) "triggers": { - "crons": ["* * * * *"], + "crons": ["* * * * *", "*/2 * * * *"], }, } diff --git a/templates/portfolio-cloudflare/src/worker.ts b/templates/portfolio-cloudflare/src/worker.ts index df373a0194..75fe3e2743 100644 --- a/templates/portfolio-cloudflare/src/worker.ts +++ b/templates/portfolio-cloudflare/src/worker.ts @@ -1,4 +1,8 @@ -// Worker entry: Astro's fetch handler plus EmDash's scheduled() handler, which -// the Cron Trigger in wrangler.jsonc drives. PluginBridge is the sandbox -// Durable Object, re-exported here so its binding resolves. -export { default, PluginBridge } from "@emdash-cms/cloudflare/worker"; +import handler, { createScheduledHandler, PluginBridge } from "@emdash-cms/cloudflare/worker"; + +export { PluginBridge }; + +export default { + ...handler, + scheduled: createScheduledHandler(), +}; diff --git a/templates/portfolio-cloudflare/wrangler.jsonc b/templates/portfolio-cloudflare/wrangler.jsonc index 62ee430e60..792c1fda4d 100644 --- a/templates/portfolio-cloudflare/wrangler.jsonc +++ b/templates/portfolio-cloudflare/wrangler.jsonc @@ -22,8 +22,8 @@ "binding": "LOADER", }, ], - // Drives scheduled publishing, plugin cron, and maintenance (see src/worker.ts) + // General maintenance plus the bounded Media Usage lane (see src/worker.ts) "triggers": { - "crons": ["* * * * *"], + "crons": ["* * * * *", "*/2 * * * *"], }, } diff --git a/templates/starter-cloudflare/src/worker.ts b/templates/starter-cloudflare/src/worker.ts index df373a0194..75fe3e2743 100644 --- a/templates/starter-cloudflare/src/worker.ts +++ b/templates/starter-cloudflare/src/worker.ts @@ -1,4 +1,8 @@ -// Worker entry: Astro's fetch handler plus EmDash's scheduled() handler, which -// the Cron Trigger in wrangler.jsonc drives. PluginBridge is the sandbox -// Durable Object, re-exported here so its binding resolves. -export { default, PluginBridge } from "@emdash-cms/cloudflare/worker"; +import handler, { createScheduledHandler, PluginBridge } from "@emdash-cms/cloudflare/worker"; + +export { PluginBridge }; + +export default { + ...handler, + scheduled: createScheduledHandler(), +}; diff --git a/templates/starter-cloudflare/wrangler.jsonc b/templates/starter-cloudflare/wrangler.jsonc index f6baeeca71..23632b0e98 100644 --- a/templates/starter-cloudflare/wrangler.jsonc +++ b/templates/starter-cloudflare/wrangler.jsonc @@ -22,8 +22,8 @@ "binding": "LOADER", }, ], - // Drives scheduled publishing, plugin cron, and maintenance (see src/worker.ts) + // General maintenance plus the bounded Media Usage lane (see src/worker.ts) "triggers": { - "crons": ["* * * * *"], + "crons": ["* * * * *", "*/2 * * * *"], }, }