perf(metadata): process episode art in the background with bounded concurrency

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

packages/core/src/concurrency.ts

@@ -0,0 +1,21 @@
+/** Run `fn` over `items` with at most `concurrency` calls in flight; never rejects. */
+export async function mapWithConcurrency<T, R>(
+  items: T[],
+  fn: (item: T) => Promise<R>,
+  concurrency: number,
+): Promise<PromiseSettledResult<R>[]> {
+  const results: PromiseSettledResult<R>[] = Array.from({ length: items.length });
+  let i = 0;
+  async function worker() {
+    while (i < items.length) {
+      const idx = i++;
+      try {
+        results[idx] = { status: "fulfilled", value: await fn(items[idx]) };
+      } catch (reason) {
+        results[idx] = { status: "rejected", reason };
+      }
+    }
+  }
+  await Promise.all(Array.from({ length: Math.min(concurrency, items.length) }, worker));
+  return results;
+}

packages/core/src/image-cache.ts

@@ -12,8 +12,12 @@ import {
 import { createLogger } from "@sofa/logger";
 import { IMAGE_CATEGORY_SIZES, type ImageCategory, tmdbCdnImageUrl } from "@sofa/tmdb/image";
 
+import { mapWithConcurrency } from "./concurrency";
+
 const log = createLogger("image-cache");
 
+const STILL_DOWNLOAD_CONCURRENCY = 6;
+
 export type { ImageCategory };
 
 export function imageCacheEnabled(): boolean {
@@ -211,11 +215,12 @@ export async function cacheEpisodeStills(titleId: string) {
       cached: await isImageCached("stills", path.basename(ep.stillPath)),
     })),
   );
-  const tasks = checks
-    .filter((c) => !c.cached)
-    .map((c) => downloadAndCacheImage(c.stillPath, "stills"));
-
-  await Promise.allSettled(tasks);
+  const uncached = checks.filter((c) => !c.cached);
+  await mapWithConcurrency(
+    uncached,
+    (c) => downloadAndCacheImage(c.stillPath, "stills"),
+    STILL_DOWNLOAD_CONCURRENCY,
+  );
 }
 
 export async function cacheProviderLogos(titleId: string) {

packages/core/src/metadata.ts

@@ -46,6 +46,7 @@ import { tmdbImageUrl } from "@sofa/tmdb/image";
 
 import { refreshAvailability } from "./availability";
 import { extractAndStoreColors, parseColorPalette } from "./colors";
+import { mapWithConcurrency } from "./concurrency";
 import { getCastForTitle, refreshCredits } from "./credits";
 import {
   cacheEpisodeStills,
@@ -64,26 +65,7 @@ import {
 
 const log = createLogger("metadata");
 
-async function mapWithConcurrency<T, R>(
-  items: T[],
-  fn: (item: T) => Promise<R>,
-  concurrency: number,
-): Promise<PromiseSettledResult<R>[]> {
-  const results: PromiseSettledResult<R>[] = Array.from({ length: items.length });
-  let i = 0;
-  async function worker() {
-    while (i < items.length) {
-      const idx = i++;
-      try {
-        results[idx] = { status: "fulfilled", value: await fn(items[idx]) };
-      } catch (reason) {
-        results[idx] = { status: "rejected", reason };
-      }
-    }
-  }
-  await Promise.all(Array.from({ length: Math.min(concurrency, items.length) }, worker));
-  return results;
-}
+const THUMBHASH_CONCURRENCY = 4;
 
 export function updateTitleWithArtInvalidation(
   title: Pick<
@@ -893,11 +875,12 @@ export async function refreshTrailer(titleId: string) {
 async function generateMissingTvChildThumbHashes(titleId: string) {
   const titleSeasons = getSeasonsForTitle(titleId);
 
-  const hashTasks: Promise<unknown>[] = [];
+  const work: (() => Promise<unknown>)[] = [];
 
   for (const s of titleSeasons) {
     if (s.posterPath && !s.posterThumbHash) {
-      hashTasks.push(generateSeasonThumbHash(s.id, s.posterPath));
+      const posterPath = s.posterPath;
+      work.push(() => generateSeasonThumbHash(s.id, posterPath));
     }
   }
 
@@ -905,11 +888,11 @@ async function generateMissingTvChildThumbHashes(titleId: string) {
   if (seasonIds.length > 0) {
     const epsNeedingHash = getEpisodesNeedingStillHash(seasonIds);
     for (const ep of epsNeedingHash) {
-      hashTasks.push(generateEpisodeThumbHash(ep.id, ep.stillPath));
+      work.push(() => generateEpisodeThumbHash(ep.id, ep.stillPath));
     }
   }
 
-  await Promise.all(hashTasks);
+  await mapWithConcurrency(work, (run) => run(), THUMBHASH_CONCURRENCY);
 }
 
 export async function syncTvChildArt(titleId: string, options?: { warmCache?: boolean }) {
@@ -922,7 +905,8 @@ export async function syncTvChildArt(titleId: string, options?: { warmCache?: bo
 
 /**
  * Warm the image cache for a title, then derive poster colors and thumbhashes.
- * Sequencing avoids duplicate TMDB downloads on cold caches.
+ * Sequencing avoids duplicate TMDB downloads on cold caches. For TV, episode stills
+ * and season/episode thumbhashes continue in the background after this resolves.
  */
 async function syncTitleArt(
   titleId: string,
@@ -932,9 +916,6 @@ async function syncTitleArt(
 ) {
   if (imageCacheEnabled()) {
     await cacheImagesForTitle(titleId);
-    if (type === "tv") {
-      await cacheEpisodeStills(titleId);
-    }
   }
 
   const posterBuffer = posterPath ? await loadImageBuffer(posterPath, "posters") : undefined;
@@ -947,7 +928,9 @@ async function syncTitleArt(
   ]);
 
   if (type === "tv") {
-    await syncTvChildArt(titleId, { warmCache: false });
+    // Episode stills and season/episode thumbhashes can number in the hundreds; callers
+    // (often a title-page request) shouldn't wait for them.
+    syncTvChildArt(titleId).catch((err) => log.debug("TV child art sync failed:", err));
   }
 }
 

packages/core/test/concurrency.test.ts

@@ -0,0 +1,64 @@
+import { describe, expect, test } from "vitest";
+
+import { mapWithConcurrency } from "../src/concurrency";
+
+describe("mapWithConcurrency", () => {
+  test("runs every item and returns results in input order", async () => {
+    const items = [1, 2, 3, 4, 5];
+    const results = await mapWithConcurrency(
+      items,
+      async (n) => {
+        // Reverse completion order relative to input order
+        await new Promise((r) => setTimeout(r, (6 - n) * 2));
+        return n * 10;
+      },
+      2,
+    );
+
+    expect(results).toEqual(items.map((n) => ({ status: "fulfilled", value: n * 10 })));
+  });
+
+  test("never has more than `concurrency` calls in flight", async () => {
+    let inFlight = 0;
+    let maxInFlight = 0;
+    const items = Array.from({ length: 20 }, (_, i) => i);
+
+    const results = await mapWithConcurrency(
+      items,
+      async (n) => {
+        inFlight++;
+        maxInFlight = Math.max(maxInFlight, inFlight);
+        await new Promise((r) => setTimeout(r, 5));
+        inFlight--;
+        return n;
+      },
+      3,
+    );
+
+    expect(results).toHaveLength(20);
+    expect(maxInFlight).toBe(3);
+  });
+
+  test("a rejecting item yields a rejected result at its index and the others still run", async () => {
+    const ran: number[] = [];
+    const results = await mapWithConcurrency(
+      [0, 1, 2, 3],
+      async (n) => {
+        ran.push(n);
+        if (n === 1) throw new Error("boom");
+        return n;
+      },
+      2,
+    );
+
+    expect(ran.toSorted()).toEqual([0, 1, 2, 3]);
+    expect(results[0]).toEqual({ status: "fulfilled", value: 0 });
+    expect(results[1].status).toBe("rejected");
+    expect(results[2]).toEqual({ status: "fulfilled", value: 2 });
+    expect(results[3]).toEqual({ status: "fulfilled", value: 3 });
+  });
+
+  test("empty input resolves to an empty array", async () => {
+    expect(await mapWithConcurrency([], async () => 1, 3)).toEqual([]);
+  });
+});

packages/core/test/title-art-detach.test.ts

@@ -0,0 +1,105 @@
+import { beforeEach, describe, expect, test, vi } from "vitest";
+
+import { clearAllTables, insertTitle } from "@sofa/test/db";
+
+const { mockRefreshCredits, mockRefreshAvailability, mockCacheEpisodeStills } = vi.hoisted(() => ({
+  mockRefreshCredits: vi.fn<(titleId: string) => Promise<void>>(async () => {}),
+  mockRefreshAvailability: vi.fn<(titleId: string) => Promise<void>>(async () => {}),
+  // Never resolves: simulates hundreds of slow episode still downloads
+  mockCacheEpisodeStills: vi.fn<(titleId: string) => Promise<void>>(
+    () => new Promise<void>(() => {}),
+  ),
+}));
+
+vi.mock("@sofa/tmdb/client", () => ({
+  getMovieDetails: async () => {
+    throw new Error("not used");
+  },
+  getRecommendations: async () => ({ results: [] }),
+  getSimilar: async () => ({ results: [] }),
+  getVideos: async () => ({ results: [] }),
+  getTvDetails: async () => ({
+    id: 778,
+    name: "Detached Show",
+    original_name: "Detached Show",
+    overview: "o",
+    first_air_date: "2020-01-05",
+    poster_path: "/poster.jpg",
+    backdrop_path: null,
+    popularity: 1,
+    vote_average: 7,
+    vote_count: 10,
+    status: "Returning Series",
+    number_of_seasons: 1,
+    genres: [],
+    external_ids: { imdb_id: "tt778", tvdb_id: 8 },
+    original_language: "en",
+    content_ratings: { results: [] },
+  }),
+  getTvSeasonDetails: async () => ({
+    season_number: 1,
+    name: "Season 1",
+    overview: null,
+    poster_path: null,
+    air_date: null,
+    episodes: [
+      {
+        episode_number: 1,
+        name: "E1",
+        overview: null,
+        still_path: "/still.jpg",
+        air_date: "2020-01-05",
+        runtime: 30,
+      },
+    ],
+  }),
+}));
+
+vi.mock("../src/credits", () => ({
+  refreshCredits: mockRefreshCredits,
+  getCastForTitle: () => [],
+}));
+
+vi.mock("../src/availability", () => ({
+  refreshAvailability: mockRefreshAvailability,
+}));
+
+vi.mock("../src/image-cache", () => ({
+  cacheEpisodeStills: mockCacheEpisodeStills,
+  cacheImagesForTitle: async () => {},
+  deleteOrphanedImage: async () => {},
+  imageCacheEnabled: () => true,
+  loadImageBuffer: async () => undefined,
+}));
+
+vi.mock("../src/thumbhash", () => ({
+  generateEpisodeThumbHash: async () => {},
+  generateSeasonThumbHash: async () => {},
+  generateTitleBackdropThumbHash: async () => {},
+  generateTitlePosterThumbHash: async () => {},
+}));
+
+vi.mock("../src/colors", () => ({
+  extractAndStoreColors: async () => {},
+  parseColorPalette: () => null,
+}));
+
+import { getOrFetchTitle } from "../src/metadata";
+
+beforeEach(() => {
+  clearAllTables();
+  mockRefreshCredits.mockClear();
+  mockRefreshAvailability.mockClear();
+  mockCacheEpisodeStills.mockClear();
+});
+
+describe("getOrFetchTitle episode art", () => {
+  test("does not wait for episode still caching on the first view of a TV show", async () => {
+    insertTitle({ id: "tv-shell", tmdbId: 778, type: "tv", title: "Shell" });
+
+    const result = await getOrFetchTitle("tv-shell");
+
+    expect(result?.title.title).toBe("Detached Show");
+    expect(mockCacheEpisodeStills).toHaveBeenCalledWith("tv-shell");
+  });
+});