diff --git a/.changeset/young-moose-act.md b/.changeset/young-moose-act.md new file mode 100644 index 000000000..a845151cc --- /dev/null +++ b/.changeset/young-moose-act.md @@ -0,0 +1,2 @@ +--- +--- diff --git a/packages/core/src/database/migrations/052_media_usage_read_index.ts b/packages/core/src/database/migrations/052_media_usage_read_index.ts new file mode 100644 index 000000000..bc7c0e3e6 --- /dev/null +++ b/packages/core/src/database/migrations/052_media_usage_read_index.ts @@ -0,0 +1,26 @@ +import type { Kysely } from "kysely"; + +const OLD_INDEX = "idx__emdash_media_usage_media_id"; +const READ_INDEX = "idx__emdash_media_usage_media_source_generation"; + +export async function up(db: Kysely): Promise { + // D1 DDL is non-transactional: create the replacement before dropping the + // old index so an interrupted migration always leaves a media-leading index. + await db.schema + .createIndex(READ_INDEX) + .ifNotExists() + .on("_emdash_media_usage") + .columns(["media_id", "source_key", "generation"]) + .execute(); + await db.schema.dropIndex(OLD_INDEX).ifExists().execute(); +} + +export async function down(db: Kysely): Promise { + await db.schema + .createIndex(OLD_INDEX) + .ifNotExists() + .on("_emdash_media_usage") + .column("media_id") + .execute(); + await db.schema.dropIndex(READ_INDEX).ifExists().execute(); +} diff --git a/packages/core/src/database/migrations/runner.ts b/packages/core/src/database/migrations/runner.ts index ca2020144..6df97e686 100644 --- a/packages/core/src/database/migrations/runner.ts +++ b/packages/core/src/database/migrations/runner.ts @@ -54,6 +54,7 @@ import * as m048 from "./048_restore_content_taxonomies_term_index.js"; import * as m049 from "./049_taxonomies_name_locale_index.js"; import * as m050 from "./050_media_usage_index_status.js"; import * as m051 from "./051_content_taxonomies_denorm.js"; +import * as m052 from "./052_media_usage_read_index.js"; const MIGRATIONS: Readonly> = Object.freeze({ "001_initial": m001, @@ -106,6 +107,7 @@ const MIGRATIONS: Readonly> = Object.freeze({ "049_taxonomies_name_locale_index": m049, "050_media_usage_index_status": m050, "051_content_taxonomies_denorm": m051, + "052_media_usage_read_index": m052, }); /** Total number of registered migrations. Exported for use in tests. */ diff --git a/packages/core/src/database/repositories/media-usage.ts b/packages/core/src/database/repositories/media-usage.ts index 7a8935034..11c1efac5 100644 --- a/packages/core/src/database/repositories/media-usage.ts +++ b/packages/core/src/database/repositories/media-usage.ts @@ -36,6 +36,23 @@ const OCCURRENCE_INSERT_BATCH_SIZE = Math.max( 1, Math.floor(SQL_BATCH_SIZE / OCCURRENCE_BIND_COLUMNS), ); +const CONTENT_SOURCE_ELIGIBILITY = sql`( + s.source_variant = 'draft_overlay' + OR ( + s.source_variant = 'columns' + AND ( + s.content_status = 'published' + OR NOT EXISTS ( + SELECT 1 + FROM _emdash_media_usage_sources AS overlay + WHERE overlay.source_type = 'content' + AND overlay.collection_slug = s.collection_slug + AND overlay.content_id = s.content_id + AND overlay.source_variant = 'draft_overlay' + ) + ) + ) +)`; export interface MediaUsageSourceInput { sourceKey: string; @@ -193,6 +210,24 @@ export interface FindMediaUsageOptions { cursor?: string; } +export interface MediaUsageCollectionIndexStatusScope { + collectionSlug: string; + status: string | null; + schemaVersion: number | null; +} + +export interface MediaUsageEntrySource { + source: MediaUsageSource; + occurrences: MediaUsageOccurrence[]; +} + +export interface MediaUsageEntryGroup { + collectionSlug: string; + contentId: string; + contentDeletedAt: string | null; + sources: MediaUsageEntrySource[]; +} + interface MediaUsageSourceRow { source_key: string; source_type: string; @@ -280,6 +315,11 @@ interface JoinedUsageRow { occurrence_created_at: string; } +interface GroupedUsageRow extends JoinedUsageRow { + entry_deleted_at: string | null; + has_more: number; +} + /** Persistence-only repository for the internal media usage projection tables. */ export class MediaUsageRepository { constructor(private db: Kysely) {} @@ -420,6 +460,166 @@ export class MediaUsageRepository { }; } + async findActiveEntryCountsByMediaIds(mediaIds: readonly string[]): Promise> { + const uniqueMediaIds = [...new Set(mediaIds)]; + const counts = new Map(uniqueMediaIds.map((mediaId) => [mediaId, 0])); + + for (const mediaIdBatch of chunks(uniqueMediaIds, SQL_BATCH_SIZE)) { + const visibleEntries = this.currentContentMediaUsageBaseQuery() + .select([ + "u.media_id as media_id", + "s.collection_slug as collection_slug", + "s.content_id as content_id", + ]) + .where("u.media_id", "in", mediaIdBatch) + .where((eb) => + eb.not( + eb.exists( + eb + .selectFrom("_emdash_media_usage_sources as deleted_source") + .select("deleted_source.source_key") + .where("deleted_source.source_type", "=", "content") + .whereRef("deleted_source.collection_slug", "=", "s.collection_slug") + .whereRef("deleted_source.content_id", "=", "s.content_id") + .where("deleted_source.source_variant", "in", ["columns", "draft_overlay"]) + .where("deleted_source.content_deleted_at", "is not", null), + ), + ), + ) + .distinct() + .as("visible_entries"); + + const rows = await this.db + .selectFrom(visibleEntries) + .select("media_id") + .select((eb) => eb.fn.countAll().as("usage_count")) + .groupBy("media_id") + .execute(); + + for (const row of rows) { + if (row.media_id !== null) counts.set(row.media_id, Number(row.usage_count)); + } + } + + return counts; + } + + async findCollectionIndexStatusScopes( + identity: Pick, + ): Promise { + const rows = await this.db + .selectFrom("_emdash_collections as collection") + .leftJoin("_emdash_media_usage_index_status as status", (join) => + join + .on("status.adapter_id", "=", identity.adapterId) + .on("status.scope_type", "=", identity.scopeType) + .onRef("status.scope_key", "=", "collection.slug"), + ) + .select([ + "collection.slug as collection_slug", + "status.status as status", + "status.schema_version as schema_version", + ]) + .orderBy("collection.slug", "asc") + .execute(); + + return rows.map((row) => ({ + collectionSlug: row.collection_slug, + status: row.status, + schemaVersion: row.schema_version === null ? null : Number(row.schema_version), + })); + } + + async findCurrentEntryUsagePageByMediaId( + mediaId: string, + options: FindMediaUsageOptions = {}, + ): Promise> { + const requestedLimit = Math.floor(options.limit ?? 50); + const limit = Number.isFinite(requestedLimit) ? Math.min(Math.max(1, requestedLimit), 100) : 50; + const cursor = options.cursor ? decodeCursor(options.cursor) : null; + let matchedGroups = this.currentContentMediaUsageBaseQuery() + .select(["s.collection_slug as collection_slug", "s.content_id as content_id"]) + .where("u.media_id", "=", mediaId) + .distinct(); + if (cursor) { + matchedGroups = matchedGroups.where((eb) => + eb.or([ + eb("s.collection_slug", ">", cursor.orderValue), + eb.and([ + eb("s.collection_slug", "=", cursor.orderValue), + eb("s.content_id", ">", cursor.id), + ]), + ]), + ); + } + matchedGroups = matchedGroups + .orderBy("s.collection_slug", "asc") + .orderBy("s.content_id", "asc") + .limit(limit + 1); + + const rows: GroupedUsageRow[] = await this.db + .with("matched_groups", () => matchedGroups) + .with("page_groups", (db) => + db + .selectFrom("matched_groups") + .selectAll() + .orderBy("collection_slug", "asc") + .orderBy("content_id", "asc") + .limit(limit), + ) + .with("entry_state", (db) => + db + .selectFrom("page_groups as page") + .crossJoin("_emdash_media_usage_sources as state") + .select(["page.collection_slug", "page.content_id"]) + .select((eb) => + eb.fn.max("state.content_deleted_at").as("entry_deleted_at"), + ) + .whereRef("page.collection_slug", "=", "state.collection_slug") + .whereRef("page.content_id", "=", "state.content_id") + .where("state.source_type", "=", "content") + .where("state.source_variant", "in", ["columns", "draft_overlay"]) + .groupBy(["page.collection_slug", "page.content_id"]), + ) + .selectFrom("entry_state as page") + .crossJoin("_emdash_media_usage_sources as s") + .crossJoin("_emdash_media_usage as u") + .whereRef("page.collection_slug", "=", "s.collection_slug") + .whereRef("page.content_id", "=", "s.content_id") + .whereRef("s.source_key", "=", "u.source_key") + .whereRef("s.current_generation", "=", "u.generation") + .select(currentUsageSelect) + .select("page.entry_deleted_at") + .select( + sql`CASE + WHEN (SELECT COUNT(*) FROM matched_groups) > ${limit} THEN 1 + ELSE 0 + END`.as("has_more"), + ) + .where("u.media_id", "=", mediaId) + .where("s.source_type", "=", "content") + .where("s.collection_slug", "is not", null) + .where("s.content_id", "is not", null) + .where("s.source_variant", "in", ["columns", "draft_overlay"]) + .where(CONTENT_SOURCE_ELIGIBILITY) + .orderBy("s.collection_slug", "asc") + .orderBy("s.content_id", "asc") + .orderBy("s.source_variant", "asc") + .orderBy("s.source_key", "asc") + .orderBy("u.field_path", "asc") + .orderBy("u.occurrence_index", "asc") + .orderBy("u.id", "asc") + .execute(); + + const items = groupUsageRows(rows); + const result: FindManyResult = { items }; + if (Number(rows[0]?.has_more ?? 0) === 1 && items.length > 0) { + const last = items.at(-1)!; + result.nextCursor = encodeCursor(last.collectionSlug, last.contentId); + } + return result; + } + async findCurrentUsageByMediaId(mediaId: string): Promise { const rows = await this.db .selectFrom("_emdash_media_usage_sources as s") @@ -889,6 +1089,20 @@ export class MediaUsageRepository { .select(currentUsageSelect); } + private currentContentMediaUsageBaseQuery() { + return this.db + .selectFrom("_emdash_media_usage as u") + .crossJoin("_emdash_media_usage_sources as s") + .innerJoin("_emdash_collections as collection", "collection.slug", "s.collection_slug") + .whereRef("s.source_key", "=", "u.source_key") + .whereRef("s.current_generation", "=", "u.generation") + .where("s.source_type", "=", "content") + .where("s.collection_slug", "is not", null) + .where("s.content_id", "is not", null) + .where("s.source_variant", "in", ["columns", "draft_overlay"]) + .where(CONTENT_SOURCE_ELIGIBILITY); + } + private async deleteSourceKeys(sourceKeys: readonly string[]): Promise { const uniqueSourceKeys = [...new Set(sourceKeys)]; if (uniqueSourceKeys.length === 0) return 0; @@ -1229,6 +1443,38 @@ const currentUsageSelect = [ "u.created_at as occurrence_created_at", ] as const; +function groupUsageRows(rows: readonly GroupedUsageRow[]): MediaUsageEntryGroup[] { + const groups: MediaUsageEntryGroup[] = []; + + for (const row of rows) { + if (row.collection_slug === null || row.content_id === null) continue; + const record = rowToUsageRecord(row); + let group = groups.at(-1); + if ( + !group || + group.collectionSlug !== row.collection_slug || + group.contentId !== row.content_id + ) { + group = { + collectionSlug: row.collection_slug, + contentId: row.content_id, + contentDeletedAt: row.entry_deleted_at, + sources: [], + }; + groups.push(group); + } + + let source = group.sources.at(-1); + if (!source || source.source.sourceKey !== record.source.sourceKey) { + source = { source: record.source, occurrences: [] }; + group.sources.push(source); + } + source.occurrences.push(record.occurrence); + } + + return groups; +} + function rowToSource(row: MediaUsageSourceRow): MediaUsageSource { return { sourceKey: row.source_key, diff --git a/packages/core/tests/integration/database/media-usage-migration.test.ts b/packages/core/tests/integration/database/media-usage-migration.test.ts index 03f920002..9bada5dfb 100644 --- a/packages/core/tests/integration/database/media-usage-migration.test.ts +++ b/packages/core/tests/integration/database/media-usage-migration.test.ts @@ -14,7 +14,7 @@ const EXPECTED_INDEXES = [ "idx__emdash_media_usage_sources_locale", "idx__emdash_media_usage_sources_deleted", "idx__emdash_media_usage_sources_translation_group", - "idx__emdash_media_usage_media_id", + "idx__emdash_media_usage_media_source_generation", "idx__emdash_media_usage_provider_asset", "idx__emdash_media_usage_source_generation", "idx__emdash_media_usage_unique_occurrence", @@ -110,6 +110,47 @@ describeEachDialect("media usage index migration", (dialect) => { } }); + it("replaces the media lookup index in retry-safe D1 statement order", async () => { + const migration = + await import("../../../src/database/migrations/052_media_usage_read_index.js"); + + await migration.down(ctx.db); + let indexNames = await listIndexNames(ctx); + expect(indexNames.has("idx__emdash_media_usage_media_id")).toBe(true); + expect(indexNames.has("idx__emdash_media_usage_media_source_generation")).toBe(false); + + // Simulate interruption after the first up statement, before the old index drops. + await ctx.db.schema + .createIndex("idx__emdash_media_usage_media_source_generation") + .ifNotExists() + .on("_emdash_media_usage") + .columns(["media_id", "source_key", "generation"]) + .execute(); + await migration.up(ctx.db); + await migration.up(ctx.db); + + indexNames = await listIndexNames(ctx); + expect(indexNames.has("idx__emdash_media_usage_media_id")).toBe(false); + expect(indexNames.has("idx__emdash_media_usage_media_source_generation")).toBe(true); + + // Simulate interruption after the first down statement, before the new index drops. + await ctx.db.schema + .createIndex("idx__emdash_media_usage_media_id") + .ifNotExists() + .on("_emdash_media_usage") + .column("media_id") + .execute(); + indexNames = await listIndexNames(ctx); + expect(indexNames.has("idx__emdash_media_usage_media_id")).toBe(true); + expect(indexNames.has("idx__emdash_media_usage_media_source_generation")).toBe(true); + await migration.down(ctx.db); + await migration.down(ctx.db); + + indexNames = await listIndexNames(ctx); + expect(indexNames.has("idx__emdash_media_usage_media_id")).toBe(true); + expect(indexNames.has("idx__emdash_media_usage_media_source_generation")).toBe(false); + }); + it("down() drops tables and up() recreates them", async () => { const migration = await import("../../../src/database/migrations/046_media_usage_index.js"); diff --git a/packages/core/tests/integration/database/media-usage-read-plan.test.ts b/packages/core/tests/integration/database/media-usage-read-plan.test.ts new file mode 100644 index 000000000..36c813cc7 --- /dev/null +++ b/packages/core/tests/integration/database/media-usage-read-plan.test.ts @@ -0,0 +1,203 @@ +/** + * Query-plan and statement-count coverage for media usage reads. + * + * SQLite runs without ANALYZE/sqlite_stat1 here, matching D1's stats-blind + * planner. Dialect result parity is covered by media-usage-read-repository. + */ + +import Database from "better-sqlite3"; +import { Kysely, SqliteDialect } from "kysely"; +import { afterEach, beforeEach, expect, it } from "vitest"; + +import { runMigrations } from "../../../src/database/migrations/runner.js"; +import { MediaUsageRepository } from "../../../src/database/repositories/media-usage.js"; +import type { Database as DatabaseSchema } from "../../../src/database/types.js"; +import { buildContentMediaUsageSourceKey } from "../../../src/media/usage/source-key.js"; +import { SQL_BATCH_SIZE } from "../../../src/utils/chunks.js"; + +interface CapturedQuery { + sql: string; + parameters: readonly unknown[]; +} + +let sqlite: Database.Database; +let db: Kysely; +let repo: MediaUsageRepository; +let captured: CapturedQuery[]; + +beforeEach(async () => { + captured = []; + sqlite = new Database(":memory:"); + db = new Kysely({ + dialect: new SqliteDialect({ database: sqlite }), + log(event) { + if (event.level === "query") { + captured.push({ sql: event.query.sql, parameters: event.query.parameters }); + } + }, + }); + await runMigrations(db); + await db + .insertInto("_emdash_collections") + .values({ id: "collection-posts", slug: "posts", label: "Posts", has_seo: 0 }) + .execute(); + repo = new MediaUsageRepository(db); + await repo.replaceSource( + { + sourceKey: buildContentMediaUsageSourceKey({ + collectionSlug: "posts", + contentId: "entry-1", + sourceVariant: "columns", + }), + sourceType: "content", + collectionSlug: "posts", + contentId: "entry-1", + sourceVariant: "columns", + locale: "en", + contentStatus: "published", + }, + [ + { + fieldSlug: "hero", + fieldPath: "hero", + referenceType: "image_field", + mediaId: "media-shared", + provider: "local", + providerAssetId: "media-shared", + }, + ], + ); + captured = []; +}); + +afterEach(async () => { + await db.destroy(); +}); + +it("seeks batched counts through the media/source/generation index", async () => { + const mediaIds = [ + "media-shared", + ...Array.from({ length: SQL_BATCH_SIZE }, (_, index) => `media-${index}`), + ]; + + await repo.findActiveEntryCountsByMediaIds(mediaIds); + + const queries = captured.filter((query) => query.sql.includes("visible_entries")); + expect(queries).toHaveLength(2); + for (const query of queries) { + const plan = explain(query); + expect(query.parameters.length).toBeLessThanOrEqual(100); + expect(firstSourceOrUsageAccess(plan)).toMatch( + /SEARCH u USING (?:COVERING )?INDEX idx__emdash_media_usage_media_source_generation/, + ); + expect(plan).toContain("idx__emdash_media_usage_media_source_generation"); + expect(plan).toContain("idx__emdash_media_usage_sources_content"); + expect(plan).not.toContain("SCAN u"); + } +}); + +it("loads coverage and one grouped page in one statement each", async () => { + await repo.findCollectionIndexStatusScopes({ + adapterId: "content-media", + scopeType: "collection", + }); + await repo.findCurrentEntryUsagePageByMediaId("media-shared", { limit: 1 }); + + const coverageQueries = captured.filter( + (query) => + query.sql.includes("_emdash_media_usage_index_status") && query.sql.includes("left join"), + ); + const groupedQueries = captured.filter((query) => query.sql.includes("matched_groups")); + expect(coverageQueries).toHaveLength(1); + expect(groupedQueries).toHaveLength(1); + expect(groupedQueries[0]!.sql).toContain("entry_state"); + expect(groupedQueries[0]!.sql).not.toContain("deleted_source"); + + const plan = explain(groupedQueries[0]!); + expect(firstSourceOrUsageAccess(plan)).toMatch( + /SEARCH u USING (?:COVERING )?INDEX idx__emdash_media_usage_media_source_generation/, + ); + expectPageBoundedHydration(plan); + expect(plan).toContain("idx__emdash_media_usage_media_source_generation"); + expect(plan).toContain("idx__emdash_media_usage_sources_content"); + expect(plan).not.toContain("SCAN u"); +}); + +it("keeps a high-cardinality grouped read to one indexed statement", async () => { + for (let index = 2; index <= 200; index++) { + const contentId = `entry-${String(index).padStart(3, "0")}`; + await repo.replaceSource( + { + sourceKey: buildContentMediaUsageSourceKey({ + collectionSlug: "posts", + contentId, + sourceVariant: "columns", + }), + sourceType: "content", + collectionSlug: "posts", + contentId, + sourceVariant: "columns", + locale: "en", + contentStatus: "published", + }, + [ + { + fieldSlug: "hero", + fieldPath: "hero", + referenceType: "image_field", + mediaId: "media-shared", + provider: "local", + providerAssetId: "media-shared", + }, + ], + ); + } + captured = []; + + const page = await repo.findCurrentEntryUsagePageByMediaId("media-shared", { limit: 2 }); + + expect(page.items.map((item) => item.contentId)).toEqual(["entry-002", "entry-003"]); + expect(page.nextCursor).toEqual(expect.any(String)); + const queries = captured.filter((query) => query.sql.includes("matched_groups")); + expect(queries).toHaveLength(1); + const plan = explain(queries[0]!); + expect(firstSourceOrUsageAccess(plan)).toMatch( + /SEARCH u USING (?:COVERING )?INDEX idx__emdash_media_usage_media_source_generation/, + ); + expectPageBoundedHydration(plan); + expect(plan).toContain("idx__emdash_media_usage_media_source_generation"); + expect(plan).toContain("idx__emdash_media_usage_sources_content"); + expect(plan).not.toContain("SCAN u"); +}); + +/** better-sqlite3 only binds primitives; coerce values captured from Kysely. */ +function bindable(parameter: unknown): unknown { + if (typeof parameter === "boolean") return parameter ? 1 : 0; + if (parameter instanceof Date) return parameter.toISOString(); + if (parameter === undefined) return null; + return parameter; +} + +function explain(query: CapturedQuery): string { + const rows = sqlite + .prepare(`EXPLAIN QUERY PLAN ${query.sql}`) + .all(...query.parameters.map(bindable)) as { detail: string }[]; + return rows.map((row) => row.detail).join("\n"); +} + +function firstSourceOrUsageAccess(plan: string): string | undefined { + return plan.split("\n").find((detail) => /\b(?:SCAN|SEARCH) (?:s|u)\b/.test(detail)); +} + +function expectPageBoundedHydration(plan: string): void { + const mediaAccesses = plan + .split("\n") + .filter( + (detail) => + detail.includes("SEARCH u USING") && + detail.includes("idx__emdash_media_usage_media_source_generation"), + ); + expect(mediaAccesses).toHaveLength(2); + expect(mediaAccesses[0]).toMatch(/\(media_id=\?\)$/); + expect(mediaAccesses[1]).toMatch(/\(media_id=\? AND source_key=\? AND generation=\?\)$/); +} 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 new file mode 100644 index 000000000..59c975de9 --- /dev/null +++ b/packages/core/tests/integration/database/media-usage-read-repository.test.ts @@ -0,0 +1,360 @@ +import { afterEach, beforeEach, expect, it } from "vitest"; + +import { MediaUsageRepository } from "../../../src/database/repositories/media-usage.js"; +import { InvalidCursorError } from "../../../src/database/repositories/types.js"; +import { + buildContentMediaUsageSourceKey, + type MediaUsageContentSourceVariant, +} from "../../../src/media/usage/source-key.js"; +import { SQL_BATCH_SIZE } from "../../../src/utils/chunks.js"; +import { + describeEachDialect, + setupForDialect, + teardownForDialect, + type DialectTestContext, +} from "../../utils/test-db.js"; + +describeEachDialect("MediaUsageRepository reads", (dialect) => { + let ctx: DialectTestContext; + let repo: MediaUsageRepository; + + beforeEach(async () => { + ctx = await setupForDialect(dialect); + repo = new MediaUsageRepository(ctx.db); + }); + + afterEach(async () => { + await teardownForDialect(ctx); + }); + + it("counts distinct active entries across occurrences, locales, and collections", async () => { + await registerCollection(ctx, "pages"); + await registerCollection(ctx, "posts"); + + await repo.replaceSource(contentSource("entry-en", "columns"), [ + occurrence("hero", "media-shared"), + occurrence("body", "media-shared"), + ]); + await repo.replaceSource( + contentSource("entry-fr", "columns", { + locale: "fr", + translationGroup: "translations-1", + }), + [occurrence("hero", "media-shared")], + ); + await repo.replaceSource(contentSource("entry-en", "columns", { collectionSlug: "pages" }), [ + occurrence("hero", "media-shared"), + ]); + + const counts = await repo.findActiveEntryCountsByMediaIds(["media-shared", "media-none"]); + + expect(counts).toEqual( + new Map([ + ["media-shared", 3], + ["media-none", 0], + ]), + ); + }); + + it("selects published sources and the non-published visible working copy", async () => { + await registerCollection(ctx, "posts"); + + await repo.replaceSource( + contentSource("published", "columns", { contentStatus: "published" }), + [occurrence("hero", "media-published")], + ); + await repo.replaceSource( + contentSource("published", "draft_overlay", { contentStatus: "published" }), + [occurrence("hero", "media-pending")], + ); + + await repo.replaceSource(contentSource("draft", "columns", { contentStatus: "draft" }), [ + occurrence("hero", "media-superseded"), + ]); + await repo.replaceSource(contentSource("draft", "draft_overlay", { contentStatus: "draft" }), [ + occurrence("hero", "media-visible"), + ]); + + await repo.replaceSource(contentSource("cleared", "columns", { contentStatus: "draft" }), [ + occurrence("hero", "media-cleared"), + ]); + await repo.replaceSource( + contentSource("cleared", "draft_overlay", { contentStatus: "draft" }), + [], + ); + + await repo.replaceSource(contentSource("failed", "columns", { contentStatus: "draft" }), [ + occurrence("hero", "media-failed-overlay"), + ]); + await repo.markSourceAttempted( + contentSource("failed", "draft_overlay", { + contentStatus: "draft", + sourceCompleteness: "failed", + lastErrorCode: "DRAFT_REVISION_INVALID", + }), + ); + + await repo.replaceSource(contentSource("no-overlay", "columns", { contentStatus: "draft" }), [ + occurrence("hero", "media-columns"), + ]); + + const counts = await repo.findActiveEntryCountsByMediaIds([ + "media-published", + "media-pending", + "media-superseded", + "media-visible", + "media-cleared", + "media-failed-overlay", + "media-columns", + ]); + + expect(Object.fromEntries(counts)).toEqual({ + "media-published": 1, + "media-pending": 1, + "media-superseded": 0, + "media-visible": 1, + "media-cleared": 0, + "media-failed-overlay": 0, + "media-columns": 1, + }); + }); + + it("excludes trash, deleted collections, and stale generations from active counts", async () => { + await registerCollection(ctx, "posts"); + + await repo.replaceSource( + contentSource("trash", "columns", { contentDeletedAt: "2026-01-01T00:00:00.000Z" }), + [occurrence("hero", "media-trash")], + ); + await repo.replaceSource( + contentSource("inconsistent-trash", "columns", { contentStatus: "published" }), + [occurrence("hero", "media-inconsistent-trash")], + ); + await repo.replaceSource( + contentSource("inconsistent-trash", "draft_overlay", { + contentStatus: "published", + contentDeletedAt: "2026-01-02T00:00:00.000Z", + }), + [occurrence("body", "media-inconsistent-trash")], + ); + await repo.replaceSource( + contentSource("ghost", "columns", { collectionSlug: "deleted_collection" }), + [occurrence("hero", "media-ghost")], + ); + await repo.replaceSource(contentSource("generation", "columns"), [ + occurrence("hero", "media-stale-generation"), + ]); + await repo.replaceSource(contentSource("generation", "columns"), [ + occurrence("hero", "media-current-generation"), + ]); + await repo.replaceSource( + contentSource("restored", "columns", { + contentDeletedAt: "2026-01-03T00:00:00.000Z", + }), + [occurrence("hero", "media-restored")], + ); + await repo.replaceSource(contentSource("restored", "columns"), [ + occurrence("hero", "media-restored"), + ]); + await repo.replaceSource(contentSource("provider-only", "columns"), [ + occurrence("hero", "unused-local-id", { + mediaId: null, + provider: "external", + providerAssetId: "media-provider-only", + }), + ]); + + const counts = await repo.findActiveEntryCountsByMediaIds([ + "media-trash", + "media-inconsistent-trash", + "media-ghost", + "media-stale-generation", + "media-current-generation", + "media-restored", + "media-provider-only", + ]); + + expect(Object.fromEntries(counts)).toEqual({ + "media-trash": 0, + "media-inconsistent-trash": 0, + "media-ghost": 0, + "media-stale-generation": 0, + "media-current-generation": 1, + "media-restored": 1, + "media-provider-only": 0, + }); + }); + + it("returns zero-filled counts across multiple D1-sized batches", async () => { + const mediaIds = Array.from({ length: SQL_BATCH_SIZE + 1 }, (_, index) => `media-${index}`); + + const counts = await repo.findActiveEntryCountsByMediaIds(mediaIds); + + expect(counts.size).toBe(mediaIds.length); + expect([...counts.values()].every((count) => count === 0)).toBe(true); + }); + + it("loads status coverage for every current collection and ignores orphan statuses", async () => { + await registerCollection(ctx, "pages"); + await registerCollection(ctx, "posts"); + await repo.upsertIndexStatus({ + adapterId: "content-media", + scopeType: "collection", + scopeKey: "posts", + status: "complete", + schemaVersion: 2, + }); + await repo.upsertIndexStatus({ + adapterId: "content-media", + scopeType: "collection", + scopeKey: "deleted_collection", + status: "failed", + }); + + const scopes = await repo.findCollectionIndexStatusScopes({ + adapterId: "content-media", + scopeType: "collection", + }); + + expect(scopes).toEqual([ + { collectionSlug: "pages", status: null, schemaVersion: null }, + { collectionSlug: "posts", status: "complete", schemaVersion: 2 }, + ]); + }); + + it("paginates complete entry groups with nested sources and occurrences", async () => { + await registerCollection(ctx, "pages"); + await registerCollection(ctx, "posts"); + + await repo.replaceSource( + contentSource("entry-a", "columns", { + contentStatus: "published", + contentTitle: "Published title", + }), + [occurrence("hero", "media-shared"), occurrence("body", "media-shared")], + ); + await repo.replaceSource( + contentSource("entry-a", "draft_overlay", { + contentStatus: "published", + contentTitle: "Draft title", + contentDeletedAt: "2026-01-02T00:00:00.000Z", + }), + [occurrence("draftHero", "media-shared")], + ); + await repo.replaceSource(contentSource("entry-b", "columns", { contentStatus: "draft" }), [ + occurrence("hero", "media-shared"), + ]); + await repo.replaceSource( + contentSource("entry-b", "draft_overlay", { contentStatus: "draft" }), + [occurrence("hero", "media-other")], + ); + await repo.replaceSource(contentSource("entry-c", "columns"), [ + occurrence("hero", "media-shared"), + ]); + await repo.replaceSource(contentSource("entry-d", "columns", { collectionSlug: "pages" }), [ + occurrence("hero", "media-shared"), + ]); + + const first = await repo.findCurrentEntryUsagePageByMediaId("media-shared", { limit: 2 }); + + expect(first.items.map(entryIdentity)).toEqual([ + ["pages", "entry-d"], + ["posts", "entry-a"], + ]); + expect(first.nextCursor).toEqual(expect.any(String)); + + const entryA = first.items[1]!; + expect(entryA.contentDeletedAt).toBe("2026-01-02T00:00:00.000Z"); + expect(entryA.sources.map((source) => source.source.sourceVariant)).toEqual([ + "columns", + "draft_overlay", + ]); + expect( + entryA.sources.flatMap((source) => source.occurrences.map((item) => item.fieldPath)), + ).toEqual(["body", "hero", "draftHero"]); + + const second = await repo.findCurrentEntryUsagePageByMediaId("media-shared", { + limit: 2, + cursor: first.nextCursor, + }); + expect(second.items.map(entryIdentity)).toEqual([["posts", "entry-c"]]); + expect(second.nextCursor).toBeUndefined(); + }); + + it("rejects malformed entry-group cursors", async () => { + await expect( + repo.findCurrentEntryUsagePageByMediaId("media-shared", { cursor: "not-a-cursor" }), + ).rejects.toBeInstanceOf(InvalidCursorError); + }); + + it("defaults non-finite entry-group limits", async () => { + await registerCollection(ctx, "posts"); + await repo.replaceSource(contentSource("entry-a", "columns"), [ + occurrence("hero", "media-shared"), + ]); + + const page = await repo.findCurrentEntryUsagePageByMediaId("media-shared", { + limit: Number.NaN, + }); + + expect(page.items.map(entryIdentity)).toEqual([["posts", "entry-a"]]); + }); +}); + +function contentSource( + contentId: string, + variant: MediaUsageContentSourceVariant, + overrides: Partial[0]> = {}, +): Parameters[0] { + const collectionSlug = overrides.collectionSlug ?? "posts"; + return { + sourceKey: buildContentMediaUsageSourceKey({ + collectionSlug, + contentId, + sourceVariant: variant, + }), + sourceType: "content", + collectionSlug, + contentId, + sourceVariant: variant, + locale: "en", + translationGroup: `tg-${contentId}`, + contentSlug: `slug-${contentId}`, + contentTitle: `Title ${contentId}`, + contentStatus: variant === "columns" ? "published" : "draft", + contentScheduledAt: null, + contentDeletedAt: null, + revisionId: `rev-${contentId}-${variant}`, + ...overrides, + }; +} + +function occurrence( + fieldSlug: string, + mediaId: string, + overrides: Partial[1][number]> = {}, +): Parameters[1][number] { + return { + fieldSlug, + fieldPath: fieldSlug, + occurrenceIndex: 0, + referenceType: "image_field", + mediaId, + provider: "local", + providerAssetId: mediaId, + mediaKind: "image", + mimeType: null, + ...overrides, + }; +} + +async function registerCollection(ctx: DialectTestContext, slug: string): Promise { + await ctx.db + .insertInto("_emdash_collections") + .values({ id: `collection-${slug}`, slug, label: slug, has_seo: 0 }) + .execute(); +} + +function entryIdentity(entry: { collectionSlug: string; contentId: string }): [string, string] { + return [entry.collectionSlug, entry.contentId]; +} diff --git a/packages/core/tests/integration/database/migrations.test.ts b/packages/core/tests/integration/database/migrations.test.ts index 6380bf5d0..c1fb5afad 100644 --- a/packages/core/tests/integration/database/migrations.test.ts +++ b/packages/core/tests/integration/database/migrations.test.ts @@ -138,6 +138,7 @@ describe("Database Migrations (Integration)", () => { "049_taxonomies_name_locale_index", "050_media_usage_index_status", "051_content_taxonomies_denorm", + "052_media_usage_read_index", ]; await db.deleteFrom("_emdash_migrations").where("name", "in", trailing).execute();