diff --git a/.changeset/controlled-media-usage-activation.md b/.changeset/controlled-media-usage-activation.md new file mode 100644 index 0000000000..3078220039 --- /dev/null +++ b/.changeset/controlled-media-usage-activation.md @@ -0,0 +1,5 @@ +--- +"emdash": patch +--- + +Adds a one-time, administrator-controlled process for enabling automatic media usage indexing in production. diff --git a/docs/src/content/docs/deployment/cloudflare.mdx b/docs/src/content/docs/deployment/cloudflare.mdx index ceea9d5ec2..8e52d3650f 100644 --- a/docs/src/content/docs/deployment/cloudflare.mdx +++ b/docs/src/content/docs/deployment/cloudflare.mdx @@ -128,6 +128,13 @@ To use different schedules, set the corresponding `generalCron` or `mediaUsageCr Without the general trigger, scheduled publishing and plugin cron do not run. Without the dedicated Media Usage trigger, automatic historical reconciliation cannot progress. Local `astro dev` still uses the in-process scheduler. +### Enable automatic media usage indexing + +Keep `mediaUsageCron` running while you enable automatic media usage indexing. Pause all application +and direct database writes, follow [Enable automatic media usage +indexing](/reference/rest-api/#enable-automatic-media-usage-indexing), then resume writes when the +endpoint returns `active`. Existing content is indexed in the background. + ## Deploy Deploy to Cloudflare Workers: diff --git a/docs/src/content/docs/deployment/nodejs.mdx b/docs/src/content/docs/deployment/nodejs.mdx index 9b7b3b3465..89b8196815 100644 --- a/docs/src/content/docs/deployment/nodejs.mdx +++ b/docs/src/content/docs/deployment/nodejs.mdx @@ -53,6 +53,21 @@ export default defineConfig({ The server runs on `http://localhost:4321` by default. Migrations are applied on the first request. If the database is empty and setup hasn't been completed, your seed file (or the built-in default if you don't have one) is also applied on that first request. +## Scheduled Tasks + +The built-in scheduler runs only while a Node process is running. It handles scheduled publishing, +plugin tasks, and background media indexing. + +Keep at least one Node process running continuously in production. If all processes stop or sleep, +scheduled tasks pause. + +### Enable automatic media usage indexing + +Keep at least one Node process running while you enable automatic media usage indexing. Pause all +application and direct database writes, follow [Enable automatic media usage +indexing](/reference/rest-api/#enable-automatic-media-usage-indexing), then resume writes when the +endpoint returns `active`. Existing content is indexed in the background. + ## Production Storage For production, use S3-compatible storage instead of local filesystem: diff --git a/docs/src/content/docs/reference/rest-api.mdx b/docs/src/content/docs/reference/rest-api.mdx index 593d6da617..607e52587f 100644 --- a/docs/src/content/docs/reference/rest-api.mdx +++ b/docs/src/content/docs/reference/rest-api.mdx @@ -364,6 +364,71 @@ Content-Type: application/json DELETE /_emdash/api/media/:id ``` +### Enable automatic media usage indexing + +Automatic media usage indexing must be enabled once for each production site. Writes must be +paused while EmDash prepares each collection so that no changes are missed. + +Both endpoints require `schema:manage`. Bearer tokens also require the `admin` scope. + +#### Check the current state + +```http +GET /_emdash/api/admin/media-usage/activation +``` + +This request does not change anything. It returns one of these states: + +- `expanded`: automatic indexing is not enabled. +- `activating`: EmDash is preparing the site's collections. +- `active`: EmDash tracks changes to media references in content. + +Status responses do not include internal lock data or raw database errors. + +#### Prepare the next collection + +```http +POST /_emdash/api/admin/media-usage/activation +Content-Type: application/json +X-EmDash-Request: 1 + +{ + "writersDrained": true, + "maintenanceReady": true +} +``` + +Each request prepares one collection. Send one request at a time until the state becomes `active`. + +Set both fields to `true`: + +- `writersDrained`: Application and direct database writes have stopped, and any writes already in + progress have finished. +- `maintenanceReady`: Scheduled background tasks are running through `mediaUsageCron` on Cloudflare + or the built-in scheduler on Node.js. + +#### Enable indexing in production + +1. Confirm that scheduled background tasks are running: `mediaUsageCron` on Cloudflare, or the + built-in scheduler on Node.js. +2. Stop all application and direct database writes. Wait for writes already in progress to finish. +3. Call `GET` to check the current state. +4. Call `POST` one request at a time until the state becomes `active`. +5. Resume writes. +6. Keep scheduled background tasks running while EmDash indexes existing content. + +If `POST` times out or returns `409` or `500`, call `GET` before sending another request. If +`lastErrorCode` is set, keep writes stopped, check the application logs, fix the problem, and try +again. Do not edit EmDash's internal database tables. + + + +When the state is `active`, EmDash tracks changes to media references in content. Existing content +may still be indexing in the background. + ### List Media Usage Work ```http diff --git a/packages/core/src/api/errors.ts b/packages/core/src/api/errors.ts index f211b7d4ba..fefbbb3cbd 100644 --- a/packages/core/src/api/errors.ts +++ b/packages/core/src/api/errors.ts @@ -92,6 +92,11 @@ export const ErrorCode = { MEDIA_USAGE_WORK_RETRY_ERROR: "MEDIA_USAGE_WORK_RETRY_ERROR", MEDIA_USAGE_COLLECTION_DELETION_LIST_ERROR: "MEDIA_USAGE_COLLECTION_DELETION_LIST_ERROR", MEDIA_USAGE_COLLECTION_DELETION_RETRY_ERROR: "MEDIA_USAGE_COLLECTION_DELETION_RETRY_ERROR", + MEDIA_USAGE_ACTIVATION_VERSION_MISMATCH: "MEDIA_USAGE_ACTIVATION_VERSION_MISMATCH", + MEDIA_USAGE_ACTIVATION_READ_ERROR: "MEDIA_USAGE_ACTIVATION_READ_ERROR", + MEDIA_USAGE_ACTIVATION_BUSY: "MEDIA_USAGE_ACTIVATION_BUSY", + MEDIA_USAGE_ACTIVATION_CONFLICT: "MEDIA_USAGE_ACTIVATION_CONFLICT", + MEDIA_USAGE_ACTIVATION_ADVANCE_ERROR: "MEDIA_USAGE_ACTIVATION_ADVANCE_ERROR", WORK_LEASE_ACTIVE: "WORK_LEASE_ACTIVE", WORK_CHANGED: "WORK_CHANGED", NO_STORAGE: "NO_STORAGE", @@ -461,6 +466,9 @@ export function mapErrorStatus(code: string | undefined): number { case ErrorCode.ENV_INCOMPATIBLE: case ErrorCode.WORK_LEASE_ACTIVE: case ErrorCode.WORK_CHANGED: + case ErrorCode.MEDIA_USAGE_ACTIVATION_VERSION_MISMATCH: + case ErrorCode.MEDIA_USAGE_ACTIVATION_BUSY: + case ErrorCode.MEDIA_USAGE_ACTIVATION_CONFLICT: return 409; // 410 Gone diff --git a/packages/core/src/api/handlers/media-usage-activation.ts b/packages/core/src/api/handlers/media-usage-activation.ts new file mode 100644 index 0000000000..de3877e43c --- /dev/null +++ b/packages/core/src/api/handlers/media-usage-activation.ts @@ -0,0 +1,102 @@ +import type { Kysely } from "kysely"; + +import type { Database } from "../../database/types.js"; +import { + activateMediaUsageCapture, + getMediaUsageActivationStatus, + MediaUsageActivationVersionMismatchError, +} from "../../media/usage/activation.js"; +import { ErrorCode } from "../errors.js"; +import type { + MediaUsageActivationAdvanceRequest, + MediaUsageActivationAdvanceResponse, + MediaUsageActivationStatus, +} from "../schemas/media-usage.js"; +import type { ApiResult } from "../types.js"; + +export async function handleMediaUsageActivationStatus( + db: Kysely, +): Promise> { + try { + return { success: true, data: await getMediaUsageActivationStatus(db) }; + } catch (error) { + if (error instanceof MediaUsageActivationVersionMismatchError) { + return { + success: false, + error: { + code: ErrorCode.MEDIA_USAGE_ACTIVATION_VERSION_MISMATCH, + message: "Media usage activation version is incompatible with this runtime", + }, + }; + } + console.error("[media-usage:activation] status read failed:", error); + return { + success: false, + error: { + code: ErrorCode.MEDIA_USAGE_ACTIVATION_READ_ERROR, + message: "Failed to read media usage activation status", + }, + }; + } +} + +export async function handleMediaUsageActivationAdvance( + db: Kysely, + input: MediaUsageActivationAdvanceRequest, +): Promise> { + try { + const result = await activateMediaUsageCapture(db, { + writersDrained: input.writersDrained, + }); + if (result.outcome === "lease_active") { + return { + success: false, + error: { + code: ErrorCode.MEDIA_USAGE_ACTIVATION_BUSY, + message: "Media usage activation is already in progress", + details: { leaseExpiresAt: result.leaseExpiresAt }, + }, + }; + } + if (result.outcome === "conflict") { + return { + success: false, + error: { + code: ErrorCode.MEDIA_USAGE_ACTIVATION_CONFLICT, + message: "Media usage activation ownership changed", + }, + }; + } + + const activation = await getMediaUsageActivationStatus(db); + if (activation.state === "expanded") { + throw new Error("Media usage activation did not advance"); + } + return { + success: true, + data: { + outcome: activation.state, + processedCollections: result.processedCollections, + activation, + }, + }; + } catch (error) { + if (error instanceof MediaUsageActivationVersionMismatchError) { + return { + success: false, + error: { + code: ErrorCode.MEDIA_USAGE_ACTIVATION_VERSION_MISMATCH, + message: "Media usage activation version is incompatible with this runtime", + }, + }; + } + console.error("[media-usage:activation] advance failed:", error); + return { + success: false, + error: { + code: ErrorCode.MEDIA_USAGE_ACTIVATION_ADVANCE_ERROR, + message: "Failed to advance media usage activation", + }, + }; + } +} diff --git a/packages/core/src/api/openapi/document.ts b/packages/core/src/api/openapi/document.ts index c7cae3bf61..37a0125000 100644 --- a/packages/core/src/api/openapi/document.ts +++ b/packages/core/src/api/openapi/document.ts @@ -44,6 +44,10 @@ import { mediaUsageCollectionDeletionListResponseSchema, mediaUsageCollectionDeletionRetryBody, mediaUsageCollectionDeletionRetryResponseSchema, + mediaUsageActivationStatusSchema, + mediaUsageActivationAdvanceBody, + mediaUsageActivationAdvanceResponseSchema, + mediaUsageActivationConflictSchema, mediaUsageRepairBody, mediaUsageRepairResponseSchema, mediaUsageWorkListQuery, @@ -822,6 +826,52 @@ function buildMediaPaths(maxUploadSize: number) { }, }, }, + "/_emdash/api/admin/media-usage/activation": { + get: { + operationId: "getMediaUsageActivation", + summary: "Get media usage activation status", + description: + "Returns the redacted status of controlled Media Usage capture activation. This operation is read-only and does not start or resume activation. Requires `schema:manage`; bearer tokens also require the `admin` scope.", + tags: ["Media"], + responses: { + "200": { + description: "Media usage activation status", + content: { + [JSON_CONTENT]: { schema: successEnvelope(mediaUsageActivationStatusSchema) }, + }, + }, + ...authErrors, + ...standardErrors(409, 500), + }, + }, + post: { + operationId: "advanceMediaUsageActivation", + summary: "Advance media usage activation", + description: + "Starts, resumes, or retries exactly one bounded activation batch after the operator confirms that all writers are drained and automatic maintenance is ready. Requires `schema:manage`; bearer tokens also require the `admin` scope.", + tags: ["Media"], + requestBody: { + required: true, + content: { [JSON_CONTENT]: { schema: mediaUsageActivationAdvanceBody } }, + }, + responses: { + "200": { + description: "Current media usage activation progress", + content: { + [JSON_CONTENT]: { + schema: successEnvelope(mediaUsageActivationAdvanceResponseSchema), + }, + }, + }, + ...authErrors, + ...standardErrors(400, 500), + "409": { + description: "Activation is busy, changed ownership, or is incompatible", + content: { [JSON_CONTENT]: { schema: mediaUsageActivationConflictSchema } }, + }, + }, + }, + }, "/_emdash/api/admin/media-usage/work/retry": { post: { operationId: "retryMediaUsageWork", diff --git a/packages/core/src/api/schemas/media-usage.ts b/packages/core/src/api/schemas/media-usage.ts index c2b7358d98..d09345fb34 100644 --- a/packages/core/src/api/schemas/media-usage.ts +++ b/packages/core/src/api/schemas/media-usage.ts @@ -182,6 +182,59 @@ export const mediaUsageWorkRetryConflictSchema = z.object({ ]), }); +export const mediaUsageActivationStateSchema = z + .enum(["expanded", "activating", "active"]) + .meta({ id: "MediaUsageActivationState" }); + +export const mediaUsageActivationStatusSchema = z + .object({ + state: mediaUsageActivationStateSchema, + collectionCursor: z.string().nullable(), + attemptCount: z.number().int().min(0), + drainConfirmedAt: z.string().nullable(), + lastAttemptedAt: z.string().nullable(), + lastErrorCode: z.literal("MEDIA_USAGE_ACTIVATION_FAILED").nullable(), + leaseExpiresAt: z.string().nullable(), + activatedAt: z.string().nullable(), + updatedAt: z.string(), + }) + .meta({ id: "MediaUsageActivationStatus" }); + +export const mediaUsageActivationAdvanceBody = z + .object({ + writersDrained: z.literal(true), + maintenanceReady: z.literal(true), + }) + .strict() + .meta({ id: "MediaUsageActivationAdvanceBody" }); + +export const mediaUsageActivationAdvanceResponseSchema = z + .object({ + outcome: z.enum(["activating", "active"]), + processedCollections: z.number().int().min(0).max(1), + activation: mediaUsageActivationStatusSchema, + }) + .meta({ id: "MediaUsageActivationAdvanceResponse" }); + +export const mediaUsageActivationConflictSchema = z.object({ + success: z.literal(false), + error: z.discriminatedUnion("code", [ + z.object({ + code: z.literal("MEDIA_USAGE_ACTIVATION_BUSY"), + message: z.string(), + details: z.object({ leaseExpiresAt: z.string() }), + }), + z.object({ + code: z.literal("MEDIA_USAGE_ACTIVATION_CONFLICT"), + message: z.string(), + }), + z.object({ + code: z.literal("MEDIA_USAGE_ACTIVATION_VERSION_MISMATCH"), + message: z.string(), + }), + ]), +}); + export const mediaUsageCollectionDeletionStateSchema = z .enum(["pending", "retry", "leased", "failed"]) .meta({ id: "MediaUsageCollectionDeletionState" }); @@ -232,6 +285,11 @@ export type MediaUsageWorkItem = z.infer; export type MediaUsageWorkListResponse = z.infer; export type MediaUsageWorkRetryRequest = z.infer; export type MediaUsageWorkRetryResponse = z.infer; +export type MediaUsageActivationStatus = z.infer; +export type MediaUsageActivationAdvanceRequest = z.infer; +export type MediaUsageActivationAdvanceResponse = z.infer< + typeof mediaUsageActivationAdvanceResponseSchema +>; export type MediaUsageCollectionDeletionListQuery = z.infer< typeof mediaUsageCollectionDeletionListQuery >; diff --git a/packages/core/src/astro/integration/routes.ts b/packages/core/src/astro/integration/routes.ts index be522628cd..9dc1db9957 100644 --- a/packages/core/src/astro/integration/routes.ts +++ b/packages/core/src/astro/integration/routes.ts @@ -258,6 +258,10 @@ export function injectCoreRoutes( pattern: "/_emdash/api/admin/media-usage/work/retry", entrypoint: resolveRoute("api/admin/media-usage/work/retry.ts"), }); + injectRoute({ + pattern: "/_emdash/api/admin/media-usage/activation", + entrypoint: resolveRoute("api/admin/media-usage/activation.ts"), + }); injectRoute({ pattern: "/_emdash/api/admin/media-usage/collection-deletions", entrypoint: resolveRoute("api/admin/media-usage/collection-deletions/index.ts"), diff --git a/packages/core/src/astro/routes/api/admin/media-usage/activation.ts b/packages/core/src/astro/routes/api/admin/media-usage/activation.ts new file mode 100644 index 0000000000..2170e277a6 --- /dev/null +++ b/packages/core/src/astro/routes/api/admin/media-usage/activation.ts @@ -0,0 +1,42 @@ +import type { APIRoute } from "astro"; + +import { requirePerm } from "#api/authorize.js"; +import { requireDb, unwrapResult } from "#api/error.js"; +import { + handleMediaUsageActivationAdvance, + handleMediaUsageActivationStatus, +} from "#api/handlers/media-usage-activation.js"; +import { isParseError, parseBody } from "#api/parse.js"; +import { mediaUsageActivationAdvanceBody } from "#api/schemas.js"; +import { requireScope } from "#auth/scopes.js"; + +export const prerender = false; + +export const GET: APIRoute = async ({ locals }) => { + const { emdash, user } = locals; + const dbErr = requireDb(emdash?.db); + if (dbErr) return dbErr; + + const denied = requirePerm(user, "schema:manage"); + if (denied) return denied; + const scopeDenied = requireScope(locals, "admin"); + if (scopeDenied) return scopeDenied; + + return unwrapResult(await handleMediaUsageActivationStatus(emdash.db)); +}; + +export const POST: APIRoute = async ({ request, locals }) => { + const { emdash, user } = locals; + const dbErr = requireDb(emdash?.db); + if (dbErr) return dbErr; + + const denied = requirePerm(user, "schema:manage"); + if (denied) return denied; + const scopeDenied = requireScope(locals, "admin"); + if (scopeDenied) return scopeDenied; + + const body = await parseBody(request, mediaUsageActivationAdvanceBody); + if (isParseError(body)) return body; + + return unwrapResult(await handleMediaUsageActivationAdvance(emdash.db, body)); +}; diff --git a/packages/core/src/client/index.ts b/packages/core/src/client/index.ts index 0221037051..7a10ea3407 100644 --- a/packages/core/src/client/index.ts +++ b/packages/core/src/client/index.ts @@ -272,6 +272,29 @@ export interface MediaUsageWorkRetryResponse { item: MediaUsageWorkItem; } +export interface MediaUsageActivationStatus { + state: "expanded" | "activating" | "active"; + collectionCursor: string | null; + attemptCount: number; + drainConfirmedAt: string | null; + lastAttemptedAt: string | null; + lastErrorCode: "MEDIA_USAGE_ACTIVATION_FAILED" | null; + leaseExpiresAt: string | null; + activatedAt: string | null; + updatedAt: string; +} + +export interface MediaUsageActivationAdvanceInput { + writersDrained: true; + maintenanceReady: true; +} + +export interface MediaUsageActivationAdvanceResponse { + outcome: "activating" | "active"; + processedCollections: number; + activation: MediaUsageActivationStatus; +} + export type MediaUsageCollectionDeletionState = "pending" | "retry" | "leased" | "failed"; export type MediaUsageCollectionDeletionPhase = | "fence" @@ -894,6 +917,22 @@ export class EmDashClient { return this.request("POST", "/admin/media-usage/repair", input); } + /** Read the redacted controlled-activation status */ + async mediaGetUsageActivation(): Promise { + return this.request("GET", "/admin/media-usage/activation"); + } + + /** Advance exactly one controlled-activation batch */ + async mediaAdvanceUsageActivation( + input: MediaUsageActivationAdvanceInput, + ): Promise { + return this.request( + "POST", + "/admin/media-usage/activation", + input, + ); + } + /** List a bounded page of durable media usage entry work */ async mediaListUsageWork( options: MediaUsageWorkListOptions, diff --git a/packages/core/src/media/usage/activation.ts b/packages/core/src/media/usage/activation.ts index 749263c44f..993cea8285 100644 --- a/packages/core/src/media/usage/activation.ts +++ b/packages/core/src/media/usage/activation.ts @@ -12,6 +12,20 @@ const ACTIVATION_KEY = "incremental_capture"; const ACTIVATION_ERROR_CODE = "MEDIA_USAGE_ACTIVATION_FAILED"; export const MEDIA_USAGE_ACTIVATION_RUNTIME_GENERATION = 1; +export class MediaUsageActivationVersionMismatchError extends Error {} + +export interface MediaUsageActivationStatus { + state: "expanded" | "activating" | "active"; + collectionCursor: string | null; + attemptCount: number; + drainConfirmedAt: string | null; + lastAttemptedAt: string | null; + lastErrorCode: "MEDIA_USAGE_ACTIVATION_FAILED" | null; + leaseExpiresAt: string | null; + activatedAt: string | null; + updatedAt: string; +} + export const MEDIA_USAGE_ACTIVATION_LIMITS = Object.freeze({ collectionsPerCall: 1, leaseDurationSeconds: 5 * 60, @@ -34,6 +48,50 @@ export interface MediaUsageCollectionCapturePreparation { resuming: boolean; } +export async function getMediaUsageActivationStatus( + db: Kysely, +): Promise { + const activation = await db + .selectFrom("_emdash_media_usage_activation") + .select([ + "state", + "runtime_generation", + "collection_cursor", + "drain_confirmed_at", + "lease_expires_at", + "attempt_count", + "last_attempted_at", + "last_error_code", + "activated_at", + "updated_at", + ]) + .where("task_key", "=", ACTIVATION_KEY) + .executeTakeFirstOrThrow(); + assertRuntimeGeneration(activation); + if ( + activation.state !== "expanded" && + activation.state !== "activating" && + activation.state !== "active" + ) { + throw new Error("Invalid media usage activation state"); + } + if (!Number.isInteger(activation.attempt_count) || activation.attempt_count < 0) { + throw new Error("Invalid media usage activation attempt count"); + } + + return { + state: activation.state, + collectionCursor: activation.collection_cursor, + attemptCount: activation.attempt_count, + drainConfirmedAt: activation.drain_confirmed_at, + lastAttemptedAt: activation.last_attempted_at, + lastErrorCode: activation.last_error_code === null ? null : ACTIVATION_ERROR_CODE, + leaseExpiresAt: activation.lease_expires_at, + activatedAt: activation.activated_at, + updatedAt: activation.updated_at, + }; +} + export async function canResumeMediaUsageCollectionCapture( db: Kysely, identity: { collectionId: string; collectionSlug: string; creationFingerprint?: string }, @@ -308,9 +366,11 @@ async function findActivationIfAvailable( return findActivation(db); } -function assertRuntimeGeneration(activation: Selectable): void { +function assertRuntimeGeneration(activation: { runtime_generation: number }): void { if (activation.runtime_generation !== MEDIA_USAGE_ACTIVATION_RUNTIME_GENERATION) { - throw new Error("Media usage activation runtime generation mismatch"); + throw new MediaUsageActivationVersionMismatchError( + "Media usage activation runtime generation mismatch", + ); } } diff --git a/packages/core/tests/unit/api/media-usage-activation-route.test.ts b/packages/core/tests/unit/api/media-usage-activation-route.test.ts new file mode 100644 index 0000000000..2b0bf03bd2 --- /dev/null +++ b/packages/core/tests/unit/api/media-usage-activation-route.test.ts @@ -0,0 +1,334 @@ +import { Role, type RoleLevel } from "@emdash-cms/auth"; +import { sql } from "kysely"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; + +import { injectCoreRoutes } from "../../../src/astro/integration/routes.js"; +import { GET, POST } from "../../../src/astro/routes/api/admin/media-usage/activation.js"; +import { + setupForDialectWithCollections, + teardownForDialect, + type DialectTestContext, +} from "../../utils/test-db.js"; + +type GetContext = Parameters[0]; + +describe("admin media usage activation status route", () => { + let ctx: DialectTestContext | undefined; + + beforeEach(async () => { + ctx = await setupForDialectWithCollections("sqlite"); + }); + + afterEach(async () => { + await teardownForDialect(ctx); + ctx = undefined; + }); + + it("registers the activation route", () => { + const routes: Array<{ pattern: string; entrypoint: string }> = []; + injectCoreRoutes((route) => routes.push(route)); + + expect(routes).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + pattern: "/_emdash/api/admin/media-usage/activation", + entrypoint: expect.stringContaining("api/admin/media-usage/activation"), + }), + ]), + ); + }); + + it("requires authentication, schema permission, and admin token scope", async () => { + const request = activationRequest(); + + await expectError(await GET(routeContext(request, null)), 401, "UNAUTHORIZED"); + await expectError(await GET(routeContext(request, Role.EDITOR)), 403, "FORBIDDEN"); + await expectError( + await GET(routeContext(request, Role.ADMIN, ["content:read"])), + 403, + "INSUFFICIENT_SCOPE", + ); + + const post = () => activationPost({ writersDrained: true, maintenanceReady: true }); + await expectError(await POST(routeContext(post(), null)), 401, "UNAUTHORIZED"); + await expectError(await POST(routeContext(post(), Role.EDITOR)), 403, "FORBIDDEN"); + await expectError( + await POST(routeContext(post(), Role.ADMIN, ["content:read"])), + 403, + "INSUFFICIENT_SCOPE", + ); + }); + + it("returns a read-only redacted activation status", async () => { + await ctx!.db + .updateTable("_emdash_media_usage_activation") + .set({ + state: "activating", + collection_cursor: "posts", + drain_confirmed_at: "2026-08-12T09:00:00.000Z", + lease_token: "private-lease-token", + lease_expires_at: "2026-08-12T09:05:00.000Z", + attempt_count: 3, + last_attempted_at: "2026-08-12T09:00:00.000Z", + last_error_code: "private-database-error", + activated_at: null, + updated_at: "2026-08-12T09:00:01.000Z", + }) + .where("task_key", "=", "incremental_capture") + .execute(); + const before = await activationRow(); + + const response = await GET(routeContext(activationRequest(), Role.ADMIN, ["admin"])); + + expect(response.status).toBe(200); + expect(response.headers.get("Cache-Control")).toBe("private, no-store"); + const body = await response.json(); + expect(body).toEqual({ + success: true, + data: { + state: "activating", + collectionCursor: "posts", + attemptCount: 3, + drainConfirmedAt: "2026-08-12T09:00:00.000Z", + lastAttemptedAt: "2026-08-12T09:00:00.000Z", + lastErrorCode: "MEDIA_USAGE_ACTIVATION_FAILED", + leaseExpiresAt: "2026-08-12T09:05:00.000Z", + activatedAt: null, + updatedAt: "2026-08-12T09:00:01.000Z", + }, + }); + expect(await activationRow()).toEqual(before); + const serialized = JSON.stringify(body); + expect(serialized).not.toContain("private-lease-token"); + expect(serialized).not.toContain("private-database-error"); + expect(serialized).not.toContain("runtime_generation"); + }); + + it("fails closed for an incompatible runtime generation", async () => { + await ctx!.db + .updateTable("_emdash_media_usage_activation") + .set({ runtime_generation: 2 }) + .where("task_key", "=", "incremental_capture") + .execute(); + + const before = await activationRow(); + await expectError( + await GET(routeContext(activationRequest(), Role.ADMIN)), + 409, + "MEDIA_USAGE_ACTIVATION_VERSION_MISMATCH", + ); + await expectError( + await POST( + routeContext(activationPost({ writersDrained: true, maintenanceReady: true }), Role.ADMIN), + ), + 409, + "MEDIA_USAGE_ACTIVATION_VERSION_MISMATCH", + ); + expect(await activationRow()).toEqual(before); + }); + + it("fails closed for missing or invalid lifecycle metadata", async () => { + await ctx!.db + .updateTable("_emdash_media_usage_activation") + .set({ state: "unexpected" }) + .where("task_key", "=", "incremental_capture") + .execute(); + await expectError( + await GET(routeContext(activationRequest(), Role.ADMIN)), + 500, + "MEDIA_USAGE_ACTIVATION_READ_ERROR", + ); + await ctx!.db + .updateTable("_emdash_media_usage_activation") + .set({ state: "expanded", attempt_count: -1 }) + .where("task_key", "=", "incremental_capture") + .execute(); + await expectError( + await GET(routeContext(activationRequest(), Role.ADMIN)), + 500, + "MEDIA_USAGE_ACTIVATION_READ_ERROR", + ); + + await ctx!.db.deleteFrom("_emdash_media_usage_activation").execute(); + await expectError( + await GET(routeContext(activationRequest(), Role.ADMIN)), + 500, + "MEDIA_USAGE_ACTIVATION_READ_ERROR", + ); + }); + + it("requires both literal confirmations without mutating activation state", async () => { + const before = await activationRow(); + for (const body of [ + {}, + { writersDrained: false, maintenanceReady: true }, + { writersDrained: true, maintenanceReady: false }, + { writersDrained: true, maintenanceReady: true, extra: true }, + ]) { + await expectError( + await POST(routeContext(activationPost(body), Role.ADMIN)), + 400, + "VALIDATION_ERROR", + ); + expect(await activationRow()).toEqual(before); + } + }); + + it("advances exactly one collection per confirmed request and is idempotent when active", async () => { + const first = await POST( + routeContext(activationPost({ writersDrained: true, maintenanceReady: true }), Role.ADMIN, [ + "admin", + ]), + ); + expect(first.status).toBe(200); + expect(await first.json()).toEqual({ + success: true, + data: { + outcome: "activating", + processedCollections: 1, + activation: expect.objectContaining({ state: "activating" }), + }, + }); + + const second = await POST( + routeContext(activationPost({ writersDrained: true, maintenanceReady: true }), Role.ADMIN), + ); + expect(second.status).toBe(200); + expect(await second.json()).toEqual({ + success: true, + data: { + outcome: "active", + processedCollections: 1, + activation: expect.objectContaining({ state: "active" }), + }, + }); + + const third = await POST( + routeContext(activationPost({ writersDrained: true, maintenanceReady: true }), Role.ADMIN), + ); + expect(await third.json()).toEqual({ + success: true, + data: { + outcome: "active", + processedCollections: 0, + activation: expect.objectContaining({ state: "active" }), + }, + }); + }); + + it("returns a stable redacted conflict for a live activation lease", async () => { + await ctx!.db + .updateTable("_emdash_media_usage_activation") + .set({ + state: "activating", + lease_token: "private-owner-token", + lease_expires_at: "2100-01-01T00:00:00.000Z", + }) + .where("task_key", "=", "incremental_capture") + .execute(); + + const response = await POST( + routeContext(activationPost({ writersDrained: true, maintenanceReady: true }), Role.ADMIN), + ); + expect(response.status).toBe(409); + const body = await response.json(); + expect(body).toEqual({ + success: false, + error: { + code: "MEDIA_USAGE_ACTIVATION_BUSY", + message: expect.any(String), + details: { leaseExpiresAt: "2100-01-01T00:00:00.000Z" }, + }, + }); + expect(JSON.stringify(body)).not.toContain("private-owner-token"); + }); + + it("maps ownership loss to a stable conflict", async () => { + await sql` + CREATE TRIGGER steal_activation_lease_from_route + AFTER UPDATE OF capture_state ON _emdash_media_usage_index_status + WHEN NEW.capture_state = 'active' + BEGIN + UPDATE _emdash_media_usage_activation + SET lease_token = 'new-owner', + lease_expires_at = '2100-01-01T00:00:00.000Z' + WHERE task_key = 'incremental_capture'; + END + `.execute(ctx!.db); + + await expectError( + await POST( + routeContext(activationPost({ writersDrained: true, maintenanceReady: true }), Role.ADMIN), + ), + 409, + "MEDIA_USAGE_ACTIVATION_CONFLICT", + ); + }); + + it("records trigger failures while returning only a stable public error", async () => { + await sql`DROP TABLE ${sql.ref("ec_page")}`.execute(ctx!.db); + + const response = await POST( + routeContext(activationPost({ writersDrained: true, maintenanceReady: true }), Role.ADMIN), + ); + expect(response.status).toBe(500); + const body = await response.json(); + expect(body).toEqual({ + success: false, + error: { + code: "MEDIA_USAGE_ACTIVATION_ADVANCE_ERROR", + message: expect.any(String), + }, + }); + expect(JSON.stringify(body)).not.toContain("ec_page"); + expect(await activationRow()).toEqual( + expect.objectContaining({ + state: "activating", + lease_token: null, + last_error_code: "MEDIA_USAGE_ACTIVATION_FAILED", + }), + ); + }); + + function activationRow() { + return ctx!.db + .selectFrom("_emdash_media_usage_activation") + .selectAll() + .where("task_key", "=", "incremental_capture") + .executeTakeFirst(); + } + + function routeContext( + request: Request, + role: RoleLevel | null, + tokenScopes?: string[], + ): GetContext { + return { + request, + locals: { + emdash: { db: ctx!.db }, + user: role == null ? null : { id: "user-1", role }, + tokenScopes, + }, + } as GetContext; + } +}); + +async function expectError(response: Response, status: number, code: string): Promise { + expect(response.status).toBe(status); + const body = (await response.json()) as { error: { code: string } }; + expect(body.error.code).toBe(code); + expect(response.headers.get("Cache-Control")).toBe("private, no-store"); +} + +function activationRequest(): Request { + return new Request("http://localhost/_emdash/api/admin/media-usage/activation"); +} + +function activationPost(body: unknown): Request { + return new Request("http://localhost/_emdash/api/admin/media-usage/activation", { + method: "POST", + headers: { "Content-Type": "application/json", "X-EmDash-Request": "1" }, + body: JSON.stringify(body), + }); +} diff --git a/packages/core/tests/unit/api/openapi.test.ts b/packages/core/tests/unit/api/openapi.test.ts index 934af02595..03db925cf9 100644 --- a/packages/core/tests/unit/api/openapi.test.ts +++ b/packages/core/tests/unit/api/openapi.test.ts @@ -121,6 +121,44 @@ describe("OpenAPI document generation", () => { ); }); + it("documents the media usage activation status operation", () => { + const doc = generateOpenApiDocument(); + const path = doc.paths?.["/_emdash/api/admin/media-usage/activation"]; + const get = path?.get as { + operationId?: string; + responses?: Record; + }; + const post = path?.post as { + operationId?: string; + requestBody?: unknown; + responses?: Record; + }; + + expect(get.operationId).toBe("getMediaUsageActivation"); + expect(get.responses).toEqual( + expect.objectContaining({ + "200": expect.any(Object), + "401": expect.any(Object), + "403": expect.any(Object), + "409": expect.any(Object), + "500": expect.any(Object), + }), + ); + expect(JSON.stringify(get.responses?.["200"])).toContain("MediaUsageActivationStatus"); + expect(post.operationId).toBe("advanceMediaUsageActivation"); + expect(post.requestBody).toBeDefined(); + expect(post.responses).toEqual( + expect.objectContaining({ + "200": expect.any(Object), + "400": expect.any(Object), + "401": expect.any(Object), + "403": expect.any(Object), + "409": expect.any(Object), + "500": expect.any(Object), + }), + ); + }); + it("includes schema paths", () => { const doc = generateOpenApiDocument(); const paths = Object.keys(doc.paths ?? {}); diff --git a/packages/core/tests/unit/client/client.test.ts b/packages/core/tests/unit/client/client.test.ts index 8d85286df3..fc36e30bcf 100644 --- a/packages/core/tests/unit/client/client.test.ts +++ b/packages/core/tests/unit/client/client.test.ts @@ -1021,6 +1021,73 @@ describe("EmDashClient", () => { }); describe("media usage work operators", () => { + it("reads the redacted activation status", async () => { + let capturedRequest: Request | undefined; + const status = { + state: "expanded" as const, + collectionCursor: null, + attemptCount: 0, + drainConfirmedAt: null, + lastAttemptedAt: null, + lastErrorCode: null, + leaseExpiresAt: null, + activatedAt: null, + updatedAt: "2026-08-12T09:00:00.000Z", + }; + const client = new EmDashClient({ + baseUrl: "http://localhost:4321", + token: "test", + interceptors: [ + async (request) => { + capturedRequest = request; + return jsonResponse(status); + }, + ], + }); + + await expect(client.mediaGetUsageActivation()).resolves.toEqual(status); + expect(capturedRequest?.method).toBe("GET"); + expect(new URL(capturedRequest!.url).pathname).toBe( + "/_emdash/api/admin/media-usage/activation", + ); + }); + + it("advances one activation batch with both explicit confirmations", async () => { + let capturedBody: unknown; + const client = new EmDashClient({ + baseUrl: "http://localhost:4321", + token: "test", + interceptors: [ + async (request) => { + expect(request.method).toBe("POST"); + expect(request.headers.get("X-EmDash-Request")).toBe("1"); + capturedBody = await request.json(); + return jsonResponse({ + outcome: "active", + processedCollections: 0, + activation: { + state: "active", + collectionCursor: null, + attemptCount: 1, + drainConfirmedAt: "2026-08-12T09:00:00.000Z", + lastAttemptedAt: "2026-08-12T09:00:00.000Z", + lastErrorCode: null, + leaseExpiresAt: null, + activatedAt: "2026-08-12T09:00:01.000Z", + updatedAt: "2026-08-12T09:00:01.000Z", + }, + }); + }, + ], + }); + + await client.mediaAdvanceUsageActivation({ + writersDrained: true, + maintenanceReady: true, + }); + expect(capturedBody).toEqual({ writersDrained: true, maintenanceReady: true }); + }); + it("serializes a bounded work-list query and returns the cursor page", async () => { let capturedUrl: URL | undefined; const page = { diff --git a/packages/core/tests/workerd/media-usage-activation-d1.test.ts b/packages/core/tests/workerd/media-usage-activation-d1.test.ts new file mode 100644 index 0000000000..2fac11f356 --- /dev/null +++ b/packages/core/tests/workerd/media-usage-activation-d1.test.ts @@ -0,0 +1,220 @@ +import { Role } from "@emdash-cms/auth"; +import { env } from "cloudflare:test"; +import { Kysely, sql } from "kysely"; +import { afterAll, beforeAll, expect, it } from "vitest"; + +import { EmDashD1Dialect, RawBindingD1Dialect } from "../../../cloudflare/src/db/d1-dialect.js"; +import { GET, POST } from "../../src/astro/routes/api/admin/media-usage/activation.js"; +import { runMigrations } from "../../src/database/migrations/runner.js"; +import type { Database } from "../../src/database/types.js"; +import { MEDIA_USAGE_ACTIVATION_RUNTIME_GENERATION } from "../../src/media/usage/activation.js"; +import { SchemaRegistry } from "../../src/schema/registry.js"; + +declare module "cloudflare:test" { + interface ProvidedEnv { + DB: D1Database; + } +} + +interface D1Measurement { + queries: number; + rowsRead: number; + rowsWritten: number; + durationMs: number; + wallDurationMs: number; + maxBinds: number; + maxSqlBytes: number; +} + +type D1Mode = "raw" | "session"; +type RouteContext = Parameters[0]; + +let adminDb: Kysely; + +beforeAll(async () => { + adminDb = new Kysely({ + dialect: new RawBindingD1Dialect({ database: env.DB }), + }); + await runMigrations(adminDb); +}); + +afterAll(async () => { + await adminDb.destroy(); +}); + +it("keeps complete authenticated activation route costs within the D1 envelope", async () => { + const evidence: Array<{ path: string; mode: D1Mode } & D1Measurement> = []; + + await record("get-expanded", "raw", GET, activationGet(), 200, evidence); + await record("empty-activation", "session", POST, activationPost(), 200, evidence); + await record("get-active-empty", "session", GET, activationGet(), 200, evidence); + + await resetActivation("expanded", null); + const registry = new SchemaRegistry(adminDb); + await registry.createCollection({ slug: "activation_alpha", label: "Activation alpha" }); + await registry.createCollection({ slug: "activation_beta", label: "Activation beta" }); + + await record("first-collection", "raw", POST, activationPost(), 200, evidence); + await record("get-activating", "session", GET, activationGet(), 200, evidence); + await record("final-collection", "session", POST, activationPost(), 200, evidence); + await record("get-active", "raw", GET, activationGet(), 200, evidence); + await record("active-idempotent", "session", POST, activationPost(), 200, evidence); + + await adminDb + .updateTable("_emdash_media_usage_activation") + .set({ + state: "activating", + collection_cursor: "activation_alpha", + lease_token: "private-owner", + lease_expires_at: "2100-01-01T00:00:00.000Z", + }) + .where("task_key", "=", "incremental_capture") + .execute(); + await record("live-lease", "raw", POST, activationPost(), 409, evidence); + await adminDb + .updateTable("_emdash_media_usage_activation") + .set({ lease_expires_at: "2000-01-01T00:00:00.000Z" }) + .where("task_key", "=", "incremental_capture") + .execute(); + await record("expired-lease-replay", "session", POST, activationPost(), 200, evidence); + + await resetActivation("expanded", "activation_beta"); + await registry.createCollection({ slug: "activation_zz_broken", label: "Activation broken" }); + await sql`DROP TABLE ${sql.ref("ec_activation_zz_broken")}`.execute(adminDb); + await record("installation-failure", "raw", POST, activationPost(), 500, evidence); + await record("failure-retry", "session", POST, activationPost(), 500, evidence); + + console.info(`PR4_D1_ACTIVATION=${JSON.stringify(evidence)}`); +}); + +async function record( + path: string, + mode: D1Mode, + handler: typeof GET, + request: Request, + expectedStatus: number, + evidence: Array<{ path: string; mode: D1Mode } & D1Measurement>, +): Promise { + const measurement = emptyMeasurement(); + const db = measuredDb(mode, measurement); + const startedAt = performance.now(); + const response = await handler(routeContext(db, request)); + measurement.wallDurationMs = Number((performance.now() - startedAt).toFixed(3)); + await db.destroy(); + + expect(response.status).toBe(expectedStatus); + expect(measurement.queries).toBeLessThanOrEqual(40); + expect(measurement.maxBinds).toBeLessThanOrEqual(100); + expect(measurement.maxSqlBytes).toBeLessThan(100 * 1024); + expect(measurement.wallDurationMs).toBeLessThan(2500); + evidence.push({ path, mode, ...measurement }); +} + +function measuredDb(mode: D1Mode, measurement: D1Measurement): Kysely { + const binding = + mode === "raw" ? env.DB : (env.DB.withSession("first-primary") as unknown as D1Database); + const database = captureD1(binding, measurement); + return new Kysely({ + dialect: + mode === "raw" ? new RawBindingD1Dialect({ database }) : new EmDashD1Dialect({ database }), + }); +} + +function routeContext(db: Kysely, request: Request): RouteContext { + return { + request, + locals: { + emdash: { db }, + user: { id: "admin-1", role: Role.ADMIN }, + tokenScopes: ["admin"], + }, + } as RouteContext; +} + +function activationGet(): Request { + return new Request("http://localhost/_emdash/api/admin/media-usage/activation"); +} + +function activationPost(): Request { + return new Request("http://localhost/_emdash/api/admin/media-usage/activation", { + method: "POST", + headers: { "Content-Type": "application/json", "X-EmDash-Request": "1" }, + body: JSON.stringify({ writersDrained: true, maintenanceReady: true }), + }); +} + +async function resetActivation(state: "expanded" | "activating", cursor: string | null) { + await adminDb + .updateTable("_emdash_media_usage_activation") + .set({ + state, + runtime_generation: MEDIA_USAGE_ACTIVATION_RUNTIME_GENERATION, + collection_cursor: cursor, + drain_confirmed_at: null, + lease_token: null, + lease_expires_at: null, + attempt_count: 0, + last_attempted_at: null, + last_error_code: null, + activated_at: null, + }) + .where("task_key", "=", "incremental_capture") + .execute(); +} + +function emptyMeasurement(): D1Measurement { + return { + queries: 0, + rowsRead: 0, + rowsWritten: 0, + durationMs: 0, + wallDurationMs: 0, + maxBinds: 0, + maxSqlBytes: 0, + }; +} + +function captureD1(database: D1Database, measurement: D1Measurement): D1Database { + return new Proxy(database, { + get(target, property) { + if (property === "prepare") { + return (query: string) => captureStatement(target.prepare(query), query, [], measurement); + } + const value: unknown = Reflect.get(target, property, target); + return typeof value === "function" ? value.bind(target) : value; + }, + }); +} + +function captureStatement( + statement: D1PreparedStatement, + query: string, + binds: unknown[], + measurement: D1Measurement, +): D1PreparedStatement { + return new Proxy(statement, { + get(target, property) { + if (property === "bind") { + return (...values: unknown[]) => + captureStatement(target.bind(...values), query, values, measurement); + } + if (property === "all") { + return async () => { + measurement.queries++; + measurement.maxBinds = Math.max(measurement.maxBinds, binds.length); + measurement.maxSqlBytes = Math.max( + measurement.maxSqlBytes, + new TextEncoder().encode(query).byteLength, + ); + const result = await target.all(); + measurement.rowsRead += result.meta.rows_read; + measurement.rowsWritten += result.meta.rows_written; + measurement.durationMs += result.meta.duration; + return result; + }; + } + const value: unknown = Reflect.get(target, property, target); + return typeof value === "function" ? value.bind(target) : value; + }, + }); +}