release: capture-path resilience — 429 backoff, durable mapping retries, ISRC dedupe (test → main)

app/workflows/__tests__/fetchChunkStep.test.ts

@@ -0,0 +1,22 @@
+import { describe, it, expect, vi, beforeEach } from "vitest";
+import { fetchChunkStep } from "../fetchChunkStep";
+
+import { fetchSpotifyAlbumPlayCounts } from "@/lib/apify/spotify/fetchSpotifyAlbumPlayCounts";
+
+vi.mock("@/lib/apify/spotify/fetchSpotifyAlbumPlayCounts", () => ({
+  fetchSpotifyAlbumPlayCounts: vi.fn(),
+}));
+
+describe("fetchChunkStep", () => {
+  beforeEach(() => vi.clearAllMocks());
+
+  it("runs the actor for the chunk and returns the serializable payload", async () => {
+    const payload = { runId: "run_1", albums: [{ id: "a1", name: "A", tracks: [] }] };
+    vi.mocked(fetchSpotifyAlbumPlayCounts).mockResolvedValue(payload);
+
+    const result = await fetchChunkStep(["a1", "a2"]);
+
+    expect(fetchSpotifyAlbumPlayCounts).toHaveBeenCalledWith(["a1", "a2"]);
+    expect(result).toEqual(payload);
+  });
+});

app/workflows/__tests__/mapAndWriteChunkStep.test.ts

@@ -0,0 +1,39 @@
+import { describe, it, expect, vi, beforeEach } from "vitest";
+import { RetryableError } from "workflow";
+import { mapAndWriteChunkStep } from "../mapAndWriteChunkStep";
+
+import { writeAlbumPlayCounts } from "@/lib/research/playcounts/writeAlbumPlayCounts";
+import { SpotifyRateLimitError } from "@/lib/spotify/SpotifyRateLimitError";
+
+vi.mock("@/lib/research/playcounts/writeAlbumPlayCounts", () => ({
+  writeAlbumPlayCounts: vi.fn(),
+}));
+
+const PAYLOAD = { runId: "run_1", albums: [{ id: "a1", name: "A", tracks: [] }] };
+
+describe("mapAndWriteChunkStep", () => {
+  beforeEach(() => vi.clearAllMocks());
+
+  it("maps + writes against the cached actor payload (no actor re-spend on retry)", async () => {
+    vi.mocked(writeAlbumPlayCounts).mockResolvedValue(12);
+
+    const written = await mapAndWriteChunkStep("snap_1", PAYLOAD);
+
+    expect(writeAlbumPlayCounts).toHaveBeenCalledWith(PAYLOAD.albums, "run_1", {
+      snapshotId: "snap_1",
+    });
+    expect(written).toBe(12);
+  });
+
+  it("escalates sustained Spotify rate limiting to a durable RetryableError", async () => {
+    vi.mocked(writeAlbumPlayCounts).mockRejectedValue(new SpotifyRateLimitError());
+
+    await expect(mapAndWriteChunkStep("snap_1", PAYLOAD)).rejects.toBeInstanceOf(RetryableError);
+  });
+
+  it("lets other failures propagate untyped (step retry/workflow catch handles them)", async () => {
+    vi.mocked(writeAlbumPlayCounts).mockRejectedValue(new Error("db down"));
+
+    await expect(mapAndWriteChunkStep("snap_1", PAYLOAD)).rejects.toThrow("db down");
+  });
+});

app/workflows/fetchChunkStep.ts

@@ -0,0 +1,19 @@
+import {
+  fetchSpotifyAlbumPlayCounts,
+  FetchSpotifyAlbumPlayCountsResult,
+} from "@/lib/apify/spotify/fetchSpotifyAlbumPlayCounts";
+
+/**
+ * Run the play-count actor for one chunk of albums. Isolated as its own step
+ * so the runtime persists the (paid) actor result — mapping retries in the
+ * next step replay against this cached payload instead of re-spending.
+ *
+ * @param albumIds - Album ids in this chunk
+ * @returns The actor run id + parsed album items (serializable)
+ */
+export async function fetchChunkStep(
+  albumIds: string[],
+): Promise<FetchSpotifyAlbumPlayCountsResult> {
+  "use step";
+  return fetchSpotifyAlbumPlayCounts(albumIds);
+}

app/workflows/mapAndWriteChunkStep.ts

@@ -0,0 +1,31 @@
+import { RetryableError } from "workflow";
+import { writeAlbumPlayCounts } from "@/lib/research/playcounts/writeAlbumPlayCounts";
+import { FetchSpotifyAlbumPlayCountsResult } from "@/lib/apify/spotify/fetchSpotifyAlbumPlayCounts";
+import { SpotifyRateLimitError } from "@/lib/spotify/SpotifyRateLimitError";
+
+/**
+ * Map + write one chunk's captured albums against the previous step's cached
+ * actor payload. Sustained Spotify rate limiting escalates to a durable
+ * RetryableError — the runtime suspends and reschedules this step without
+ * re-running the actor (its result is memoized in {@link fetchChunkStep}).
+ *
+ * @param snapshotId - The snapshot job id (lineage)
+ * @param payload - The cached actor result for this chunk
+ * @returns Number of measurements written
+ */
+export async function mapAndWriteChunkStep(
+  snapshotId: string,
+  payload: FetchSpotifyAlbumPlayCountsResult,
+): Promise<number> {
+  "use step";
+  try {
+    return await writeAlbumPlayCounts(payload.albums, payload.runId, { snapshotId });
+  } catch (error) {
+    if (error instanceof SpotifyRateLimitError) {
+      throw new RetryableError("Spotify rate limited during identifier mapping", {
+        retryAfter: "30s",
+      });
+    }
+    throw error;
+  }
+}

app/workflows/playcountSnapshotWorkflow.ts

@@ -1,13 +1,16 @@
 import { getSnapshotStep } from "@/app/workflows/getSnapshotStep";
 import { markSnapshotStep } from "@/app/workflows/markSnapshotStep";
-import { captureSnapshotChunkStep } from "@/app/workflows/captureSnapshotChunkStep";
+import { fetchChunkStep } from "@/app/workflows/fetchChunkStep";
+import { mapAndWriteChunkStep } from "@/app/workflows/mapAndWriteChunkStep";
 
 const CHUNK_SIZE = 100;
 
 /**
  * Durable snapshot capture (recoupable/chat#1791 write path): mark the job
- * running, capture albums in retryable chunks (one actor call per chunk,
- * measurements written with run + snapshot lineage), mark done/failed.
+ * running, then per chunk: a fetch step (actor call — result memoized, so
+ * mapping retries never re-spend) and a map+write step (identifier bootstrap
+ * + measurements; sustained Spotify rate limits escalate to RetryableError
+ * for durable rescheduling), then mark done/failed.
  * Started fire-and-forget from `createSnapshot`; observable in the Vercel
  * dashboard like the sibling workflows.
  */
@@ -21,7 +24,8 @@ export async function playcountSnapshotWorkflow(snapshotId: string) {
     const albumIds = snapshot.album_ids ?? [];
     let written = 0;
     for (let i = 0; i < albumIds.length; i += CHUNK_SIZE) {
-      written += await captureSnapshotChunkStep(snapshotId, albumIds.slice(i, i + CHUNK_SIZE));
+      const payload = await fetchChunkStep(albumIds.slice(i, i + CHUNK_SIZE));
+      written += await mapAndWriteChunkStep(snapshotId, payload);
     }
 
     await markSnapshotStep(snapshotId, { state: "done" });

lib/research/playcounts/__tests__/mapUnmappedAlbumTracks.test.ts

@@ -5,6 +5,7 @@ import generateAccessToken from "@/lib/spotify/generateAccessToken";
 import getTracks from "@/lib/spotify/getTracks";
 import { upsertSongs } from "@/lib/supabase/songs/upsertSongs";
 import { upsertSongIdentifiers } from "@/lib/supabase/song_identifiers/upsertSongIdentifiers";
+import { SpotifyRateLimitError } from "@/lib/spotify/SpotifyRateLimitError";
 
 vi.mock("@/lib/spotify/generateAccessToken", () => ({ default: vi.fn() }));
 vi.mock("@/lib/spotify/getTracks", () => ({ default: vi.fn() }));
@@ -61,6 +62,42 @@ describe("mapUnmappedAlbumTracks", () => {
     expect(mapped.size).toBe(0);
   });
 
+  it("dedupes songs by ISRC across albums (reissues share recordings) before upserting", async () => {
+    const twoAlbums = [
+      { id: "album_std", name: "Album", tracks: [{ id: "t_std", name: "Song", streamCount: 1 }] },
+      {
+        id: "album_dlx",
+        name: "Album (Deluxe)",
+        tracks: [{ id: "t_dlx", name: "Song", streamCount: 1 }],
+      },
+    ];
+    vi.mocked(getTracks).mockResolvedValue({
+      tracks: [
+        { id: "t_std", name: "Song", external_ids: { isrc: "SAME_ISRC" } },
+        { id: "t_dlx", name: "Song", external_ids: { isrc: "SAME_ISRC" } },
+      ],
+      error: null,
+    } as never);
+
+    const mapped = await mapUnmappedAlbumTracks(twoAlbums, new Set());
+
+    // one songs row (DO UPDATE chokes on in-batch duplicates), both track mappings
+    expect(vi.mocked(upsertSongs).mock.calls[0][0]).toHaveLength(1);
+    const idRows = vi.mocked(upsertSongIdentifiers).mock.calls[0][0];
+    expect(
+      idRows.filter((r: { identifier_type: string }) => r.identifier_type === "track_id"),
+    ).toHaveLength(2);
+    expect(mapped.size).toBe(2);
+  });
+
+  it("rethrows sustained rate limiting so workflow steps can escalate durably", async () => {
+    vi.mocked(getTracks).mockRejectedValue(new SpotifyRateLimitError());
+
+    await expect(mapUnmappedAlbumTracks(ALBUMS, new Set())).rejects.toBeInstanceOf(
+      SpotifyRateLimitError,
+    );
+  });
+
   it("degrades to an empty map (no throw) when Spotify auth fails — capture proceeds for already-mapped tracks", async () => {
     const consoleError = vi.spyOn(console, "error").mockImplementation(() => {});
     vi.mocked(generateAccessToken).mockResolvedValue({

lib/research/playcounts/mapUnmappedAlbumTracks.ts

@@ -3,6 +3,7 @@ import getTracks from "@/lib/spotify/getTracks";
 import { upsertSongs } from "@/lib/supabase/songs/upsertSongs";
 import { upsertSongIdentifiers } from "@/lib/supabase/song_identifiers/upsertSongIdentifiers";
 import { SpotifyAlbumPlayCounts } from "@/lib/apify/spotify/fetchSpotifyAlbumPlayCounts";
+import { SpotifyRateLimitError } from "@/lib/spotify/SpotifyRateLimitError";
 
 /**
  * Self-mapping bootstrap (chat#1794): resolve ISRCs for actor tracks that have
@@ -53,9 +54,15 @@ export async function mapUnmappedAlbumTracks(
     });
     if (resolved.length === 0) return new Map();
 
-    await upsertSongs(
-      resolved.map(r => ({ isrc: r.isrc, name: r.name ?? null, album: r.albumName ?? null })),
+    // Dedupe by ISRC: reissues put the same recording on several albums in one
+    // batch, and upsertSongs' DO UPDATE cannot affect the same row twice.
+    const songsByIsrc = new Map(
+      resolved.map(r => [
+        r.isrc,
+        { isrc: r.isrc, name: r.name ?? null, album: r.albumName ?? null },
+      ]),
     );
+    await upsertSongs([...songsByIsrc.values()]);
     await upsertSongIdentifiers(
       resolved.flatMap(r => [
         { song: r.isrc, platform: "spotify", identifier_type: "track_id", value: r.trackId },
@@ -67,6 +74,7 @@ export async function mapUnmappedAlbumTracks(
 
     return new Map(resolved.map(r => [r.trackId, r.isrc]));
   } catch (error) {
+    if (error instanceof SpotifyRateLimitError) throw error;
     console.error("[playcounts] identifier bootstrap failed:", error);
     return new Map();
   }

lib/spotify/SpotifyRateLimitError.ts

@@ -0,0 +1,11 @@
+/**
+ * Spotify rate limiting that survived in-process Retry-After backoff.
+ * Callers in workflow steps escalate it to a durable RetryableError; plain
+ * request-path callers degrade gracefully.
+ */
+export class SpotifyRateLimitError extends Error {
+  constructor(message = "Spotify rate limited after retries") {
+    super(message);
+    this.name = "SpotifyRateLimitError";
+  }
+}

lib/spotify/__tests__/getTracks.test.ts

@@ -1,5 +1,6 @@
 import { describe, it, expect, vi, beforeEach } from "vitest";
 import getTracks from "../getTracks";
+import { SpotifyRateLimitError } from "../SpotifyRateLimitError";
 
 const mockFetch = vi.fn();
 global.fetch = mockFetch as never;
@@ -24,11 +25,44 @@ describe("getTracks", () => {
     expect(tracks).toHaveLength(2);
   });
 
-  it("returns an error on a failed response", async () => {
-    mockFetch.mockResolvedValue({ ok: false, status: 429 } as never);
+  it("retries a 429 after Retry-After and succeeds (rate-limit backoff)", async () => {
+    mockFetch
+      .mockResolvedValueOnce({
+        ok: false,
+        status: 429,
+        headers: { get: (h: string) => (h === "Retry-After" ? "0" : null) },
+      } as never)
+      .mockResolvedValueOnce({
+        ok: true,
+        json: async () => ({ tracks: [{ id: "t1", external_ids: { isrc: "I1" } }] }),
+      } as never);
 
     const { tracks, error } = await getTracks({ ids: ["t1"], accessToken: "tok" });
 
+    expect(mockFetch).toHaveBeenCalledTimes(2);
+    expect(error).toBeNull();
+    expect(tracks).toHaveLength(1);
+  });
+
+  it("throws SpotifyRateLimitError after exhausting 429 retries (callers escalate durably)", async () => {
+    mockFetch.mockResolvedValue({
+      ok: false,
+      status: 429,
+      headers: { get: (h: string) => (h === "Retry-After" ? "0" : null) },
+    } as never);
+
+    await expect(getTracks({ ids: ["t1"], accessToken: "tok" })).rejects.toBeInstanceOf(
+      SpotifyRateLimitError,
+    );
+    expect(mockFetch).toHaveBeenCalledTimes(4); // 1 + 3 retries
+  });
+
+  it("returns an error on a non-429 failed response without retrying", async () => {
+    mockFetch.mockResolvedValue({ ok: false, status: 500 } as never);
+
+    const { tracks, error } = await getTracks({ ids: ["t1"], accessToken: "tok" });
+
+    expect(mockFetch).toHaveBeenCalledTimes(1);
     expect(tracks).toBeNull();
     expect(error).toBeInstanceOf(Error);
   });