From 8c8a34e320dc456e5c716ae2acbceaef7b0b1158 Mon Sep 17 00:00:00 2001 From: Jonathan Date: Tue, 14 Jul 2026 21:28:54 +0200 Subject: [PATCH] feat(library): replace/upgrade an owned album on demand MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add Request.force (migration add_request_force): the pipeline's intake now skips the "already in library" dedupe for a forced job, so an owned album is re-downloaded. The import step still keeps the new copy only when it's higher quality, so a forced upgrade can never downgrade what's on disk. - Worker: _is_force_job → dedupe bypassed when Request.force. - POST /api/library/[id]/upgrade finds the matching MonitoredRelease and enqueues a force request (guards against stacking on an in-flight job). - Library route exposes monitoredReleaseId; the album modal shows a "Replace / upgrade" button when a release exists. Worker + web tests added (pipeline force re-acquire; upgrade route). Co-Authored-By: Claude Opus 4.8 (1M context) --- .../migration.sql | 2 + web/prisma/schema.prisma | 4 ++ .../api/library/[id]/upgrade/route.test.ts | 54 ++++++++++++++++++ web/src/app/api/library/[id]/upgrade/route.ts | 41 +++++++++++++ web/src/app/api/library/route.ts | Bin 1507 -> 1561 bytes web/src/app/library/library-client.tsx | 18 ++++++ worker/lyra_worker/pipeline.py | 20 ++++++- worker/tests/conftest.py | 8 +-- worker/tests/test_pipeline.py | 19 ++++++ 9 files changed, 160 insertions(+), 6 deletions(-) create mode 100644 web/prisma/migrations/20260714192517_add_request_force/migration.sql create mode 100644 web/src/app/api/library/[id]/upgrade/route.test.ts create mode 100644 web/src/app/api/library/[id]/upgrade/route.ts diff --git a/web/prisma/migrations/20260714192517_add_request_force/migration.sql b/web/prisma/migrations/20260714192517_add_request_force/migration.sql new file mode 100644 index 0000000..4e94543 --- /dev/null +++ b/web/prisma/migrations/20260714192517_add_request_force/migration.sql @@ -0,0 +1,2 @@ +-- AlterTable +ALTER TABLE "Request" ADD COLUMN "force" BOOLEAN NOT NULL DEFAULT false; diff --git a/web/prisma/schema.prisma b/web/prisma/schema.prisma index 23e6e5c..7b16ea8 100644 --- a/web/prisma/schema.prisma +++ b/web/prisma/schema.prisma @@ -35,6 +35,10 @@ model Request { album String status RequestStatus @default(pending) createdAt DateTime @default(now()) + // Force re-acquisition: skip the "already in library" dedupe so an owned album is + // re-downloaded on demand. The import step still keeps the copy only if it's higher + // quality, so a forced upgrade never downgrades. Set by the Library "Replace / upgrade". + force Boolean @default(false) job Job? libraryItem LibraryItem? diff --git a/web/src/app/api/library/[id]/upgrade/route.test.ts b/web/src/app/api/library/[id]/upgrade/route.test.ts new file mode 100644 index 0000000..fc9fa28 --- /dev/null +++ b/web/src/app/api/library/[id]/upgrade/route.test.ts @@ -0,0 +1,54 @@ +import { describe, it, expect } from "vitest"; +import { prisma } from "@/lib/db"; +import { POST } from "./route"; + +const ctx = (id: string) => ({ params: Promise.resolve({ id }) }); + +async function owned(artist: string, album: string, rgMbid: string | null) { + return prisma.libraryItem.create({ + data: { artist, album, path: "/m", source: "scan", format: "FLAC", qualityClass: 2, rgMbid }, + }); +} + +describe("POST /api/library/[id]/upgrade", () => { + it("enqueues a forced re-acquire request against the matching release", async () => { + await prisma.monitoredRelease.create({ + data: { artistMbid: "a1", artistName: "Adele", rgMbid: "rg-25", album: "25", secondaryTypes: [], + monitored: true, state: "fulfilled", currentQualityClass: 2 }, + }); + const item = await owned("Adele", "25", "rg-25"); + + const res = await POST(new Request("http://x", { method: "POST" }), ctx(item.id)); + expect(res.status).toBe(202); + expect((await res.json()).enqueued).toBe(true); + + const reqRow = await prisma.request.findFirst({ where: { album: "25" }, include: { job: true } }); + expect(reqRow?.force).toBe(true); + expect(reqRow?.monitoredReleaseId).toBeTruthy(); + expect(reqRow?.job).toBeTruthy(); + }); + + it("400s when there is no matching release", async () => { + const item = await owned("Nobody", "Untracked", null); + const res = await POST(new Request("http://x", { method: "POST" }), ctx(item.id)); + expect(res.status).toBe(400); + }); + + it("does not stack a second job while one is in flight", async () => { + const mr = await prisma.monitoredRelease.create({ + data: { artistMbid: "a1", artistName: "Adele", rgMbid: "rg-21", album: "21", secondaryTypes: [], monitored: true, state: "wanted" }, + }); + const item = await owned("Adele", "21", "rg-21"); + await prisma.request.create({ + data: { artist: "Adele", album: "21", monitoredReleaseId: mr.id, job: { create: { state: "downloading" } } }, + }); + const res = await POST(new Request("http://x", { method: "POST" }), ctx(item.id)); + expect(res.status).toBe(202); + expect((await res.json()).enqueued).toBe(false); + }); + + it("404s an unknown library item", async () => { + const res = await POST(new Request("http://x", { method: "POST" }), ctx("nope")); + expect(res.status).toBe(404); + }); +}); diff --git a/web/src/app/api/library/[id]/upgrade/route.ts b/web/src/app/api/library/[id]/upgrade/route.ts new file mode 100644 index 0000000..32bad36 --- /dev/null +++ b/web/src/app/api/library/[id]/upgrade/route.ts @@ -0,0 +1,41 @@ +import { prisma } from "@/lib/db"; + +// POST /api/library/[id]/upgrade — force a re-acquire of an owned album: enqueue a job that +// skips the "already in library" dedupe (Request.force). The pipeline downloads the best +// available copy and the import step keeps it only if it's higher quality, so this can upgrade +// but never downgrades. Requires a matching MonitoredRelease (scanned/monitored albums have +// one) to carry the acquisition context. +export async function POST(_request: Request, { params }: { params: Promise<{ id: string }> }) { + const { id } = await params; + const item = await prisma.libraryItem.findUnique({ where: { id } }); + if (!item) return Response.json({ error: "not found" }, { status: 404 }); + + // Prefer the release-group match; fall back to an artist+album match. + const release = + (item.rgMbid ? await prisma.monitoredRelease.findUnique({ where: { rgMbid: item.rgMbid } }) : null) ?? + (await prisma.monitoredRelease.findFirst({ where: { artistName: item.artist, album: item.album } })); + if (!release) { + return Response.json( + { error: "no release info for this album — follow the artist or re-scan to enable upgrades" }, + { status: 400 }, + ); + } + + // Don't stack a second job on a release that's already in flight. + const activeJob = await prisma.job.findFirst({ + where: { request: { monitoredReleaseId: release.id }, state: { notIn: ["imported", "needs_attention"] } }, + }); + if (activeJob) return Response.json({ enqueued: false }, { status: 202 }); + + await prisma.request.create({ + data: { + artist: release.artistName, + album: release.album, + monitoredReleaseId: release.id, + force: true, + job: { create: {} }, + }, + }); + await prisma.monitoredRelease.update({ where: { id: release.id }, data: { lastSearchedAt: new Date() } }); + return Response.json({ enqueued: true }, { status: 202 }); +} diff --git a/web/src/app/api/library/route.ts b/web/src/app/api/library/route.ts index cc70ef6297085e66be5fa34d44c28fb87dc27137..9d5711bd65ca7d968ffc81bc45daa8a227232a03 100644 GIT binary patch delta 51 zcmaFNJ(Fj{T}IB#6f1?2qS92I&370tGfU*==Vg}U7p0~IrRJn27N>dwrE`ny^(N0| G^#A|?>JuCQ delta 16 YcmbQq^O$?XUB=DN880zUe#&YB06^3SLjV8( diff --git a/web/src/app/library/library-client.tsx b/web/src/app/library/library-client.tsx index 5e93ad9..82696f0 100644 --- a/web/src/app/library/library-client.tsx +++ b/web/src/app/library/library-client.tsx @@ -19,6 +19,7 @@ type Album = { year: string | null; primaryType: string | null; artistMbid: string | null; + monitoredReleaseId: string | null; trackNames: string[]; }; @@ -115,6 +116,18 @@ export function LibraryClient() { refresh(); }, []); + async function upgrade() { + if (!open) return; + const res = await fetch(`/api/library/${open.id}/upgrade`, { method: "POST" }).catch(() => null); + if (res && (res.status === 202 || res.ok)) { + const body = await res.json().catch(() => null); + toast(body?.enqueued === false ? `Already searching for ${open.album}` : `Re-acquiring ${open.album} — keeps it only if better`); + show(null); + } else { + toast((res && (await res.json().catch(() => null))?.error) || "Couldn't start the upgrade"); + } + } + async function checkIntegrity() { setChecking(true); try { @@ -349,6 +362,11 @@ export function LibraryClient() { ) : ( <> + {open.monitoredReleaseId ? ( + + ) : null} diff --git a/worker/lyra_worker/pipeline.py b/worker/lyra_worker/pipeline.py index 8bc4ea6..709dc9e 100644 --- a/worker/lyra_worker/pipeline.py +++ b/worker/lyra_worker/pipeline.py @@ -146,6 +146,19 @@ def _is_upgrade_job(conn: psycopg.Connection, job_id: str) -> bool: return bool(row and row[0]) +def _is_force_job(conn: psycopg.Connection, job_id: str) -> bool: + """A user-forced re-acquire (Library 'Replace / upgrade'): skip the already-in-library + dedupe so an owned album is re-downloaded. The import step still keeps the new copy only + if it's higher quality, so a forced upgrade can never downgrade what's on disk.""" + with conn.cursor() as cur: + cur.execute( + 'SELECT force FROM "Request" WHERE id = (SELECT "requestId" FROM "Job" WHERE id = %s)', + (job_id,), + ) + row = cur.fetchone() + return bool(row and row[0]) + + def _already_in_library_at_cutoff(conn: psycopg.Connection, target: MBTarget, cutoff: int) -> bool: with conn.cursor() as cur: cur.execute( @@ -237,8 +250,11 @@ def run_pipeline( if resolved is not None: target = resolved # dedupe: a monitor-driven upgrade job proceeds unless the existing copy already meets - # the cutoff; a plain request short-circuits on any existing copy (slice-1 behavior). - if upgrade_cutoff is not None and _is_upgrade_job(conn, job_id): + # the cutoff; a plain request short-circuits on any existing copy (slice-1 behavior). A + # user-forced re-acquire skips the dedupe entirely (keep-if-better still guards the import). + if _is_force_job(conn, job_id): + _dedupe_hit = False + elif upgrade_cutoff is not None and _is_upgrade_job(conn, job_id): _dedupe_hit = _already_in_library_at_cutoff(conn, target, upgrade_cutoff) else: _dedupe_hit = _already_in_library(conn, target) diff --git a/worker/tests/conftest.py b/worker/tests/conftest.py index 5ef2559..f70cf0c 100644 --- a/worker/tests/conftest.py +++ b/worker/tests/conftest.py @@ -56,13 +56,13 @@ def conn(): connection.close() -def insert_request(conn, artist="Artist", album="Album"): +def insert_request(conn, artist="Artist", album="Album", force=False): """Insert a Request + its Job (state 'requested') and return the job id.""" with conn.cursor() as cur: cur.execute( - 'INSERT INTO "Request" (id, artist, album, status, "createdAt") ' - "VALUES (gen_random_uuid()::text, %s, %s, 'pending', now()) RETURNING id", - (artist, album), + 'INSERT INTO "Request" (id, artist, album, status, force, "createdAt") ' + "VALUES (gen_random_uuid()::text, %s, %s, 'pending', %s, now()) RETURNING id", + (artist, album, force), ) request_id = cur.fetchone()[0] cur.execute( diff --git a/worker/tests/test_pipeline.py b/worker/tests/test_pipeline.py index f2245ff..fd2f795 100644 --- a/worker/tests/test_pipeline.py +++ b/worker/tests/test_pipeline.py @@ -124,6 +124,25 @@ def test_dedupe_skips_already_in_library(conn): assert cur.fetchone()[0] == 1 # not duplicated +def test_force_reacquires_an_album_already_in_library(conn): + # First acquisition puts the album in the library. + job1 = insert_request(conn, artist="Radiohead", album="In Rainbows") + claim_next(conn) + run_pipeline(conn, job1, [FakeQobuz()], dest_root="/tmp/lib") + # A forced re-acquire skips the dedupe → it searches + persists candidates again. + job2 = insert_request(conn, artist="Radiohead", album="In Rainbows", force=True) + claim_next(conn) + run_pipeline(conn, job2, [FakeQobuz()], dest_root="/tmp/lib", upgrade_cutoff=2) + + assert _job_state(conn, job2) == ("imported", "import") + with conn.cursor() as cur: + cur.execute('SELECT count(*) FROM "Candidate" WHERE "jobId" = %s', (job2,)) + assert cur.fetchone()[0] >= 1 # dedupe bypassed — it actually ran the search + cur.execute('SELECT count(*) FROM "LibraryItem" WHERE artist = %s AND album = %s', + ("Radiohead", "In Rainbows")) + assert cur.fetchone()[0] == 1 # still not duplicated (keep-if-better import) + + def test_below_threshold_candidates_are_persisted_for_diagnosis(conn): from lyra_worker.adapters.fakes import FakeQobuz from lyra_worker.types import MBTarget