diff --git a/worker/lyra_worker/monitor.py b/worker/lyra_worker/monitor.py index 9c13766..733457f 100644 --- a/worker/lyra_worker/monitor.py +++ b/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 diff --git a/worker/tests/test_monitor_enqueue.py b/worker/tests/test_monitor_enqueue.py new file mode 100644 index 0000000..a86707f --- /dev/null +++ b/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"