feat(research): measurement-jobs write resource + retryable backfill drain (chat#1796)

app/api/research/measurement-jobs/route.ts

@@ -0,0 +1,26 @@
+import { NextRequest, NextResponse } from "next/server";
+import { getCorsHeaders } from "@/lib/networking/getCorsHeaders";
+import { createMeasurementJobHandler } from "@/lib/research/measurement_jobs/createMeasurementJobHandler";
+
+export const maxDuration = 60;
+
+/**
+ * OPTIONS /api/research/measurement-jobs — CORS preflight.
+ *
+ * @returns CORS-enabled 200 response
+ */
+export async function OPTIONS() {
+  return new NextResponse(null, { status: 200, headers: getCorsHeaders() });
+}
+
+/**
+ * POST /api/research/measurement-jobs — create an async ingest job.
+ * `source:"current"` captures present counts; `source:"historical"` enqueues
+ * Songstats deep backfill. Body: `{ scope, source, platforms? }`.
+ *
+ * @param request - body: scope (one of catalog_id/album_ids/isrcs) + source
+ * @returns 202 JSON job payload or error
+ */
+export async function POST(request: NextRequest) {
+  return createMeasurementJobHandler(request);
+}

app/workflows/__tests__/backfillTrackStep.test.ts

@@ -77,11 +77,12 @@ describe("backfillTrackStep", () => {
     expect(result).toEqual({ ok: true, hitsSpent: 1 });
   });
 
-  it("marks the row failed on upstream error and still records the spend attempt", async () => {
+  it("marks a transient upstream error (429) as failed (reclaimable) and records the spend", async () => {
     vi.mocked(fetchSongstats).mockResolvedValue({ status: 429, data: {} });
 
     const result = await backfillTrackStep(ROW);
 
+    // transient -> 'failed' so the daily reclaim sweep returns it to 'pending'
     expect(updateSongstatsBackfillQueue).toHaveBeenCalledWith("q1", { status: "failed" });
     expect(insertSongstatsQuotaLedger).toHaveBeenCalledWith({
       hits: 1,
@@ -90,4 +91,42 @@ describe("backfillTrackStep", () => {
     expect(upsertSongMeasurements).not.toHaveBeenCalled();
     expect(result).toEqual({ ok: false, hitsSpent: 1 });
   });
+
+  it("marks a transient 5xx as failed (reclaimable)", async () => {
+    vi.mocked(fetchSongstats).mockResolvedValue({ status: 504, data: {} });
+
+    const result = await backfillTrackStep(ROW);
+
+    expect(updateSongstatsBackfillQueue).toHaveBeenCalledWith("q1", { status: "failed" });
+    expect(result).toEqual({ ok: false, hitsSpent: 1 });
+  });
+
+  it("marks a permanent client error (403) as done so reclaim never recycles it", async () => {
+    vi.mocked(fetchSongstats).mockResolvedValue({ status: 403, data: {} });
+
+    const result = await backfillTrackStep(ROW);
+
+    // non-retryable 4xx (not 408/429) is terminal -> 'done', not 'failed'
+    expect(updateSongstatsBackfillQueue).toHaveBeenCalledWith("q1", { status: "done" });
+    expect(insertSongstatsQuotaLedger).toHaveBeenCalledWith({
+      hits: 1,
+      purpose: "backfill USA2P2015959 (terminal 403)",
+    });
+    expect(result).toEqual({ ok: false, hitsSpent: 1 });
+  });
+
+  it("marks a definitive 404 (no history exists) as done so it is never retried", async () => {
+    vi.mocked(fetchSongstats).mockResolvedValue({ status: 404, data: {} });
+
+    const result = await backfillTrackStep(ROW);
+
+    // terminal no-data -> 'done', not 'failed' — the reclaim sweep must not resurrect it
+    expect(updateSongstatsBackfillQueue).toHaveBeenCalledWith("q1", { status: "done" });
+    expect(insertSongstatsQuotaLedger).toHaveBeenCalledWith({
+      hits: 1,
+      purpose: "backfill USA2P2015959 (no data 404)",
+    });
+    expect(upsertSongMeasurements).not.toHaveBeenCalled();
+    expect(result).toEqual({ ok: false, hitsSpent: 1 });
+  });
 });

app/workflows/backfillTrackStep.ts

@@ -26,11 +26,22 @@ export async function backfillTrackStep(
   });
 
   if (result.status !== 200) {
-    await insertSongstatsQuotaLedger({
-      hits: 1,
-      purpose: `backfill ${row.song} (failed ${result.status})`,
-    });
-    await updateSongstatsBackfillQueue(row.id, { status: "failed" });
+    const status = result.status;
+    const isNoData = status === 404;
+    // Only transient errors are retryable: 408 (timeout), 429 (quota), any 5xx.
+    const isRetryable = status === 408 || status === 429 || status >= 500;
+
+    // `failed` is reclaimable (the daily sweep returns it to `pending`, bounded
+    // by the rolling-window budget). 404 no-data and other permanent 4xx are
+    // terminal → `done`, so reclaim never recycles a track that can't succeed.
+    const nextStatus = isRetryable ? "failed" : "done";
+
+    let outcome = `terminal ${status}`;
+    if (isNoData) outcome = "no data 404";
+    else if (isRetryable) outcome = `failed ${status}`;
+
+    await insertSongstatsQuotaLedger({ hits: 1, purpose: `backfill ${row.song} (${outcome})` });
+    await updateSongstatsBackfillQueue(row.id, { status: nextStatus });
     return { ok: false, hitsSpent: 1 };
   }
 

lib/research/measurement_jobs/__tests__/createMeasurementJob.test.ts

@@ -0,0 +1,61 @@
+import { describe, it, expect, vi, beforeEach } from "vitest";
+import { createMeasurementJob } from "../createMeasurementJob";
+import { createSnapshot } from "@/lib/research/playcounts/createSnapshot";
+import { enqueueHistoricalBackfill } from "../enqueueHistoricalBackfill";
+
+vi.mock("@/lib/research/playcounts/createSnapshot", () => ({ createSnapshot: vi.fn() }));
+vi.mock("../enqueueHistoricalBackfill", () => ({ enqueueHistoricalBackfill: vi.fn() }));
+
+const req = (source: "current" | "historical") => ({
+  accountId: "acc_1",
+  body: { scope: { album_ids: ["AL1"] }, source, platforms: ["spotify"] as ["spotify"] },
+});
+
+describe("createMeasurementJob", () => {
+  beforeEach(() => vi.clearAllMocks());
+
+  it("historical → delegates to enqueueHistoricalBackfill with the scope", async () => {
+    vi.mocked(enqueueHistoricalBackfill).mockResolvedValue({
+      data: { status: "success", source: "historical", id: null, enqueued: 5, skipped: 1 },
+    });
+    const r = await createMeasurementJob(req("historical"));
+    expect(enqueueHistoricalBackfill).toHaveBeenCalledWith({ album_ids: ["AL1"] });
+    expect(createSnapshot).not.toHaveBeenCalled();
+    expect(r).toEqual({
+      data: { status: "success", source: "historical", id: null, enqueued: 5, skipped: 1 },
+    });
+  });
+
+  it("current → reuses the snapshot pipeline and maps snapshot_id to the job id", async () => {
+    vi.mocked(createSnapshot).mockResolvedValue({
+      data: {
+        status: "success",
+        snapshot_id: "snap_9",
+        state: "queued",
+        album_count: 3,
+        estimated_cost_usd: 0.009,
+      },
+    });
+    const r = await createMeasurementJob(req("current"));
+    expect(createSnapshot).toHaveBeenCalledWith({
+      accountId: "acc_1",
+      body: { album_ids: ["AL1"], platforms: ["spotify"], schedule: "once" },
+    });
+    expect(r).toEqual({
+      data: {
+        status: "success",
+        source: "current",
+        id: "snap_9",
+        state: "queued",
+        album_count: 3,
+        estimated_cost_usd: 0.009,
+      },
+    });
+  });
+
+  it("current → propagates a snapshot error (e.g. 429 cap)", async () => {
+    vi.mocked(createSnapshot).mockResolvedValue({ error: "cap reached", status: 429 });
+    const r = await createMeasurementJob(req("current"));
+    expect(r).toEqual({ error: "cap reached", status: 429 });
+  });
+});

lib/research/measurement_jobs/__tests__/createMeasurementJobHandler.test.ts

@@ -0,0 +1,55 @@
+import { describe, it, expect, vi, beforeEach } from "vitest";
+import { NextRequest, NextResponse } from "next/server";
+import { createMeasurementJobHandler } from "../createMeasurementJobHandler";
+import { validateCreateMeasurementJobRequest } from "../validateCreateMeasurementJobRequest";
+import { createMeasurementJob } from "../createMeasurementJob";
+
+vi.mock("@/lib/networking/getCorsHeaders", () => ({ getCorsHeaders: vi.fn(() => ({})) }));
+vi.mock("../validateCreateMeasurementJobRequest", () => ({
+  validateCreateMeasurementJobRequest: vi.fn(),
+}));
+vi.mock("../createMeasurementJob", () => ({ createMeasurementJob: vi.fn() }));
+
+const req = () => new NextRequest("http://x/api/research/measurement-jobs", { method: "POST" });
+const validated = {
+  accountId: "acc_1",
+  body: { scope: { isrcs: ["X"] }, source: "historical", platforms: ["spotify"] },
+} as never;
+
+describe("createMeasurementJobHandler", () => {
+  beforeEach(() => vi.clearAllMocks());
+
+  it("returns the job payload with 202 on success", async () => {
+    vi.mocked(validateCreateMeasurementJobRequest).mockResolvedValue(validated);
+    vi.mocked(createMeasurementJob).mockResolvedValue({
+      data: { status: "success", source: "historical", id: null, enqueued: 5, skipped: 0 },
+    });
+    const res = await createMeasurementJobHandler(req());
+    expect(res.status).toBe(202);
+    expect(await res.json()).toMatchObject({ enqueued: 5, source: "historical" });
+  });
+
+  it("short-circuits with the validation response", async () => {
+    vi.mocked(validateCreateMeasurementJobRequest).mockResolvedValue(
+      NextResponse.json({ status: "error" }, { status: 400 }) as never,
+    );
+    const res = await createMeasurementJobHandler(req());
+    expect(res.status).toBe(400);
+    expect(createMeasurementJob).not.toHaveBeenCalled();
+  });
+
+  it("maps error results to error responses", async () => {
+    vi.mocked(validateCreateMeasurementJobRequest).mockResolvedValue(validated);
+    vi.mocked(createMeasurementJob).mockResolvedValue({ error: "cap reached", status: 429 });
+    const res = await createMeasurementJobHandler(req());
+    expect(res.status).toBe(429);
+  });
+
+  it("returns 500 on unexpected errors", async () => {
+    const consoleError = vi.spyOn(console, "error").mockImplementation(() => {});
+    vi.mocked(validateCreateMeasurementJobRequest).mockRejectedValue(new Error("boom"));
+    const res = await createMeasurementJobHandler(req());
+    expect(res.status).toBe(500);
+    consoleError.mockRestore();
+  });
+});

lib/research/measurement_jobs/__tests__/enqueueHistoricalBackfill.test.ts

@@ -0,0 +1,60 @@
+import { describe, it, expect, vi, beforeEach } from "vitest";
+import { enqueueHistoricalBackfill } from "../enqueueHistoricalBackfill";
+import { resolveScopeSongs } from "../resolveScopeSongs";
+import { selectSongMeasurements } from "@/lib/supabase/song_measurements/selectSongMeasurements";
+import { upsertSongstatsBackfillQueue } from "@/lib/supabase/songstats_backfill_queue/upsertSongstatsBackfillQueue";
+
+vi.mock("../resolveScopeSongs", () => ({ resolveScopeSongs: vi.fn() }));
+vi.mock("@/lib/supabase/song_measurements/selectSongMeasurements", () => ({
+  selectSongMeasurements: vi.fn(),
+}));
+vi.mock("@/lib/supabase/songstats_backfill_queue/upsertSongstatsBackfillQueue", () => ({
+  upsertSongstatsBackfillQueue: vi.fn(),
+}));
+
+describe("enqueueHistoricalBackfill", () => {
+  beforeEach(() => {
+    vi.clearAllMocks();
+    vi.mocked(upsertSongstatsBackfillQueue).mockResolvedValue(undefined);
+  });
+
+  it("400s when the scope resolves to no recordings", async () => {
+    vi.mocked(resolveScopeSongs).mockResolvedValue([]);
+    const r = await enqueueHistoricalBackfill({ isrcs: ["X"] });
+    expect(r).toEqual({
+      error: "No recordings resolvable from the given scope — no identifier mappings exist yet",
+      status: 400,
+    });
+  });
+
+  it("enqueues un-backfilled songs ranked by latest count, skips already-backfilled ones", async () => {
+    vi.mocked(resolveScopeSongs).mockResolvedValue(["I1", "I2", "I3"]);
+    // newest-first rows; I2 already has a songstats row (already backfilled)
+    vi.mocked(selectSongMeasurements).mockResolvedValue([
+      { song: "I1", value: 500, data_source: "apify_spotify_playcount" },
+      { song: "I2", value: 300, data_source: "songstats" },
+      { song: "I3", value: 100, data_source: "apify_spotify_playcount" },
+    ] as never);
+
+    const r = await enqueueHistoricalBackfill({ isrcs: ["I1", "I2", "I3"] });
+
+    expect(upsertSongstatsBackfillQueue).toHaveBeenCalledTimes(2);
+    expect(upsertSongstatsBackfillQueue).toHaveBeenCalledWith({ song: "I1", rank_score: 500 });
+    expect(upsertSongstatsBackfillQueue).toHaveBeenCalledWith({ song: "I3", rank_score: 100 });
+    expect(upsertSongstatsBackfillQueue).not.toHaveBeenCalledWith(
+      expect.objectContaining({ song: "I2" }),
+    );
+    expect(r).toEqual({
+      data: { status: "success", source: "historical", id: null, enqueued: 2, skipped: 1 },
+    });
+  });
+
+  it("ranks a song with no prior measurement at 0", async () => {
+    vi.mocked(resolveScopeSongs).mockResolvedValue(["I9"]);
+    vi.mocked(selectSongMeasurements).mockResolvedValue([] as never);
+
+    await enqueueHistoricalBackfill({ isrcs: ["I9"] });
+
+    expect(upsertSongstatsBackfillQueue).toHaveBeenCalledWith({ song: "I9", rank_score: 0 });
+  });
+});

lib/research/measurement_jobs/__tests__/ensureSongstatsPaymentMethod.test.ts

@@ -0,0 +1,54 @@
+import { describe, it, expect, vi, beforeEach } from "vitest";
+import { ensureSongstatsPaymentMethod } from "../ensureSongstatsPaymentMethod";
+import { findStripeCustomerForAccount } from "@/lib/stripe/findStripeCustomerForAccount";
+import { findDefaultPaymentMethodForCustomer } from "@/lib/stripe/findDefaultPaymentMethodForCustomer";
+import { createStripeSession } from "@/lib/stripe/createStripeSession";
+
+vi.mock("@/lib/networking/getCorsHeaders", () => ({ getCorsHeaders: vi.fn(() => ({})) }));
+vi.mock("@/lib/stripe/findStripeCustomerForAccount", () => ({
+  findStripeCustomerForAccount: vi.fn(),
+}));
+vi.mock("@/lib/stripe/findDefaultPaymentMethodForCustomer", () => ({
+  findDefaultPaymentMethodForCustomer: vi.fn(),
+}));
+vi.mock("@/lib/stripe/createStripeSession", () => ({ createStripeSession: vi.fn() }));
+
+describe("ensureSongstatsPaymentMethod", () => {
+  beforeEach(() => vi.clearAllMocks());
+
+  it("returns null (proceed) when the account has a card on file", async () => {
+    vi.mocked(findStripeCustomerForAccount).mockResolvedValue("cus_1");
+    vi.mocked(findDefaultPaymentMethodForCustomer).mockResolvedValue("pm_1");
+
+    const r = await ensureSongstatsPaymentMethod("acc_1");
+
+    expect(r).toBeNull();
+    expect(createStripeSession).not.toHaveBeenCalled();
+  });
+
+  it("402s with a free-tier checkout link when there is no Stripe customer", async () => {
+    vi.mocked(findStripeCustomerForAccount).mockResolvedValue(null);
+    vi.mocked(createStripeSession).mockResolvedValue({ url: "https://checkout/free" } as never);
+
+    const r = await ensureSongstatsPaymentMethod("acc_1");
+
+    expect(findDefaultPaymentMethodForCustomer).not.toHaveBeenCalled();
+    expect(createStripeSession).toHaveBeenCalledWith("acc_1", expect.any(String));
+    expect((r as Response).status).toBe(402);
+    expect(await (r as Response).json()).toMatchObject({
+      status: "error",
+      checkoutUrl: "https://checkout/free",
+    });
+  });
+
+  it("402s with a checkout link when the customer exists but has no card", async () => {
+    vi.mocked(findStripeCustomerForAccount).mockResolvedValue("cus_1");
+    vi.mocked(findDefaultPaymentMethodForCustomer).mockResolvedValue(null);
+    vi.mocked(createStripeSession).mockResolvedValue({ url: "https://checkout/free" } as never);
+
+    const r = await ensureSongstatsPaymentMethod("acc_1");
+
+    expect((r as Response).status).toBe(402);
+    expect(await (r as Response).json()).toMatchObject({ checkoutUrl: "https://checkout/free" });
+  });
+});

lib/research/measurement_jobs/__tests__/resolveScopeSongs.test.ts

@@ -0,0 +1,42 @@
+import { describe, it, expect, vi, beforeEach } from "vitest";
+import { resolveScopeSongs } from "../resolveScopeSongs";
+import { selectSongIdentifiers } from "@/lib/supabase/song_identifiers/selectSongIdentifiers";
+import { selectCatalogSongsWithArtists } from "@/lib/supabase/catalog_songs/selectCatalogSongsWithArtists";
+
+vi.mock("@/lib/supabase/song_identifiers/selectSongIdentifiers", () => ({
+  selectSongIdentifiers: vi.fn(),
+}));
+vi.mock("@/lib/supabase/catalog_songs/selectCatalogSongsWithArtists", () => ({
+  selectCatalogSongsWithArtists: vi.fn(),
+}));
+
+describe("resolveScopeSongs", () => {
+  beforeEach(() => vi.clearAllMocks());
+
+  it("returns deduped isrcs directly", async () => {
+    expect(await resolveScopeSongs({ isrcs: ["A", "A", "B"] })).toEqual(["A", "B"]);
+    expect(selectSongIdentifiers).not.toHaveBeenCalled();
+  });
+
+  it("resolves album_ids to their mapped song isrcs", async () => {
+    vi.mocked(selectSongIdentifiers).mockResolvedValue([
+      { song: "I1", platform: "spotify", identifier_type: "album_id", value: "AL1" },
+      { song: "I2", platform: "spotify", identifier_type: "album_id", value: "AL1" },
+    ]);
+    const out = await resolveScopeSongs({ album_ids: ["AL1"] });
+    expect(selectSongIdentifiers).toHaveBeenCalledWith({
+      platform: "spotify",
+      identifierType: "album_id",
+      values: ["AL1"],
+    });
+    expect(out).toEqual(["I1", "I2"]);
+  });
+
+  it("resolves a catalog_id to its song isrcs", async () => {
+    vi.mocked(selectCatalogSongsWithArtists).mockResolvedValue({
+      songs: [{ isrc: "I1" }, { isrc: "I2" }],
+    } as never);
+    const out = await resolveScopeSongs({ catalog_id: "cat-1" });
+    expect(out).toEqual(["I1", "I2"]);
+  });
+});