feat: monitor outcome reconcile

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()

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