feat: monitor retry/enqueue phase

worker/lyra_worker/monitor.py

@@ -74,3 +74,68 @@ def discover(conn: psycopg.Connection, browser: MbBrowser, cfg: MonitorConfig) -
             cur.execute('UPDATE "WatchedArtist" SET "lastPolledAt" = now() WHERE id = %s', (artist_id,))
     conn.commit()
     return inserted
+
+
+def _set_state(conn: psycopg.Connection, rel_id: str, state: str) -> None:
+    with conn.cursor() as cur:
+        cur.execute('UPDATE "MonitoredRelease" SET state = %s WHERE id = %s', (state, rel_id))
+
+
+def _in_library_at_cutoff(conn: psycopg.Connection, artist: str, album: str, 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',
+            (artist, album, cutoff),
+        )
+        return cur.fetchone() is not None
+
+
+def _enqueue_one(conn: psycopg.Connection, rel_id: str, artist: str, album: str) -> None:
+    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),
+        )
+        request_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())",
+            (request_id,),
+        )
+        cur.execute('UPDATE "MonitoredRelease" SET "lastSearchedAt" = now() WHERE id = %s', (rel_id,))
+
+
+def enqueue_due(conn: psycopg.Connection, cfg: MonitorConfig) -> int:
+    """Enqueue due wanted/upgrade releases as pipeline jobs. Returns #jobs enqueued."""
+    with conn.cursor() as cur:
+        cur.execute(
+            'SELECT mr.id, mr."artistName", mr.album, mr.state, mr."currentQualityClass", '
+            '  (mr."firstGrabbedAt" IS NOT NULL '
+            '   AND mr."firstGrabbedAt" < now() - make_interval(days => %s)) AS window_expired '
+            'FROM "MonitoredRelease" mr '
+            'WHERE mr.monitored = true '
+            "  AND mr.state <> 'fulfilled' "
+            '  AND (mr."lastSearchedAt" IS NULL '
+            '       OR mr."lastSearchedAt" < now() - make_interval(hours => %s)) '
+            '  AND NOT EXISTS ('
+            '    SELECT 1 FROM "Request" r JOIN "Job" j ON j."requestId" = r.id '
+            '    WHERE r."monitoredReleaseId" = mr.id '
+            "      AND j.state NOT IN ('imported', 'needs_attention'))",
+            (cfg.upgrade_window_days, cfg.retry_interval_hours),
+        )
+        rows = cur.fetchall()
+
+    enqueued = 0
+    for rel_id, artist, album, state, quality, window_expired in rows:
+        if _in_library_at_cutoff(conn, artist, album, cfg.quality_cutoff):
+            _set_state(conn, rel_id, "fulfilled")
+            continue
+        if state == "grabbed" and (window_expired or (quality is not None and quality >= cfg.quality_cutoff)):
+            _set_state(conn, rel_id, "fulfilled")
+            continue
+        _enqueue_one(conn, rel_id, artist, album)
+        enqueued += 1
+    conn.commit()
+    return enqueued

worker/tests/test_monitor_enqueue.py

@@ -0,0 +1,83 @@
+from lyra_worker.monitor import MonitorConfig, enqueue_due
+from tests.conftest import insert_monitored_release
+
+CFG = MonitorConfig(enabled=True, retry_interval_hours=6, quality_cutoff=2, upgrade_window_days=14)
+
+
+def _job_count_for(conn, rel_id):
+    with conn.cursor() as cur:
+        cur.execute(
+            'SELECT count(*) FROM "Request" r JOIN "Job" j ON j."requestId" = r.id '
+            'WHERE r."monitoredReleaseId" = %s',
+            (rel_id,),
+        )
+        return cur.fetchone()[0]
+
+
+def _state(conn, rel_id):
+    with conn.cursor() as cur:
+        cur.execute('SELECT state FROM "MonitoredRelease" WHERE id = %s', (rel_id,))
+        return cur.fetchone()[0]
+
+
+def test_enqueues_a_wanted_release(conn):
+    rel_id = insert_monitored_release(conn, artist_name="John Mayer", album="Continuum",
+                                      rg_mbid="rg-1", monitored=True, state="wanted")
+    assert enqueue_due(conn, CFG) == 1
+    assert _job_count_for(conn, rel_id) == 1
+    with conn.cursor() as cur:
+        cur.execute(
+            'SELECT artist, album FROM "Request" WHERE "monitoredReleaseId" = %s', (rel_id,)
+        )
+        assert cur.fetchone() == ("John Mayer", "Continuum")
+
+
+def test_skips_unmonitored_and_fulfilled(conn):
+    insert_monitored_release(conn, rg_mbid="rg-a", monitored=False, state="wanted")
+    insert_monitored_release(conn, rg_mbid="rg-b", monitored=True, state="fulfilled")
+    assert enqueue_due(conn, CFG) == 0
+
+
+def test_skips_when_recently_searched(conn):
+    insert_monitored_release(conn, rg_mbid="rg-1", monitored=True, state="wanted",
+                             last_searched_sql="now() - interval '1 hour'")  # < 6h retry
+    assert enqueue_due(conn, CFG) == 0
+
+
+def test_skips_release_with_active_job(conn):
+    rel_id = insert_monitored_release(conn, rg_mbid="rg-1", monitored=True, state="wanted")
+    assert enqueue_due(conn, CFG) == 1          # first sweep enqueues
+    assert enqueue_due(conn, CFG) == 0          # active (requested) job blocks a second
+
+
+def test_grabbed_within_window_below_cutoff_re_searches(conn):
+    rel_id = insert_monitored_release(conn, rg_mbid="rg-1", monitored=True, state="grabbed",
+                                      current_quality_class=1, first_grabbed_sql="now() - interval '1 day'")
+    assert enqueue_due(conn, CFG) == 1
+
+
+def test_grabbed_past_window_is_fulfilled_not_enqueued(conn):
+    rel_id = insert_monitored_release(conn, rg_mbid="rg-1", monitored=True, state="grabbed",
+                                      current_quality_class=1, first_grabbed_sql="now() - interval '30 days'")
+    assert enqueue_due(conn, CFG) == 0
+    assert _state(conn, rel_id) == "fulfilled"
+
+
+def test_already_in_library_at_cutoff_is_fulfilled(conn):
+    rel_id = insert_monitored_release(conn, artist_name="A", album="B", rg_mbid="rg-1",
+                                      monitored=True, state="wanted")
+    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 "LibraryItem" (id, "requestId", artist, album, path, source, format, '
+            '"qualityClass", "importedAt") '
+            "VALUES (gen_random_uuid()::text, %s, 'A', 'B', '/m/A/B', 'qobuz', 'FLAC', 2, now())",
+            (req,),
+        )
+    conn.commit()
+    assert enqueue_due(conn, CFG) == 0
+    assert _state(conn, rel_id) == "fulfilled"