MediaCache: content-addressed media store with ledger and GC sweep

packages/host/config/schema/1785953445767_schema.sql → packages/host/config/schema/1787258803063_schema.sql

@@ -76,6 +76,21 @@
    PRIMARY KEY ( id ) 
 );
 
+ CREATE TABLE IF NOT EXISTS media_cache_ledger (
+   realm_url TEXT NOT NULL,
+   source_url TEXT NOT NULL,
+   capture_spec_hash TEXT NOT NULL,
+   source_generation INTEGER NOT NULL,
+   object_key TEXT NOT NULL,
+   source_content_hash TEXT,
+   lane TEXT NOT NULL,
+   content_type TEXT NOT NULL,
+   size_bytes NOT NULL,
+   created_at NOT NULL,
+   last_accessed_at NOT NULL,
+   PRIMARY KEY ( realm_url, source_url, capture_spec_hash, source_generation ) 
+);
+
  CREATE TABLE IF NOT EXISTS module_transpile_cache (
    realm_url TEXT NOT NULL,
    canonical_path TEXT NOT NULL,

packages/postgres/migrations/1787258803063_add-media-cache-ledger.js

@@ -0,0 +1,52 @@
+exports.shorthands = undefined;
+
+// The MediaCache ledger: one row per derived-media capture (a screenshot of
+// one source instance under one canonical capture spec at one generation),
+// pointing at a content-addressed object (`object_key` = hash of the output
+// bytes) in the configured media store. The ledger is the store's only
+// catalog — GC reclaims objects by scanning these rows, never by enumerating
+// the bucket — so every object write must be paired with a ledger row.
+//
+// Several rows may share one `object_key`: identical output bytes are stored
+// once (dedupe-on-write) and the object is reclaimable only when its last
+// referencing row is gone.
+//
+// `lane` separates GC policy: 'declared' rows (indexing-time declared
+// screenshots) are superseded by newer generations of the same capture,
+// while 'on-demand' rows (URL-DSL / POST captures) additionally age out by
+// last access. `created_at` / `last_accessed_at` are unix-ms bigints like
+// `prerendered_html.rendered_at` (pg returns them as JS strings).
+
+exports.up = (pgm) => {
+  pgm.createTable('media_cache_ledger', {
+    realm_url: { type: 'varchar', notNull: true },
+    source_url: { type: 'varchar', notNull: true },
+    capture_spec_hash: { type: 'varchar', notNull: true },
+    source_generation: { type: 'integer', notNull: true },
+    object_key: { type: 'varchar', notNull: true },
+    // The source file's content hash for captures keyed by file content
+    // rather than by generation (FileDef posters); null otherwise.
+    source_content_hash: { type: 'varchar' },
+    lane: { type: 'varchar', notNull: true },
+    content_type: { type: 'varchar', notNull: true },
+    size_bytes: { type: 'bigint', notNull: true },
+    created_at: { type: 'bigint', notNull: true },
+    last_accessed_at: { type: 'bigint', notNull: true },
+  });
+  pgm.addConstraint('media_cache_ledger', 'media_cache_ledger_pkey', {
+    primaryKey: [
+      'realm_url',
+      'source_url',
+      'capture_spec_hash',
+      'source_generation',
+    ],
+  });
+  // Reference counting at GC time: is this object's key still referenced?
+  pgm.createIndex('media_cache_ledger', ['object_key']);
+  // The on-demand age-out lane scans by lane + last access.
+  pgm.createIndex('media_cache_ledger', ['lane', 'last_accessed_at']);
+};
+
+exports.down = (pgm) => {
+  pgm.dropTable('media_cache_ledger');
+};

packages/postgres/pg-queue.ts

@@ -34,6 +34,7 @@ import { FROM_SCRATCH_JOB_TIMEOUT_SEC } from '@cardstack/runtime-common/tasks/in
 // coalesce handlers registered before publish() is called.
 import '@cardstack/runtime-common/tasks/copy';
 import '@cardstack/runtime-common/tasks/full-reindex';
+import '@cardstack/runtime-common/tasks/media-cache-gc';
 import '@cardstack/runtime-common/tasks/prerender-html';
 import '@cardstack/runtime-common/tasks/prerender-html-reconcile';
 import type { PgAdapter } from './pg-adapter.ts';

packages/realm-server/lib/cron-scheduler.ts

@@ -19,6 +19,12 @@ import {
   PRERENDER_HTML_RECONCILE_CRON_TZ,
   createPrerenderHtmlReconcileCronJob,
 } from './prerender-html-reconcile-config.ts';
+import { enqueueMediaCacheGc } from '../scripts/media-cache-gc.ts';
+import {
+  MEDIA_CACHE_GC_CRON_SCHEDULE,
+  MEDIA_CACHE_GC_CRON_TZ,
+  createMediaCacheGcCronJob,
+} from './media-cache-gc-config.ts';
 
 let log = logger('cron-scheduler');
 
@@ -39,6 +45,11 @@ export function startCronJobs(): void {
   if (prerenderHtmlReconcileJob) {
     jobs.push(prerenderHtmlReconcileJob);
   }
+
+  let mediaCacheGcJob = startMediaCacheGcCron();
+  if (mediaCacheGcJob) {
+    jobs.push(mediaCacheGcJob);
+  }
 }
 
 export function stopCronJobs(): void {
@@ -121,3 +132,23 @@ function startPrerenderHtmlReconcileCron(): CronJob | undefined {
   );
   return job;
 }
+
+function startMediaCacheGcCron(): CronJob | undefined {
+  let job = createMediaCacheGcCronJob(
+    async () => {
+      try {
+        await enqueueMediaCacheGc();
+      } catch (error) {
+        Sentry.captureException(error);
+        log.error('media-cache-gc cron failed to enqueue job', error);
+      }
+    },
+    { runOnInit: false },
+  );
+
+  job.start();
+  log.info(
+    `media-cache-gc cron scheduled for ${MEDIA_CACHE_GC_CRON_SCHEDULE} ${MEDIA_CACHE_GC_CRON_TZ}`,
+  );
+  return job;
+}

packages/realm-server/lib/media-cache-gc-config.ts

@@ -0,0 +1,25 @@
+import { CronJob } from 'cron';
+
+// Daily is plenty: the sweep's min-age delay is measured in hours and the
+// on-demand TTL in days, so nothing it reclaims is urgent. Off the top of
+// the hour to stay clear of the hourly prerender-html reconcile. Cadence is
+// a tuning knob via the env override.
+export const MEDIA_CACHE_GC_CRON_SCHEDULE =
+  process.env.MEDIA_CACHE_GC_CRON_SCHEDULE ?? '30 2 * * *';
+export const MEDIA_CACHE_GC_CRON_TZ =
+  process.env.MEDIA_CACHE_GC_CRON_TZ ?? 'America/New_York';
+
+export function createMediaCacheGcCronJob(
+  onTick: () => void,
+  options: { runOnInit?: boolean } = {},
+) {
+  return new CronJob(
+    MEDIA_CACHE_GC_CRON_SCHEDULE,
+    onTick,
+    null,
+    false,
+    MEDIA_CACHE_GC_CRON_TZ,
+    null,
+    options.runOnInit ?? false,
+  );
+}

packages/realm-server/media-cache/index.ts

@@ -0,0 +1,37 @@
+import type { MediaCacheAdapter } from '@cardstack/runtime-common';
+import { logger } from '@cardstack/runtime-common';
+import { S3MediaCacheAdapter } from './s3-adapter.ts';
+import { LocalDiskMediaCacheAdapter } from './local-disk-adapter.ts';
+
+export { S3MediaCacheAdapter } from './s3-adapter.ts';
+export { LocalDiskMediaCacheAdapter } from './local-disk-adapter.ts';
+
+const log = logger('media-cache');
+
+// The process-level choice of MediaCache object store:
+//   MEDIA_CACHE_BUCKET  → S3 (deployed environments; optional
+//                         MEDIA_CACHE_KEY_PREFIX / MEDIA_CACHE_REGION)
+//   MEDIA_CACHE_DIR     → local disk (dev / tests)
+//   neither             → no store; media-cache tasks no-op.
+// Bucket wins if both are set, so a deployed env var can't be shadowed by a
+// stray local one.
+export function createMediaCacheAdapterFromEnv():
+  | MediaCacheAdapter
+  | undefined {
+  let bucket = process.env.MEDIA_CACHE_BUCKET?.trim();
+  if (bucket) {
+    return new S3MediaCacheAdapter({
+      bucket,
+      keyPrefix: process.env.MEDIA_CACHE_KEY_PREFIX?.trim() || '',
+      region: process.env.MEDIA_CACHE_REGION?.trim() || undefined,
+    });
+  }
+  let dir = process.env.MEDIA_CACHE_DIR?.trim();
+  if (dir) {
+    return new LocalDiskMediaCacheAdapter({ dir });
+  }
+  log.info(
+    'neither MEDIA_CACHE_BUCKET nor MEDIA_CACHE_DIR is set; media cache is disabled for this process',
+  );
+  return undefined;
+}

packages/realm-server/media-cache/local-disk-adapter.ts

@@ -0,0 +1,73 @@
+// Local-disk MediaCache object store for dev and tests. Objects live at
+// `<dir>/<key[0..2]>/<key>` — the two-character fan-out keeps any one
+// directory from accumulating every object. No metadata is stored beside the
+// bytes: the ledger's `content_type` column is the serving-path source of
+// truth, so `head` reports size only.
+
+import { createReadStream } from 'node:fs';
+import { mkdir, rename, rm, stat, writeFile } from 'node:fs/promises';
+import { dirname, join } from 'node:path';
+import type {
+  MediaCacheAdapter,
+  MediaObjectStat,
+} from '@cardstack/runtime-common';
+
+export class LocalDiskMediaCacheAdapter implements MediaCacheAdapter {
+  #dir: string;
+
+  constructor({ dir }: { dir: string }) {
+    this.#dir = dir;
+  }
+
+  private pathFor(key: string): string {
+    return join(this.#dir, key.slice(0, 2), key);
+  }
+
+  async put(
+    key: string,
+    bytes: Uint8Array,
+    _opts: { contentType: string },
+  ): Promise<void> {
+    let path = this.pathFor(key);
+    // The key is a hash of the bytes, so an existing file under this key
+    // already holds them — skip the write (dedupe-on-write).
+    if (await this.head(key)) {
+      return;
+    }
+    await mkdir(dirname(path), { recursive: true });
+    // Write-then-rename so a reader never sees a half-written object and a
+    // crashed write leaves only a stray temp file, not a corrupt object. The
+    // temp name carries the pid so two processes writing the same key (both
+    // saw it absent) don't collide; rename is atomic, and last-writer-wins is
+    // harmless because both hold the same bytes.
+    let tempPath = `${path}.${process.pid}.tmp`;
+    await writeFile(tempPath, bytes);
+    await rename(tempPath, path);
+  }
+
+  async head(key: string): Promise<MediaObjectStat | undefined> {
+    try {
+      let stats = await stat(this.pathFor(key));
+      return { size: stats.size };
+    } catch (error: any) {
+      if (error?.code === 'ENOENT') {
+        return undefined;
+      }
+      throw error;
+    }
+  }
+
+  async getStream(key: string): Promise<AsyncIterable<Uint8Array> | undefined> {
+    // Existence-check first: createReadStream reports a missing file only as
+    // an async 'error' event, which would escape as an unhandled stream
+    // error rather than this interface's `undefined`.
+    if (!(await this.head(key))) {
+      return undefined;
+    }
+    return createReadStream(this.pathFor(key));
+  }
+
+  async delete(key: string): Promise<void> {
+    await rm(this.pathFor(key), { force: true });
+  }
+}