diff --git a/worker/lyra_worker/monitor.py b/worker/lyra_worker/monitor.py index 733457f..a884056 100644 --- a/worker/lyra_worker/monitor.py +++ b/worker/lyra_worker/monitor.py @@ -139,3 +139,36 @@ def enqueue_due(conn: psycopg.Connection, cfg: MonitorConfig) -> int: enqueued += 1 conn.commit() return enqueued + + +def reconcile(conn: psycopg.Connection, job_id: str, cfg: MonitorConfig) -> None: + """Feed a finished job's outcome back into its MonitoredRelease (if any).""" + with conn.cursor() as cur: + cur.execute( + 'SELECT j.state, r."monitoredReleaseId", r.artist, r.album ' + 'FROM "Job" j JOIN "Request" r ON r.id = j."requestId" WHERE j.id = %s', + (job_id,), + ) + row = cur.fetchone() + if row is None: + return + state, rel_id, artist, album = row + if rel_id is None or state != "imported": + return # needs_attention (or in-flight) leaves the release wanted + + with conn.cursor() as cur: + cur.execute( + 'SELECT "qualityClass" FROM "LibraryItem" WHERE artist = %s AND album = %s', + (artist, album), + ) + lib = cur.fetchone() + quality = lib[0] if lib else None + new_state = "fulfilled" if (quality is not None and quality >= cfg.quality_cutoff) else "grabbed" + with conn.cursor() as cur: + cur.execute( + 'UPDATE "MonitoredRelease" ' + 'SET "currentQualityClass" = %s, state = %s, ' + ' "firstGrabbedAt" = COALESCE("firstGrabbedAt", now()) WHERE id = %s', + (quality, new_state, rel_id), + ) + conn.commit() diff --git a/worker/tests/test_monitor_reconcile.py b/worker/tests/test_monitor_reconcile.py new file mode 100644 index 0000000..23a3946 --- /dev/null +++ b/worker/tests/test_monitor_reconcile.py @@ -0,0 +1,86 @@ +from lyra_worker.monitor import MonitorConfig, reconcile +from tests.conftest import insert_monitored_release + +CFG = MonitorConfig(enabled=True, quality_cutoff=2) + + +def _make_job(conn, rel_id, artist, album, state): + 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 = cur.fetchone()[0] + cur.execute( + 'INSERT INTO "Job" (id, "requestId", state, "currentStage", attempts, "createdAt", "updatedAt") ' + "VALUES (gen_random_uuid()::text, %s, %s, 'import', 1, now(), now()) RETURNING id", + (req, state), + ) + job = cur.fetchone()[0] + conn.commit() + return req, job + + +def _add_library(conn, req, artist, album, quality): + with conn.cursor() as cur: + cur.execute( + 'INSERT INTO "LibraryItem" (id, "requestId", artist, album, path, source, format, ' + '"qualityClass", "importedAt") ' + "VALUES (gen_random_uuid()::text, %s, %s, %s, '/m', 'qobuz', 'FLAC', %s, now())", + (req, artist, album, quality), + ) + conn.commit() + + +def _release(conn, rel_id): + with conn.cursor() as cur: + cur.execute( + 'SELECT state, "currentQualityClass", "firstGrabbedAt" IS NOT NULL ' + 'FROM "MonitoredRelease" WHERE id = %s', + (rel_id,), + ) + return cur.fetchone() + + +def test_imported_below_cutoff_becomes_grabbed(conn): + rel = insert_monitored_release(conn, artist_name="A", album="B", rg_mbid="rg-1", + monitored=True, state="wanted") + req, job = _make_job(conn, rel, "A", "B", "imported") + _add_library(conn, req, "A", "B", 1) + reconcile(conn, job, CFG) + assert _release(conn, rel) == ("grabbed", 1, True) + + +def test_imported_at_cutoff_becomes_fulfilled(conn): + rel = insert_monitored_release(conn, artist_name="A", album="B", rg_mbid="rg-1", + monitored=True, state="wanted") + req, job = _make_job(conn, rel, "A", "B", "imported") + _add_library(conn, req, "A", "B", 3) + reconcile(conn, job, CFG) + assert _release(conn, rel) == ("fulfilled", 3, True) + + +def test_needs_attention_leaves_wanted(conn): + rel = insert_monitored_release(conn, artist_name="A", album="B", rg_mbid="rg-1", + monitored=True, state="wanted") + _req, job = _make_job(conn, rel, "A", "B", "needs_attention") + reconcile(conn, job, CFG) + assert _release(conn, rel) == ("wanted", None, False) + + +def test_unlinked_job_is_ignored(conn): + with conn.cursor() as cur: + cur.execute( + 'INSERT INTO "Request" (id, artist, album, status, "createdAt") ' + "VALUES (gen_random_uuid()::text, 'A', 'B', 'completed', now()) RETURNING id" + ) + req = cur.fetchone()[0] + cur.execute( + 'INSERT INTO "Job" (id, "requestId", state, "currentStage", attempts, "createdAt", "updatedAt") ' + "VALUES (gen_random_uuid()::text, %s, 'imported', 'import', 1, now(), now()) RETURNING id", + (req,), + ) + job = cur.fetchone()[0] + conn.commit() + reconcile(conn, job, CFG) # no monitoredReleaseId → no-op, must not raise