fix(loader): seek taxonomy-filtered listings via a denormalized pivot

.changeset/taxonomy-filter-pivot-seek.md

@@ -0,0 +1,5 @@
+---
+"emdash": patch
+---
+
+Fixes taxonomy-filtered collection listings reading the entire collection on SQLite/D1 when the term is selective. Such listings now seek the matching entries directly, cutting D1 rows-read from tens of thousands to the page size.

packages/core/src/database/migrations/051_content_taxonomies_denorm.ts

@@ -0,0 +1,120 @@
+import type { Kysely } from "kysely";
+import { sql } from "kysely";
+
+import { columnExists, listTablesLike } from "../dialect-helpers.js";
+import { validateIdentifier } from "../validate.js";
+
+/**
+ * Denormalize the filter + sort columns from `ec_*` onto `content_taxonomies`
+ * and mirror `ec_*`'s composite sort indexes onto the pivot (#1834).
+ *
+ * A taxonomy-filtered listing used to apply the term filter as a correlated
+ * `EXISTS` on `SELECT * FROM ec_<collection> … ORDER BY … LIMIT ?`. The pivot
+ * carried only `(collection, entry_id, taxonomy_id)`, so the filter
+ * (`deleted_at`, `status`, `locale`) and the sort (`published_at`/`created_at`)
+ * could only be evaluated on the `ec_*` row. On stats-blind SQLite/D1 a
+ * selective term never fills the `LIMIT`, so the query walked the whole
+ * collection (~75k D1 rows read for a one-row page).
+ *
+ * Copying those columns onto the pivot and indexing them keyed by
+ * `(taxonomy_id, collection, …)` lets the loader seek the term and walk a
+ * sort-ordered pivot index, short-circuiting on `LIMIT`, then touch `ec_*` only
+ * by primary key to hydrate the page.
+ *
+ * The columns are advisory (D1 has no transactions; the write-path re-stamp is
+ * non-atomic), so the read path re-checks the real predicates on the joined
+ * `ec_*` row — see loader.ts. `updated_at` is deliberately not denormalized: it
+ * moves on every edit and is seldom a public sort, so its write cost outweighs
+ * its read value; `updated_at`-sorted taxonomy listings temp-sort instead.
+ *
+ * Order matters: add the columns, backfill, THEN build the indexes so each
+ * index is built once over populated data rather than maintained through the
+ * backfill `UPDATE`.
+ *
+ * Forward-only.
+ */
+
+const DENORM_COLUMNS = [
+	"status",
+	"scheduled_at",
+	"deleted_at",
+	"locale",
+	"published_at",
+	"created_at",
+] as const;
+
+// `entry_id DESC` (not ASC) matters: the listing orders `<sort> DESC, entry_id
+// DESC` (the common case), and a DESC tiebreaker lets SQLite satisfy the whole
+// ORDER BY from the index — no temp B-tree, clean early-`LIMIT`. An ASC
+// tiebreaker forces `USE TEMP B-TREE FOR LAST TERM OF ORDER BY`, which buffers
+// a whole equal-`sortval` block (e.g. a bulk import sharing one timestamp)
+// before emitting. `DESC` also serves the rarer ASC listing via a backward
+// index scan. Mirrors `ec_*`'s `(deleted_at, [locale,] <sort> DESC, id DESC)`.
+const INDEXES: { name: string; columns: string }[] = [
+	{
+		name: "idx_content_taxonomies_pub",
+		columns: "taxonomy_id, collection, deleted_at, published_at DESC, entry_id DESC",
+	},
+	{
+		name: "idx_content_taxonomies_crt",
+		columns: "taxonomy_id, collection, deleted_at, created_at DESC, entry_id DESC",
+	},
+	{
+		name: "idx_content_taxonomies_loc_pub",
+		columns: "taxonomy_id, collection, deleted_at, locale, published_at DESC, entry_id DESC",
+	},
+	{
+		name: "idx_content_taxonomies_loc_crt",
+		columns: "taxonomy_id, collection, deleted_at, locale, created_at DESC, entry_id DESC",
+	},
+];
+
+export async function up(db: Kysely<unknown>): Promise<void> {
+	// 1. Add the six denormalized columns (all nullable, no default). Guarded so
+	//    a partial apply — or a re-run after a mid-migration failure — can retry
+	//    (`ALTER TABLE ADD COLUMN` is not itself idempotent on either dialect).
+	for (const column of DENORM_COLUMNS) {
+		if (await columnExists(db, "content_taxonomies", column)) continue;
+		await sql`
+			ALTER TABLE content_taxonomies ADD COLUMN ${sql.ref(column)} TEXT
+		`.execute(db);
+	}
+
+	// 2. Backfill from each content table by primary key. Drive from the actual
+	//    `ec_*` tables (not `_emdash_collections`) so collections whose table was
+	//    dropped are skipped naturally; the pivot's `collection` value is the
+	//    slug, i.e. the table name without its `ec_` prefix.
+	const tableNames = await listTablesLike(db, "ec_%");
+	for (const tableName of tableNames) {
+		validateIdentifier(tableName, "content table name");
+		const slug = tableName.slice("ec_".length);
+		await sql`
+			UPDATE content_taxonomies
+			SET (status, scheduled_at, deleted_at, locale, published_at, created_at) = (
+				SELECT status, scheduled_at, deleted_at, locale, published_at, created_at
+				FROM ${sql.ref(tableName)}
+				WHERE ${sql.ref(tableName)}.id = content_taxonomies.entry_id
+			)
+			WHERE collection = ${slug}
+		`.execute(db);
+	}
+
+	// 3. Build the composite sort indexes over the now-populated columns.
+	for (const index of INDEXES) {
+		await sql`
+			CREATE INDEX IF NOT EXISTS ${sql.ref(index.name)}
+			ON content_taxonomies (${sql.raw(index.columns)})
+		`.execute(db);
+	}
+}
+
+export async function down(db: Kysely<unknown>): Promise<void> {
+	for (const index of INDEXES) {
+		await sql`DROP INDEX IF EXISTS ${sql.ref(index.name)}`.execute(db);
+	}
+	for (const column of DENORM_COLUMNS) {
+		await sql`
+			ALTER TABLE content_taxonomies DROP COLUMN ${sql.ref(column)}
+		`.execute(db);
+	}
+}

packages/core/src/database/migrations/runner.ts

@@ -53,6 +53,7 @@ import * as m047 from "./047_restore_taxonomy_parent_index.js";
 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";
 
 const MIGRATIONS: Readonly<Record<string, Migration>> = Object.freeze({
 	"001_initial": m001,
@@ -104,6 +105,7 @@ const MIGRATIONS: Readonly<Record<string, Migration>> = Object.freeze({
 	"048_restore_content_taxonomies_term_index": m048,
 	"049_taxonomies_name_locale_index": m049,
 	"050_media_usage_index_status": m050,
+	"051_content_taxonomies_denorm": m051,
 });
 
 /** Total number of registered migrations. Exported for use in tests. */

packages/core/src/database/repositories/content.ts

@@ -637,6 +637,18 @@ export class ContentRepository {
 			.where("deleted_at" as never, "is", null)
 			.execute();
 
+		// Re-stamp the taxonomy pivot only when a denormalized column actually
+		// moved (status/publishedAt/scheduledAt). A plain content edit bumps
+		// `updated_at` — which is not denormalized — so it needs no pivot write,
+		// keeping the common edit path free of taxonomy write amplification.
+		if (
+			input.status !== undefined ||
+			input.publishedAt !== undefined ||
+			input.scheduledAt !== undefined
+		) {
+			await this.restampEntryPivot(type, id);
+		}
+
 		invalidateCollectionCache(type);
 
 		const updated = await this.findById(type, id);
@@ -662,7 +674,10 @@ export class ContentRepository {
 		`.execute(this.db);
 
 		const changed = (result.numAffectedRows ?? 0n) > 0n;
-		if (changed) invalidateCollectionCache(type);
+		if (changed) {
+			await this.restampEntryPivot(type, id);
+			invalidateCollectionCache(type);
+		}
 		return changed;
 	}
 
@@ -683,10 +698,36 @@ export class ContentRepository {
 		const restored = result.rows[0];
 		if (!restored) return null;
 
+		await this.restampEntryPivot(type, id);
 		invalidateCollectionCache(type);
 		return this.mapRow(type, restored);
 	}
 
+	/**
+	 * Re-stamp the denormalized filter + sort columns on every
+	 * `content_taxonomies` pivot row for an entry from its authoritative `ec_*`
+	 * row (migration 051). Called after any mutation that moves one of those
+	 * columns so a taxonomy-filtered listing can seek the entry directly.
+	 *
+	 * A single correlated `UPDATE` reads the post-mutation values from `ec_*`, so
+	 * the pivot converges to the authoritative row. This is NOT atomic with the
+	 * `ec_*` mutation on D1 (no transactions), which is why the read path
+	 * re-checks the real predicates on the joined `ec_*` row. Untagged entries
+	 * have no pivot rows, so the statement is a cheap no-op for them.
+	 */
+	private async restampEntryPivot(type: string, id: string): Promise<void> {
+		const tableName = getTableName(type);
+		await sql`
+			UPDATE content_taxonomies
+			SET (status, scheduled_at, deleted_at, locale, published_at, created_at) = (
+				SELECT status, scheduled_at, deleted_at, locale, published_at, created_at
+				FROM ${sql.ref(tableName)}
+				WHERE ${sql.ref(tableName)}.id = ${id}
+			)
+			WHERE collection = ${type} AND entry_id = ${id}
+		`.execute(this.db);
+	}
+
 	/**
 	 * Permanently delete content (cannot be undone)
 	 */
@@ -996,6 +1037,7 @@ export class ContentRepository {
 			AND deleted_at IS NULL
 		`.execute(this.db);
 
+		await this.restampEntryPivot(type, id);
 		invalidateCollectionCache(type);
 
 		const updated = await this.findById(type, id);
@@ -1035,6 +1077,7 @@ export class ContentRepository {
 			AND deleted_at IS NULL
 		`.execute(this.db);
 
+		await this.restampEntryPivot(type, id);
 		invalidateCollectionCache(type);
 
 		const updated = await this.findById(type, id);
@@ -1339,6 +1382,8 @@ export class ContentRepository {
 			}
 			publishCommitted = true;
 
+			await this.restampEntryPivot(type, id);
+
 			const updated = await this.findById(type, id);
 			if (!updated) {
 				throw new Error("Content not found");
@@ -1427,6 +1472,7 @@ export class ContentRepository {
 			AND deleted_at IS NULL
 		`.execute(this.db);
 
+		await this.restampEntryPivot(type, id);
 		invalidateCollectionCache(type);
 
 		const updated = await this.findById(type, id);

packages/core/src/database/repositories/taxonomy.ts

@@ -1,8 +1,33 @@
-import type { Kysely, Selectable } from "kysely";
+import { sql, type Kysely, type Selectable } from "kysely";
 import { ulid } from "ulidx";
 
 import { invalidateTaxonomyObjectCache } from "../../object-cache/index.js";
-import type { Database, TaxonomyTable, ContentTaxonomyTable } from "../types.js";
+import { isMissingTableError } from "../../utils/db-errors.js";
+import type { Database, TaxonomyTable } from "../types.js";
+import { validateIdentifier } from "../validate.js";
+
+/**
+ * Filter + sort columns denormalized from an entry's `ec_*` row onto its
+ * `content_taxonomies` pivot rows (migration 051). Stamped at insert time so a
+ * newly-tagged entry is immediately seekable by a taxonomy-filtered listing.
+ */
+interface PivotDenorm {
+	status: string | null;
+	scheduled_at: string | null;
+	deleted_at: string | null;
+	locale: string | null;
+	published_at: string | null;
+	created_at: string | null;
+}
+
+const EMPTY_DENORM: PivotDenorm = {
+	status: null,
+	scheduled_at: null,
+	deleted_at: null,
+	locale: null,
+	published_at: null,
+	created_at: null,
+};
 
 export interface Taxonomy {
 	id: string;
@@ -251,14 +276,10 @@ export class TaxonomyRepository {
 		const group = await this.resolveTranslationGroup(taxonomyId);
 		if (!group) return;
 
-		const row: ContentTaxonomyTable = {
-			collection,
-			entry_id: entryId,
-			taxonomy_id: group,
-		};
+		const denorm = await this.fetchEntryDenorm(collection, entryId);
 		await this.db
 			.insertInto("content_taxonomies")
-			.values(row)
+			.values({ collection, entry_id: entryId, taxonomy_id: group, ...denorm })
 			.onConflict((oc) => oc.doNothing())
 			.execute();
 		invalidateTaxonomyObjectCache();
@@ -342,13 +363,15 @@ export class TaxonomyRepository {
 
 		const toAdd = [...newGroups].filter((g) => !currentGroups.has(g));
 		if (toAdd.length > 0) {
+			const denorm = await this.fetchEntryDenorm(collection, entryId);
 			await this.db
 				.insertInto("content_taxonomies")
 				.values(
 					toAdd.map((taxonomy_id) => ({
 						collection,
 						entry_id: entryId,
 						taxonomy_id,
+						...denorm,
 					})),
 				)
 				.onConflict((oc) => oc.doNothing())
@@ -387,20 +410,46 @@ export class TaxonomyRepository {
 			.execute();
 		if (rows.length === 0) return;
 
+		// Stamp the TARGET entry's current values — the copy inherits the source's
+		// term memberships but the target's own status/dates/locale.
+		const denorm = await this.fetchEntryDenorm(collection, targetEntryId);
 		await this.db
 			.insertInto("content_taxonomies")
 			.values(
 				rows.map((r) => ({
 					collection,
 					entry_id: targetEntryId,
 					taxonomy_id: r.taxonomy_id,
+					...denorm,
 				})),
 			)
 			.onConflict((oc) => oc.doNothing())
 			.execute();
 		invalidateTaxonomyObjectCache();
 	}
 
+	/**
+	 * Read the denormalized filter + sort columns from an entry's `ec_*` row so
+	 * they can be stamped onto new pivot rows (migration 051). A missing table or
+	 * missing row yields all-nulls: the pivot columns are advisory, and the
+	 * listing read path re-checks the authoritative `ec_*` row regardless.
+	 */
+	private async fetchEntryDenorm(collection: string, entryId: string): Promise<PivotDenorm> {
+		validateIdentifier(collection, "collection type");
+		const tableName = `ec_${collection}`;
+		try {
+			const result = await sql<PivotDenorm>`
+				SELECT status, scheduled_at, deleted_at, locale, published_at, created_at
+				FROM ${sql.ref(tableName)}
+				WHERE id = ${entryId}
+			`.execute(this.db);
+			return result.rows[0] ?? EMPTY_DENORM;
+		} catch (error) {
+			if (isMissingTableError(error)) return EMPTY_DENORM;
+			throw error;
+		}
+	}
+
 	/**
 	 * Count content entries that use any translation of this term. Accepts
 	 * either a term id or a translation_group — we normalise to the group.

packages/core/src/database/types.ts

@@ -28,6 +28,27 @@ export interface ContentTaxonomyTable {
 	collection: string; // e.g., 'posts'
 	entry_id: string; // ID in the ec_* table
 	taxonomy_id: string; // stores taxonomies.translation_group (locale-agnostic)
+	// Denormalized filter + sort columns mirrored from the entry's ec_* row
+	// (migration 051). They let a taxonomy-filtered listing seek the matching
+	// entries directly on the pivot instead of scanning the whole collection.
+	//
+	// ADVISORY, not authoritative: D1 has no transactions, so the write-path
+	// re-stamp (ContentRepository / TaxonomyRepository) is a separate statement
+	// from the ec_* mutation and can be transiently stale. The read path narrows
+	// the candidate set with these columns but re-checks the real filter
+	// predicates on the joined ec_* row. `updated_at` is deliberately NOT
+	// denormalized — it moves on every edit, so denormalizing it would force a
+	// pivot re-stamp on the common edit path for little read value.
+	//
+	// Nullable/Generated so inserts that predate a re-stamp (or the migration
+	// backfill) leave them NULL; every insert site in the repositories stamps
+	// them explicitly.
+	status: Generated<string | null>;
+	scheduled_at: Generated<string | null>;
+	deleted_at: Generated<string | null>;
+	locale: Generated<string | null>;
+	published_at: Generated<string | null>;
+	created_at: Generated<string | null>;
 }
 
 export interface TaxonomyDefTable {