Fix scheduled cleanup of superseded media-usage generations

.changeset/cleanup-media-usage-generations.md

@@ -0,0 +1,5 @@
+---
+"emdash": patch
+---
+
+Fixes retention of superseded media-usage projection generations during scheduled maintenance.

AGENTS.md

@@ -367,7 +367,7 @@ In libraries used in a Worker but not themselves Workers, install `@cloudflare/w
 # Testing
 
 - **Framework:** vitest. Tests in `packages/core/tests/`.
-- **No mocks for the DB.** SQLite (`better-sqlite3`) by default. PostgreSQL parity tests via a real `pg` connection with per-test schema isolation (set `PG_CONNECTION_STRING` to opt in).
+- **No mocks for the DB.** SQLite (`better-sqlite3`) by default. PostgreSQL parity tests via a real `pg` connection with per-test schema isolation (set `EMDASH_TEST_PG` to a connection string for a role with `CREATEDB` to opt in).
 - **Utilities:** `tests/utils/test-db.ts` exposes `setupTestDatabase()`, `setupTestDatabaseWithCollections()`, `teardownTestDatabase()` for SQLite and `setupTestPostgresDatabase()` etc. for Postgres. Dialect-agnostic: `setupForDialect`, `setupForDialectWithCollections`, `teardownForDialect`, plus `describeEachDialect(name, fn)`. Use the dialect wrapper for query-builder code -- regressions tend to be dialect-specific.
 - **Structure:** `tests/unit/`, `tests/integration/`, `tests/e2e/` (Playwright). Test files mirror source structure. Each test gets a fresh DB.
 

packages/core/src/cleanup.ts

@@ -17,6 +17,7 @@ import { MediaRepository } from "./database/repositories/media.js";
 import { RevisionRepository } from "./database/repositories/revision.js";
 import type { Database } from "./database/types.js";
 import { removeUploadAttempt } from "./media/upload-attempts.js";
+import { cleanupMediaUsage } from "./media/usage/cleanup.js";
 import type { Storage } from "./storage/types.js";
 
 /**
@@ -30,6 +31,7 @@ export interface CleanupResult {
 	pendingUploadFiles: number;
 	uploadAttempts: number;
 	revisionsPruned: number;
+	mediaUsage: number;
 }
 
 /** Max revisions to keep per entry during periodic pruning */
@@ -60,6 +62,7 @@ export async function runSystemCleanup(
 		pendingUploadFiles: -1,
 		uploadAttempts: -1,
 		revisionsPruned: -1,
+		mediaUsage: -1,
 	};
 
 	// 1. Passkey challenges (expire after 60s, clean anything past 5 min)
@@ -136,6 +139,13 @@ export async function runSystemCleanup(
 		console.error("[cleanup] Failed to prune revisions:", error);
 	}
 
+	try {
+		const mediaUsage = await cleanupMediaUsage(db);
+		result.mediaUsage = mediaUsage.status === "failed" ? -1 : mediaUsage.deletedRows;
+	} catch (error) {
+		console.error("[cleanup] Failed to clean media usage:", error);
+	}
+
 	return result;
 }
 

packages/core/src/database/migrations/055_media_usage_cleanup.ts

@@ -0,0 +1,77 @@
+import { sql, type Kysely } from "kysely";
+
+import { currentTimestamp } from "../dialect-helpers.js";
+
+const CLEANUP_TASK_KEY = "projection_gc";
+const INITIAL_ELIGIBLE_AT = "1970-01-01T00:00:00.000Z";
+
+export async function up(db: Kysely<unknown>): Promise<void> {
+	await db.schema
+		.createIndex("idx__emdash_media_usage_cleanup_scan")
+		.ifNotExists()
+		.on("_emdash_media_usage")
+		.columns(["created_at", "id", "source_key", "generation"])
+		.execute();
+
+	await db.schema
+		.createTable("_emdash_media_usage_cleanup")
+		.ifNotExists()
+		.addColumn("task_key", "text", (c) => c.primaryKey())
+		.addColumn("lease_token", "text")
+		.addColumn("lease_expires_at", "text")
+		.addColumn("next_eligible_at", "text", (c) => c.notNull())
+		.addColumn("cursor_created_at", "text")
+		.addColumn("cursor_id", "text")
+		.addColumn("scan_before_at", "text")
+		.addColumn("consecutive_failures", "integer", (c) => c.notNull().defaultTo(0))
+		.addColumn("last_started_at", "text")
+		.addColumn("last_completed_at", "text")
+		.addColumn("last_candidate_count", "integer", (c) => c.notNull().defaultTo(0))
+		.addColumn("last_deleted_orphans", "integer", (c) => c.notNull().defaultTo(0))
+		.addColumn("last_deleted_stale", "integer", (c) => c.notNull().defaultTo(0))
+		.addColumn("last_deleted_abandoned", "integer", (c) => c.notNull().defaultTo(0))
+		.addColumn("last_deleted_write_leases", "integer", (c) => c.notNull().defaultTo(0))
+		.addColumn("last_backlog_lower_bound", "integer", (c) => c.notNull().defaultTo(0))
+		.addColumn("last_scan_has_more", "integer", (c) => c.notNull().defaultTo(0))
+		.addColumn("last_duration_ms", "integer", (c) => c.notNull().defaultTo(0))
+		.addColumn("last_error_code", "text")
+		.addColumn("updated_at", "text", (c) => c.notNull().defaultTo(currentTimestamp(db)))
+		.execute();
+
+	await db.schema
+		.createTable("_emdash_media_usage_generation_writes")
+		.ifNotExists()
+		.addColumn("source_key", "text", (c) => c.notNull())
+		.addColumn("generation", "text", (c) => c.notNull())
+		.addColumn("lease_token", "text", (c) => c.primaryKey())
+		.addColumn("expires_at", "text", (c) => c.notNull())
+		.addColumn("created_at", "text", (c) => c.notNull().defaultTo(currentTimestamp(db)))
+		.addUniqueConstraint("_emdash_media_usage_generation_writes_source_generation", [
+			"source_key",
+			"generation",
+		])
+		.execute();
+
+	await db.schema
+		.createIndex("idx__emdash_media_usage_generation_writes_expiry")
+		.ifNotExists()
+		.on("_emdash_media_usage_generation_writes")
+		.columns(["expires_at", "lease_token"])
+		.execute();
+
+	await sql`
+		INSERT INTO _emdash_media_usage_cleanup (task_key, next_eligible_at)
+		VALUES (${CLEANUP_TASK_KEY}, ${INITIAL_ELIGIBLE_AT})
+		ON CONFLICT (task_key) DO NOTHING
+	`.execute(db);
+}
+
+export async function down(db: Kysely<unknown>): Promise<void> {
+	await db.schema
+		.dropIndex("idx__emdash_media_usage_generation_writes_expiry")
+		.ifExists()
+		.execute();
+	await db.schema.dropTable("_emdash_media_usage_generation_writes").ifExists().execute();
+	await db.schema.dropTable("_emdash_media_usage_cleanup").ifExists().execute();
+	await db.schema.dropIndex("idx__emdash_media_usage_cleanup_scan").ifExists().execute();
+}