diff --git a/worker/lyra_worker/browser.py b/worker/lyra_worker/browser.py index 84932a3..503686b 100644 --- a/worker/lyra_worker/browser.py +++ b/worker/lyra_worker/browser.py @@ -1,4 +1,4 @@ -from dataclasses import dataclass, field +from dataclasses import dataclass from typing import Protocol diff --git a/worker/lyra_worker/main.py b/worker/lyra_worker/main.py index 4d8af2a..657ab51 100644 --- a/worker/lyra_worker/main.py +++ b/worker/lyra_worker/main.py @@ -37,7 +37,8 @@ def run_forever() -> None: time.sleep(IDLE_SLEEP) continue print(f"worker: claimed job {job_id}", flush=True) - run_pipeline(conn, job_id, adapters, resolver=resolver, tagger=tagger) + run_pipeline(conn, job_id, adapters, resolver=resolver, tagger=tagger, + upgrade_cutoff=mcfg.quality_cutoff) if mcfg.enabled: try: reconcile(conn, job_id, mcfg) diff --git a/worker/lyra_worker/pipeline.py b/worker/lyra_worker/pipeline.py index 8a1d12d..3aad183 100644 --- a/worker/lyra_worker/pipeline.py +++ b/worker/lyra_worker/pipeline.py @@ -58,6 +58,27 @@ def _already_in_library(conn: psycopg.Connection, target: MBTarget) -> bool: return cur.fetchone() is not None +def _is_upgrade_job(conn: psycopg.Connection, job_id: str) -> bool: + with conn.cursor() as cur: + cur.execute( + 'SELECT "monitoredReleaseId" IS NOT NULL 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( + 'SELECT 1 FROM "LibraryItem" WHERE artist = %s AND album = %s ' + 'AND "qualityClass" >= %s LIMIT 1', + (target.artist, target.album, cutoff), + ) + return cur.fetchone() is not None + + def _persist_candidates(conn: psycopg.Connection, job_id: str, candidates: list[Candidate]) -> None: with conn.cursor() as cur: for c in candidates: @@ -110,6 +131,7 @@ def run_pipeline( tagger=None, min_confidence: float = 0.7, dest_root: str = "/music", + upgrade_cutoff: int | None = None, ) -> None: """Real staged acquisition using source-agnostic adapters. Fakes in this plan.""" names = [a.name for a in adapters] @@ -123,7 +145,13 @@ def run_pipeline( resolved = resolver.resolve(target.artist, target.album) if resolved is not None: target = resolved - if _already_in_library(conn, target): + # 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): + _dedupe_hit = _already_in_library_at_cutoff(conn, target, upgrade_cutoff) + else: + _dedupe_hit = _already_in_library(conn, target) + if _dedupe_hit: _set_state(conn, job_id, "imported", "import") with conn.cursor() as cur: cur.execute( diff --git a/worker/tests/test_upgrade_e2e.py b/worker/tests/test_upgrade_e2e.py new file mode 100644 index 0000000..6086f65 --- /dev/null +++ b/worker/tests/test_upgrade_e2e.py @@ -0,0 +1,126 @@ +from lyra_worker.adapters.qobuz import QobuzAdapter +from lyra_worker.claim import claim_next +from lyra_worker.monitor import MonitorConfig, enqueue_due, reconcile +from lyra_worker.pipeline import run_pipeline +from tests.conftest import insert_monitored_release +from tests.test_qobuz_adapter import FakeQobuzClient + +CFG = MonitorConfig(enabled=True, quality_cutoff=2, retry_interval_hours=6, upgrade_window_days=14) + + +def _seed_library_item(conn, artist, album, quality_class, path): + """Seed a throwaway Request (for its requestId) + a LibraryItem at the given quality, + mirroring how test_monitor_enqueue.py's already-in-library test seeds a LibraryItem.""" + with conn.cursor() as cur: + cur.execute( + 'INSERT INTO "Request" (id, artist, album, status, "createdAt") ' + "VALUES (gen_random_uuid()::text, %s, %s, 'completed', now()) RETURNING id", + (artist, album), + ) + req_id = cur.fetchone()[0] + cur.execute( + 'INSERT INTO "LibraryItem" (id, "requestId", artist, album, path, source, format, ' + '"qualityClass", "importedAt") ' + "VALUES (gen_random_uuid()::text, %s, %s, %s, %s, 'qobuz', 'FLAC', %s, now())", + (req_id, artist, album, path, quality_class), + ) + conn.commit() + + +def _library_row(conn, artist, album): + with conn.cursor() as cur: + cur.execute( + 'SELECT path, "qualityClass", source FROM "LibraryItem" WHERE artist = %s AND album = %s', + (artist, album), + ) + return cur.fetchone() + + +def _job_state(conn, job_id): + with conn.cursor() as cur: + cur.execute('SELECT state FROM "Job" WHERE id = %s', (job_id,)) + return cur.fetchone()[0] + + +def _monitored_release_row(conn, rel_id): + with conn.cursor() as cur: + cur.execute( + 'SELECT state, "currentQualityClass" FROM "MonitoredRelease" WHERE id = %s', (rel_id,) + ) + return cur.fetchone() + + +def _enqueue_monitor_job_directly(conn, rel_id, artist, album): + """Insert a Request+Job tagged with monitoredReleaseId, the same shape enqueue_due's + _enqueue_one produces. Used when enqueue_due itself would refuse to enqueue (e.g. the + existing copy already meets the cutoff, in which case enqueue_due marks it fulfilled and + skips it -- see test_monitor_enqueue.test_already_in_library_at_cutoff_is_fulfilled).""" + with conn.cursor() as cur: + cur.execute( + 'INSERT INTO "Request" (id, artist, album, status, "monitoredReleaseId", "createdAt") ' + "VALUES (gen_random_uuid()::text, %s, %s, 'pending', %s, now()) RETURNING id", + (artist, album, rel_id), + ) + req_id = cur.fetchone()[0] + cur.execute( + 'INSERT INTO "Job" (id, "requestId", state, "currentStage", attempts, "createdAt", "updatedAt") ' + "VALUES (gen_random_uuid()::text, %s, 'requested', 'intake', 0, now(), now()) RETURNING id", + (req_id,), + ) + job_id = cur.fetchone()[0] + conn.commit() + return job_id + + +def test_below_cutoff_copy_is_upgraded_end_to_end(conn): + rel_id = insert_monitored_release( + conn, artist_name="John Mayer", album="Continuum", rg_mbid="rg-upgrade-below", + monitored=True, state="grabbed", current_quality_class=1, + first_grabbed_sql="now() - interval '1 day'", + ) + _seed_library_item(conn, "John Mayer", "Continuum", quality_class=1, path="/m/old-lossy") + + # the grabbed-below-cutoff release is due and gets enqueued as a monitor-driven job. + assert enqueue_due(conn, CFG) == 1 + + job_id = claim_next(conn) + assert job_id is not None + + # upgrade_cutoff=2 so intake does NOT short-circuit on the existing quality-1 copy. + run_pipeline(conn, job_id, [QobuzAdapter(FakeQobuzClient())], upgrade_cutoff=2) + + assert _job_state(conn, job_id) == "imported" + path, quality, source = _library_row(conn, "John Mayer", "Continuum") + # Q: FakeQobuzClient's candidate is 24-bit/96kHz FLAC -> hi-res, qualityClass 3. + assert quality == 3 + assert quality > 1 + assert source == "qobuz" + assert path != "/m/old-lossy" # the upsert replaced the below-cutoff copy + + reconcile(conn, job_id, CFG) + state, cqc = _monitored_release_row(conn, rel_id) + assert state == "fulfilled" # Q(=3) >= cutoff(=2) + assert cqc == 3 + + +def test_already_at_cutoff_copy_is_not_replaced(conn): + """Guard test pinning the fix: when the existing copy already meets the cutoff, a + monitor-driven job must still short-circuit at intake (no needless re-download).""" + rel_id = insert_monitored_release( + conn, artist_name="John Mayer", album="Continuum", rg_mbid="rg-upgrade-at-cutoff", + monitored=True, state="grabbed", current_quality_class=3, + first_grabbed_sql="now() - interval '1 day'", + ) + _seed_library_item(conn, "John Mayer", "Continuum", quality_class=3, path="/m/already-hires") + + # enqueue_due would mark this release fulfilled and refuse to enqueue it (already at + # cutoff), so build the monitor-driven job directly to exercise run_pipeline's intake path. + job_id = _enqueue_monitor_job_directly(conn, rel_id, "John Mayer", "Continuum") + claimed = claim_next(conn) + assert claimed == job_id + + run_pipeline(conn, job_id, [QobuzAdapter(FakeQobuzClient())], upgrade_cutoff=2) + + assert _job_state(conn, job_id) == "imported" + # unchanged: still the original below-Q, already-at-cutoff copy, untouched. + assert _library_row(conn, "John Mayer", "Continuum") == ("/m/already-hires", 3, "qobuz")