diff --git a/app/api/research/measurement-jobs/route.ts b/app/api/research/measurement-jobs/route.ts new file mode 100644 index 00000000..51866da6 --- /dev/null +++ b/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); +} diff --git a/app/workflows/__tests__/backfillTrackStep.test.ts b/app/workflows/__tests__/backfillTrackStep.test.ts index 90fc701e..ebe77370 100644 --- a/app/workflows/__tests__/backfillTrackStep.test.ts +++ b/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 }); + }); }); diff --git a/app/workflows/backfillTrackStep.ts b/app/workflows/backfillTrackStep.ts index bc250d9c..3247ff4c 100644 --- a/app/workflows/backfillTrackStep.ts +++ b/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 }; } diff --git a/lib/research/measurement_jobs/__tests__/createMeasurementJob.test.ts b/lib/research/measurement_jobs/__tests__/createMeasurementJob.test.ts new file mode 100644 index 00000000..88fd7bd8 --- /dev/null +++ b/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 }); + }); +}); diff --git a/lib/research/measurement_jobs/__tests__/createMeasurementJobHandler.test.ts b/lib/research/measurement_jobs/__tests__/createMeasurementJobHandler.test.ts new file mode 100644 index 00000000..18225479 --- /dev/null +++ b/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(); + }); +}); diff --git a/lib/research/measurement_jobs/__tests__/enqueueHistoricalBackfill.test.ts b/lib/research/measurement_jobs/__tests__/enqueueHistoricalBackfill.test.ts new file mode 100644 index 00000000..20cde74a --- /dev/null +++ b/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 }); + }); +}); diff --git a/lib/research/measurement_jobs/__tests__/ensureSongstatsPaymentMethod.test.ts b/lib/research/measurement_jobs/__tests__/ensureSongstatsPaymentMethod.test.ts new file mode 100644 index 00000000..1e2c1f07 --- /dev/null +++ b/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" }); + }); +}); diff --git a/lib/research/measurement_jobs/__tests__/resolveScopeSongs.test.ts b/lib/research/measurement_jobs/__tests__/resolveScopeSongs.test.ts new file mode 100644 index 00000000..7c0bc8d9 --- /dev/null +++ b/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"]); + }); +}); diff --git a/lib/research/measurement_jobs/__tests__/validateCreateMeasurementJobRequest.test.ts b/lib/research/measurement_jobs/__tests__/validateCreateMeasurementJobRequest.test.ts new file mode 100644 index 00000000..e36042a7 --- /dev/null +++ b/lib/research/measurement_jobs/__tests__/validateCreateMeasurementJobRequest.test.ts @@ -0,0 +1,88 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; +import { NextRequest, NextResponse } from "next/server"; +import { validateCreateMeasurementJobRequest } from "../validateCreateMeasurementJobRequest"; +import { validateAuthContext } from "@/lib/auth/validateAuthContext"; +import { ensureSongstatsPaymentMethod } from "../ensureSongstatsPaymentMethod"; + +vi.mock("@/lib/auth/validateAuthContext", () => ({ validateAuthContext: vi.fn() })); +vi.mock("../ensureSongstatsPaymentMethod", () => ({ ensureSongstatsPaymentMethod: vi.fn() })); + +const post = (body: unknown) => + new NextRequest("http://x/api/research/measurement-jobs", { + method: "POST", + body: JSON.stringify(body), + headers: { "content-type": "application/json" }, + }); + +describe("validateCreateMeasurementJobRequest", () => { + beforeEach(() => { + vi.clearAllMocks(); + vi.mocked(validateAuthContext).mockResolvedValue({ accountId: "acc_1" } as never); + vi.mocked(ensureSongstatsPaymentMethod).mockResolvedValue(null); // card on file by default + }); + + it("returns the auth response (401) when auth fails", async () => { + vi.mocked(validateAuthContext).mockResolvedValue( + NextResponse.json({ status: "error" }, { status: 401 }) as never, + ); + const r = await validateCreateMeasurementJobRequest( + post({ scope: { isrcs: ["X"] }, source: "historical" }), + ); + expect((r as NextResponse).status).toBe(401); + }); + + it("400 when source is missing", async () => { + const r = await validateCreateMeasurementJobRequest(post({ scope: { isrcs: ["X"] } })); + expect((r as NextResponse).status).toBe(400); + }); + + it("400 when source is not current|historical", async () => { + const r = await validateCreateMeasurementJobRequest( + post({ scope: { isrcs: ["X"] }, source: "future" }), + ); + expect((r as NextResponse).status).toBe(400); + }); + + it("400 when scope has zero keys", async () => { + const r = await validateCreateMeasurementJobRequest(post({ scope: {}, source: "current" })); + expect((r as NextResponse).status).toBe(400); + }); + + it("400 when scope has more than one key", async () => { + const r = await validateCreateMeasurementJobRequest( + post({ scope: { isrcs: ["X"], album_ids: ["A"] }, source: "current" }), + ); + expect((r as NextResponse).status).toBe(400); + }); + + it("passes auth + body through on success, defaulting platforms to spotify", async () => { + const r = await validateCreateMeasurementJobRequest( + post({ scope: { album_ids: ["A1"] }, source: "historical" }), + ); + expect(r).toEqual({ + accountId: "acc_1", + body: { scope: { album_ids: ["A1"] }, source: "historical", platforms: ["spotify"] }, + }); + }); + + it("historical: short-circuits with the 402 when no card is on file", async () => { + vi.mocked(ensureSongstatsPaymentMethod).mockResolvedValue( + NextResponse.json( + { status: "error", checkoutUrl: "https://checkout" }, + { status: 402 }, + ) as never, + ); + const r = await validateCreateMeasurementJobRequest( + post({ scope: { isrcs: ["X"] }, source: "historical" }), + ); + expect((r as NextResponse).status).toBe(402); + }); + + it("current: does NOT require a card (Apify-only, exempt from the Songstats gate)", async () => { + const r = await validateCreateMeasurementJobRequest( + post({ scope: { album_ids: ["A1"] }, source: "current" }), + ); + expect(ensureSongstatsPaymentMethod).not.toHaveBeenCalled(); + expect(r).toMatchObject({ accountId: "acc_1", body: { source: "current" } }); + }); +}); diff --git a/lib/research/measurement_jobs/createMeasurementJob.ts b/lib/research/measurement_jobs/createMeasurementJob.ts new file mode 100644 index 00000000..adb383e0 --- /dev/null +++ b/lib/research/measurement_jobs/createMeasurementJob.ts @@ -0,0 +1,47 @@ +import { createSnapshot } from "@/lib/research/playcounts/createSnapshot"; +import { enqueueHistoricalBackfill } from "./enqueueHistoricalBackfill"; +import type { ValidatedCreateMeasurementJobRequest } from "./validateCreateMeasurementJobRequest"; + +export type CreateMeasurementJobResult = { data: unknown } | { error: string; status: number }; + +type SnapshotData = { + snapshot_id: string; + state: string; + album_count: number; + estimated_cost_usd: number; +}; + +/** + * Dispatch a measurement job by `source`. `historical` enqueues Songstats deep + * backfill; `current` reuses the snapshot capture pipeline (replacing + * `POST /api/research/snapshots`) and maps `snapshot_id` to the resource's `id`. + * + * @param req - The validated create request (accountId + body) + */ +export async function createMeasurementJob( + req: ValidatedCreateMeasurementJobRequest, +): Promise { + const { accountId, body } = req; + + if (body.source === "historical") { + return enqueueHistoricalBackfill(body.scope); + } + + const result = await createSnapshot({ + accountId, + body: { ...body.scope, platforms: body.platforms, schedule: "once" }, + }); + if ("error" in result) return result; + + const d = result.data as SnapshotData; + return { + data: { + status: "success", + source: "current", + id: d.snapshot_id, + state: d.state, + album_count: d.album_count, + estimated_cost_usd: d.estimated_cost_usd, + }, + }; +} diff --git a/lib/research/measurement_jobs/createMeasurementJobHandler.ts b/lib/research/measurement_jobs/createMeasurementJobHandler.ts new file mode 100644 index 00000000..663e02dd --- /dev/null +++ b/lib/research/measurement_jobs/createMeasurementJobHandler.ts @@ -0,0 +1,31 @@ +import { type NextRequest, NextResponse } from "next/server"; +import { errorResponse } from "@/lib/networking/errorResponse"; +import { getCorsHeaders } from "@/lib/networking/getCorsHeaders"; +import { createMeasurementJob } from "@/lib/research/measurement_jobs/createMeasurementJob"; +import { validateCreateMeasurementJobRequest } from "@/lib/research/measurement_jobs/validateCreateMeasurementJobRequest"; + +/** + * POST /api/research/measurement-jobs + * + * One async ingest resource. `source:"current"` captures present counts via the + * snapshot pipeline (replaces `POST /api/research/snapshots`); `source:"historical"` + * enqueues Songstats deep backfill ranked by all-time streams (idempotent — + * already-backfilled songs are skipped). Returns 202. + * + * @param request - The incoming HTTP request. + * @returns The JSON response. + */ +export async function createMeasurementJobHandler(request: NextRequest): Promise { + try { + const validated = await validateCreateMeasurementJobRequest(request); + if (validated instanceof NextResponse) return validated; + + const result = await createMeasurementJob(validated); + if ("error" in result) return errorResponse(result.error, result.status); + + return NextResponse.json(result.data, { status: 202, headers: getCorsHeaders() }); + } catch (error) { + console.error("[ERROR] createMeasurementJobHandler:", error); + return errorResponse("Internal error", 500); + } +} diff --git a/lib/research/measurement_jobs/enqueueHistoricalBackfill.ts b/lib/research/measurement_jobs/enqueueHistoricalBackfill.ts new file mode 100644 index 00000000..3c0cc7af --- /dev/null +++ b/lib/research/measurement_jobs/enqueueHistoricalBackfill.ts @@ -0,0 +1,65 @@ +import { resolveScopeSongs } from "./resolveScopeSongs"; +import { selectSongMeasurements } from "@/lib/supabase/song_measurements/selectSongMeasurements"; +import { upsertSongstatsBackfillQueue } from "@/lib/supabase/songstats_backfill_queue/upsertSongstatsBackfillQueue"; +import type { CreateMeasurementJobBody } from "./validateCreateMeasurementJobRequest"; + +const METRIC = "platform_displayed_play_count"; +/** Max concurrent queue upserts per batch (bounds round trips on large catalogs). */ +const ENQUEUE_CONCURRENCY = 25; + +export type EnqueueHistoricalBackfillResult = { data: unknown } | { error: string; status: number }; + +/** + * `source: "historical"` path of a measurement job. Resolves the scope to song + * ISRCs, enqueues each for Songstats deep backfill ranked by its latest known + * count, and skips songs that already carry `songstats` history (so no track is + * fetched from Songstats twice). Free — no credit deduction. The queue drains + * via the daily maintenance worker. + * + * @param scope - The job scope (catalog_id / album_ids / isrcs) + * @returns 202 payload with `enqueued`/`skipped`, or a 400 when nothing resolves + */ +export async function enqueueHistoricalBackfill( + scope: CreateMeasurementJobBody["scope"], +): Promise { + const isrcs = await resolveScopeSongs(scope); + if (isrcs.length === 0) { + return { + error: "No recordings resolvable from the given scope — no identifier mappings exist yet", + status: 400, + }; + } + + const measurements = await selectSongMeasurements({ + songs: isrcs, + platform: "spotify", + metric: METRIC, + }); + + // rows are newest-first: the first row per song is its latest count (the rank); + // a song is already backfilled if any of its rows is `songstats`-sourced. + const latestValue = new Map(); + const alreadyBackfilled = new Set(); + for (const row of measurements) { + if (!latestValue.has(row.song)) latestValue.set(row.song, Number(row.value) || 0); + if (row.data_source === "songstats") alreadyBackfilled.add(row.song); + } + + const candidates = isrcs.filter(isrc => !alreadyBackfilled.has(isrc)); + const skipped = isrcs.length - candidates.length; + + // Enqueue in bounded-concurrency batches so a large catalog doesn't fan out + // into N serial round trips (which would blow the route's duration budget). + let enqueued = 0; + for (let i = 0; i < candidates.length; i += ENQUEUE_CONCURRENCY) { + const batch = candidates.slice(i, i + ENQUEUE_CONCURRENCY); + await Promise.all( + batch.map(isrc => + upsertSongstatsBackfillQueue({ song: isrc, rank_score: latestValue.get(isrc) ?? 0 }), + ), + ); + enqueued += batch.length; + } + + return { data: { status: "success", source: "historical", id: null, enqueued, skipped } }; +} diff --git a/lib/research/measurement_jobs/ensureSongstatsPaymentMethod.ts b/lib/research/measurement_jobs/ensureSongstatsPaymentMethod.ts new file mode 100644 index 00000000..71e3dd37 --- /dev/null +++ b/lib/research/measurement_jobs/ensureSongstatsPaymentMethod.ts @@ -0,0 +1,35 @@ +import { NextResponse } from "next/server"; +import { getCorsHeaders } from "@/lib/networking/getCorsHeaders"; +import { findStripeCustomerForAccount } from "@/lib/stripe/findStripeCustomerForAccount"; +import { findDefaultPaymentMethodForCustomer } from "@/lib/stripe/findDefaultPaymentMethodForCustomer"; +import { createStripeSession } from "@/lib/stripe/createStripeSession"; +import { CREDIT_AUTO_RECHARGE_FALLBACK_SUCCESS_URL } from "@/lib/credits/const"; + +/** + * Payment-method gate for Songstats-backed work (the heavily quota-capped + * provider). The authenticated account must have a card on file before it can + * spend Songstats quota. Returns `null` to proceed; otherwise a **402** carrying + * a Stripe Checkout URL for the free tier (subscription + trial) so the caller + * can add a card. Mirrors the credit gate's short-circuit shape. + * + * @param accountId - The authenticated account. + * @returns `null` when a card exists, else a 402 NextResponse with `checkoutUrl`. + */ +export async function ensureSongstatsPaymentMethod( + accountId: string, +): Promise { + const customerId = await findStripeCustomerForAccount(accountId); + const paymentMethod = customerId ? await findDefaultPaymentMethodForCustomer(customerId) : null; + if (paymentMethod) return null; + + const session = await createStripeSession(accountId, CREDIT_AUTO_RECHARGE_FALLBACK_SUCCESS_URL); + return NextResponse.json( + { + status: "error", + error: + "A payment method is required to use Songstats-backed endpoints. Add a card to continue.", + checkoutUrl: session.url, + }, + { status: 402, headers: getCorsHeaders() }, + ); +} diff --git a/lib/research/measurement_jobs/resolveScopeSongs.ts b/lib/research/measurement_jobs/resolveScopeSongs.ts new file mode 100644 index 00000000..9da9e5ab --- /dev/null +++ b/lib/research/measurement_jobs/resolveScopeSongs.ts @@ -0,0 +1,35 @@ +import { selectSongIdentifiers } from "@/lib/supabase/song_identifiers/selectSongIdentifiers"; +import { selectCatalogSongsWithArtists } from "@/lib/supabase/catalog_songs/selectCatalogSongsWithArtists"; +import type { CreateMeasurementJobBody } from "./validateCreateMeasurementJobRequest"; + +/** + * Resolve a measurement-job `scope` to a deduped list of song ISRCs. + * `isrcs` pass through; `album_ids` resolve through the album_id identifier + * mappings; `catalog_id` resolves through its songs. Mirrors + * {@link resolveSnapshotAlbums} but returns songs (the backfill queue is keyed + * by ISRC). + * + * @param scope - Exactly one of catalog_id / album_ids / isrcs + * @returns Unique song ISRCs ([] when nothing is mapped yet) + */ +export async function resolveScopeSongs( + scope: CreateMeasurementJobBody["scope"], +): Promise { + if (scope.isrcs) return [...new Set(scope.isrcs)]; + + if (scope.album_ids) { + const rows = await selectSongIdentifiers({ + platform: "spotify", + identifierType: "album_id", + values: scope.album_ids, + }); + return [...new Set(rows.map(r => r.song))]; + } + + if (scope.catalog_id) { + const { songs } = await selectCatalogSongsWithArtists({ catalogId: scope.catalog_id }); + return [...new Set(songs.map(s => s.isrc))]; + } + + return []; +} diff --git a/lib/research/measurement_jobs/validateCreateMeasurementJobRequest.ts b/lib/research/measurement_jobs/validateCreateMeasurementJobRequest.ts new file mode 100644 index 00000000..8218fa5f --- /dev/null +++ b/lib/research/measurement_jobs/validateCreateMeasurementJobRequest.ts @@ -0,0 +1,57 @@ +import { type NextRequest, NextResponse } from "next/server"; +import { z } from "zod"; +import { validateAuthContext } from "@/lib/auth/validateAuthContext"; +import { errorResponse } from "@/lib/networking/errorResponse"; +import { ensureSongstatsPaymentMethod } from "@/lib/research/measurement_jobs/ensureSongstatsPaymentMethod"; + +const scopeSchema = z + .object({ + catalog_id: z.string().uuid("scope.catalog_id must be a valid UUID").optional(), + album_ids: z.array(z.string()).min(1).optional(), + isrcs: z.array(z.string()).min(1).optional(), + }) + .refine(s => [s.catalog_id, s.album_ids, s.isrcs].filter(Boolean).length === 1, { + message: "Provide exactly one of scope.catalog_id, scope.album_ids, scope.isrcs", + }); + +export const createMeasurementJobBodySchema = z.object({ + scope: scopeSchema, + source: z.enum(["current", "historical"], { message: "source must be current or historical" }), + platforms: z.array(z.enum(["spotify"])).default(["spotify"]), +}); + +export type CreateMeasurementJobBody = z.infer; + +export type ValidatedCreateMeasurementJobRequest = { + accountId: string; + body: CreateMeasurementJobBody; +}; + +/** + * Validates `POST /api/research/measurement-jobs` — auth + body: a `scope` + * (exactly one of `catalog_id` / `album_ids` / `isrcs`) and a `source` + * (`current` = Apify capture, `historical` = Songstats backfill). + * + * @param request - The incoming HTTP request. + */ +export async function validateCreateMeasurementJobRequest( + request: NextRequest, +): Promise { + const authResult = await validateAuthContext(request); + if (authResult instanceof NextResponse) return authResult; + + const raw = await request.json().catch(() => null); + const result = createMeasurementJobBodySchema.safeParse(raw); + if (!result.success) { + return errorResponse(result.error.issues[0].message, 400); + } + + // `historical` spends Songstats quota → require a card on file (402 + free-tier + // checkout link if absent). `current` is Apify-only, so it's exempt. + if (result.data.source === "historical") { + const short = await ensureSongstatsPaymentMethod(authResult.accountId); + if (short) return short; + } + + return { accountId: authResult.accountId, body: result.data }; +} diff --git a/lib/research/playcounts/__tests__/playcountMaintenanceHandler.test.ts b/lib/research/playcounts/__tests__/playcountMaintenanceHandler.test.ts new file mode 100644 index 00000000..e0b7ad9a --- /dev/null +++ b/lib/research/playcounts/__tests__/playcountMaintenanceHandler.test.ts @@ -0,0 +1,62 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; +import { NextRequest, NextResponse } from "next/server"; +import { playcountMaintenanceHandler } from "../playcountMaintenanceHandler"; +import { validateCronRequest } from "@/lib/internal/validateCronRequest"; +import { start } from "workflow/api"; +import { startDueMonthlySnapshots } from "../startDueMonthlySnapshots"; +import { reclaimStaleSongstatsBackfillRows } from "@/lib/supabase/songstats_backfill_queue/updateSongstatsBackfillQueue"; + +vi.mock("@/lib/networking/getCorsHeaders", () => ({ getCorsHeaders: vi.fn(() => ({})) })); +vi.mock("@/lib/internal/validateCronRequest", () => ({ validateCronRequest: vi.fn() })); +vi.mock("workflow/api", () => ({ start: vi.fn() })); +vi.mock("@/app/workflows/songstatsBackfillWorkflow", () => ({ + songstatsBackfillWorkflow: vi.fn(), +})); +vi.mock("../startDueMonthlySnapshots", () => ({ startDueMonthlySnapshots: vi.fn() })); +vi.mock("@/lib/supabase/songstats_backfill_queue/updateSongstatsBackfillQueue", () => ({ + reclaimStaleSongstatsBackfillRows: vi.fn(), +})); + +const req = () => new NextRequest("http://x/api/internal/playcount-maintenance"); + +describe("playcountMaintenanceHandler", () => { + beforeEach(() => { + vi.clearAllMocks(); + vi.mocked(validateCronRequest).mockReturnValue(null as never); + vi.mocked(start).mockResolvedValue({ runId: "run_1" } as never); + vi.mocked(startDueMonthlySnapshots).mockResolvedValue(2 as never); + vi.mocked(reclaimStaleSongstatsBackfillRows).mockResolvedValue(5); + }); + + it("reclaims stale rows BEFORE starting the drain, and reports the count", async () => { + const order: string[] = []; + vi.mocked(reclaimStaleSongstatsBackfillRows).mockImplementation(async () => { + order.push("reclaim"); + return 5; + }); + vi.mocked(start).mockImplementation(async () => { + order.push("drain"); + return { runId: "run_1" } as never; + }); + + const res = await playcountMaintenanceHandler(req()); + + expect(order).toEqual(["reclaim", "drain"]); + expect(res.status).toBe(202); + expect(await res.json()).toMatchObject({ + status: "success", + reclaimed: 5, + backfill_run_id: "run_1", + monthly_snapshots_started: 2, + }); + }); + + it("denies non-cron requests", async () => { + vi.mocked(validateCronRequest).mockReturnValue( + NextResponse.json({ status: "error" }, { status: 401 }) as never, + ); + const res = await playcountMaintenanceHandler(req()); + expect(res.status).toBe(401); + expect(reclaimStaleSongstatsBackfillRows).not.toHaveBeenCalled(); + }); +}); diff --git a/lib/research/playcounts/playcountMaintenanceHandler.ts b/lib/research/playcounts/playcountMaintenanceHandler.ts index facba435..092420ec 100644 --- a/lib/research/playcounts/playcountMaintenanceHandler.ts +++ b/lib/research/playcounts/playcountMaintenanceHandler.ts @@ -3,6 +3,7 @@ import { start } from "workflow/api"; import { validateCronRequest } from "@/lib/internal/validateCronRequest"; import { songstatsBackfillWorkflow } from "@/app/workflows/songstatsBackfillWorkflow"; import { startDueMonthlySnapshots } from "@/lib/research/playcounts/startDueMonthlySnapshots"; +import { reclaimStaleSongstatsBackfillRows } from "@/lib/supabase/songstats_backfill_queue/updateSongstatsBackfillQueue"; import { getCorsHeaders } from "@/lib/networking/getCorsHeaders"; /** @@ -18,11 +19,19 @@ export async function playcountMaintenanceHandler(request: NextRequest): Promise if (denied) return denied; try { + // Return transiently-failed / orphaned rows to `pending` before draining, so + // tracks that 429'd in a prior run get retried instead of being stranded. + const reclaimed = await reclaimStaleSongstatsBackfillRows(); const run = await start(songstatsBackfillWorkflow); const monthlyStarted = await startDueMonthlySnapshots(); return NextResponse.json( - { status: "success", backfill_run_id: run.runId, monthly_snapshots_started: monthlyStarted }, + { + status: "success", + reclaimed, + backfill_run_id: run.runId, + monthly_snapshots_started: monthlyStarted, + }, { status: 202, headers: getCorsHeaders() }, ); } catch (error) { diff --git a/lib/supabase/songstats_backfill_queue/__tests__/updateSongstatsBackfillQueue.test.ts b/lib/supabase/songstats_backfill_queue/__tests__/updateSongstatsBackfillQueue.test.ts index f9321530..eb8d0230 100644 --- a/lib/supabase/songstats_backfill_queue/__tests__/updateSongstatsBackfillQueue.test.ts +++ b/lib/supabase/songstats_backfill_queue/__tests__/updateSongstatsBackfillQueue.test.ts @@ -1,5 +1,8 @@ import { describe, it, expect, vi, beforeEach } from "vitest"; -import { updateSongstatsBackfillQueue } from "../updateSongstatsBackfillQueue"; +import { + updateSongstatsBackfillQueue, + reclaimStaleSongstatsBackfillRows, +} from "../updateSongstatsBackfillQueue"; import supabase from "../../serverClient"; vi.mock("../../serverClient", () => { @@ -33,3 +36,34 @@ describe("updateSongstatsBackfillQueue", () => { ); }); }); + +describe("reclaimStaleSongstatsBackfillRows", () => { + beforeEach(() => vi.clearAllMocks()); + + it("resets failed + AND-grouped stale in_progress rows to pending and returns the count", async () => { + const select = vi.fn().mockResolvedValue({ data: [{ id: "a" }, { id: "b" }], error: null }); + const or = vi.fn().mockReturnValue({ select }); + const update = vi.fn().mockReturnValue({ or }); + vi.mocked(supabase.from).mockReturnValue({ update } as never); + + const count = await reclaimStaleSongstatsBackfillRows(); + + expect(update).toHaveBeenCalledWith({ status: "pending" }); + // The and() grouping around in_progress + updated_at is load-bearing — it is + // what prevents reclaiming `done`/`pending` rows. Assert the exact structure, + // not just substrings. + const orArg = or.mock.calls[0][0] as string; + expect(orArg).toMatch(/^status\.eq\.failed,and\(status\.eq\.in_progress,updated_at\.lt\..+\)$/); + expect(count).toBe(2); + }); + + it("throws on a reclaim DB error instead of masking it as 0", async () => { + const select = vi.fn().mockResolvedValue({ data: null, error: { message: "boom" } }); + const or = vi.fn().mockReturnValue({ select }); + vi.mocked(supabase.from).mockReturnValue({ update: vi.fn().mockReturnValue({ or }) } as never); + + await expect(reclaimStaleSongstatsBackfillRows()).rejects.toThrow( + "Failed to reclaim stale songstats backfill rows: boom", + ); + }); +}); diff --git a/lib/supabase/songstats_backfill_queue/updateSongstatsBackfillQueue.ts b/lib/supabase/songstats_backfill_queue/updateSongstatsBackfillQueue.ts index df3c8a16..b462381a 100644 --- a/lib/supabase/songstats_backfill_queue/updateSongstatsBackfillQueue.ts +++ b/lib/supabase/songstats_backfill_queue/updateSongstatsBackfillQueue.ts @@ -18,3 +18,33 @@ export async function updateSongstatsBackfillQueue( throw new Error(`Failed to update songstats backfill queue: ${error.message}`); } } + +/** In-progress rows older than this are treated as orphaned (crashed mid-claim). */ +const STALE_IN_PROGRESS_MS = 60 * 60 * 1000; // 1 hour + +/** + * Reset reclaimable rows to `pending` so the next drain retries them: every + * `failed` row (transient upstream errors — `backfillTrackStep` reserves + * `failed` for retryable 408/429/5xx; terminal no-data / permanent 4xx is + * marked `done`) plus any `in_progress` row orphaned by a crashed run. + * Idempotent — safe before every drain. Throws on a DB error so the caller can + * fail loudly rather than silently skip the reclaim. + * + * @returns Number of rows returned to `pending`. + * @throws Error if the update fails + */ +export async function reclaimStaleSongstatsBackfillRows(): Promise { + const staleBefore = new Date(Date.now() - STALE_IN_PROGRESS_MS).toISOString(); + + const { data, error } = await supabase + .from("songstats_backfill_queue") + .update({ status: "pending" }) + .or(`status.eq.failed,and(status.eq.in_progress,updated_at.lt.${staleBefore})`) + .select("id"); + + if (error) { + throw new Error(`Failed to reclaim stale songstats backfill rows: ${error.message}`); + } + + return data?.length ?? 0; +}