diff --git a/web/prisma/migrations/20260713220000_add_job_download_progress/migration.sql b/web/prisma/migrations/20260713220000_add_job_download_progress/migration.sql new file mode 100644 index 0000000..12fe0fe --- /dev/null +++ b/web/prisma/migrations/20260713220000_add_job_download_progress/migration.sql @@ -0,0 +1,2 @@ +-- AlterTable +ALTER TABLE "Job" ADD COLUMN "downloadProgress" DOUBLE PRECISION NOT NULL DEFAULT 0; diff --git a/web/prisma/schema.prisma b/web/prisma/schema.prisma index 77763e9..f73d998 100644 --- a/web/prisma/schema.prisma +++ b/web/prisma/schema.prisma @@ -43,17 +43,18 @@ model Request { } model Job { - id String @id @default(cuid()) - request Request @relation(fields: [requestId], references: [id], onDelete: Cascade) - requestId String @unique - state JobState @default(requested) - currentStage String @default("intake") - attempts Int @default(0) - error String? - claimedAt DateTime? - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt - candidates Candidate[] + id String @id @default(cuid()) + request Request @relation(fields: [requestId], references: [id], onDelete: Cascade) + requestId String @unique + state JobState @default(requested) + currentStage String @default("intake") + attempts Int @default(0) + error String? + claimedAt DateTime? + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + downloadProgress Float @default(0) + candidates Candidate[] } model Candidate { diff --git a/web/src/app/api/requests/route.test.ts b/web/src/app/api/requests/route.test.ts index 26dc6cd..51130fa 100644 --- a/web/src/app/api/requests/route.test.ts +++ b/web/src/app/api/requests/route.test.ts @@ -62,7 +62,15 @@ describe("requests API", () => { data: { artist: "Chosen Artist", album: "Chosen Album", - job: { create: { state: "downloading", currentStage: "download", attempts: 2, error: "boom" } }, + job: { + create: { + state: "downloading", + currentStage: "download", + attempts: 2, + error: "boom", + downloadProgress: 0.42, + }, + }, }, include: { job: true }, }); @@ -99,6 +107,7 @@ describe("requests API", () => { expect(found.job.chosen).toEqual({ source: "qobuz", format: "flac", trackCount: 12 }); expect(found.job.error).toBe("boom"); expect(found.job.attempts).toBe(2); + expect(found.job.downloadProgress).toBe(0.42); }); it("reports no chosen candidate and zero count when a job has none", async () => { @@ -109,5 +118,6 @@ describe("requests API", () => { const found = body.requests.find((r: { id: string }) => r.id === created.id); expect(found.job.candidateCount).toBe(0); expect(found.job.chosen).toBeNull(); + expect(found.job.downloadProgress).toBe(0); }); }); diff --git a/web/src/app/api/requests/route.ts b/web/src/app/api/requests/route.ts index d7ac3a4..8087281 100644 --- a/web/src/app/api/requests/route.ts +++ b/web/src/app/api/requests/route.ts @@ -63,6 +63,7 @@ export async function GET() { error: r.job.error, claimedAt: r.job.claimedAt, updatedAt: r.job.updatedAt, + downloadProgress: r.job.downloadProgress, candidateCount: cands.length, chosen: chosen ? { source: chosen.source, format: chosen.format, trackCount: chosen.trackCount } : null, }; diff --git a/worker/lyra_worker/pipeline.py b/worker/lyra_worker/pipeline.py index c3dba71..a1f4659 100644 --- a/worker/lyra_worker/pipeline.py +++ b/worker/lyra_worker/pipeline.py @@ -1,4 +1,6 @@ +import os import shutil +import threading from dataclasses import replace from typing import Sequence @@ -11,6 +13,39 @@ from lyra_worker.quality import quality_class from lyra_worker.ranker import rank_candidates from lyra_worker.types import Candidate, MBTarget +_AUDIO_EXT = {".flac", ".mp3", ".m4a", ".opus", ".ogg", ".aac", ".wav"} + + +def _count_staged_audio(staging: str) -> int: + """Count audio files anywhere under the staging dir (streamrip nests them in a subfolder).""" + n = 0 + for _root, _dirs, files in os.walk(staging): + for f in files: + if os.path.splitext(f)[1].lower() in _AUDIO_EXT: + n += 1 + return n + + +def _poll_download_progress(job_id: str, staging: str, expected: int, stop: "threading.Event") -> None: + """Until `stop`, periodically write Job.downloadProgress = files-in-staging / expected + (capped 0.99). Uses its own short-lived connection (psycopg conns aren't shareable across + threads). A missing DSN or expected<=0 just no-ops.""" + dsn = os.environ.get("DATABASE_URL") + if not dsn or expected <= 0: + return + conn = psycopg.connect(dsn) + try: + while not stop.is_set(): + frac = min(_count_staged_audio(staging) / expected, 0.99) + with conn.cursor() as cur: + cur.execute('UPDATE "Job" SET "downloadProgress" = %s WHERE id = %s', (frac, job_id)) + conn.commit() + stop.wait(1.5) + except Exception as e: # a progress poller must never affect the job + print(f"pipeline: download progress poller error: {e}", flush=True) + finally: + conn.close() + def _set_state(conn: psycopg.Connection, job_id: str, state: str, stage: str) -> None: with conn.cursor() as cur: @@ -21,6 +56,12 @@ def _set_state(conn: psycopg.Connection, job_id: str, state: str, stage: str) -> conn.commit() +def _set_download_progress(conn: psycopg.Connection, job_id: str, frac: float) -> None: + with conn.cursor() as cur: + cur.execute('UPDATE "Job" SET "downloadProgress" = %s WHERE id = %s', (frac, job_id)) + conn.commit() + + def _request_id(conn: psycopg.Connection, job_id: str) -> str: with conn.cursor() as cur: cur.execute('SELECT "requestId" FROM "Job" WHERE id = %s', (job_id,)) @@ -201,20 +242,32 @@ def run_pipeline( staging = staging_dir(staging_root or f"{dest_root}/.staging", job_id) winner = None result = None + _set_download_progress(conn, job_id, 0.0) # reset for this run + expected = target.track_count or (ranked[0].track_count if ranked else 0) + _stop = threading.Event() + _poller = threading.Thread( + target=_poll_download_progress, args=(job_id, staging, expected, _stop), daemon=True + ) + _poller.start() try: - for candidate in ranked: - adapter = by_source.get(candidate.source) - if adapter is None: - continue - _mark_chosen(conn, job_id, candidate.source_ref) # reflect the source now in flight - shutil.rmtree(staging, ignore_errors=True) # clean slate per attempt - result = adapter.download(candidate, staging, lambda _pct: None) - if result.ok: - winner = candidate - break + try: + for candidate in ranked: + adapter = by_source.get(candidate.source) + if adapter is None: + continue + _mark_chosen(conn, job_id, candidate.source_ref) # reflect the source now in flight + shutil.rmtree(staging, ignore_errors=True) # clean slate per attempt + result = adapter.download(candidate, staging, lambda _pct: None) + if result.ok: + winner = candidate + break + finally: + _stop.set() + _poller.join(timeout=3) if winner is None or result is None or not result.ok: _fail(conn, job_id, "all downloads failed") return + _set_download_progress(conn, job_id, 1.0) # download done — UI shows 100% into Finishing # 5. verify completeness on the staging copy, then promote + tag _set_state(conn, job_id, "tagging", "tag") diff --git a/worker/tests/test_download_progress.py b/worker/tests/test_download_progress.py new file mode 100644 index 0000000..9be3ad5 --- /dev/null +++ b/worker/tests/test_download_progress.py @@ -0,0 +1,22 @@ +import os + +from lyra_worker.pipeline import _count_staged_audio + + +def test_count_staged_audio_counts_nested_audio_files_only(tmp_path): + nested = tmp_path / "Artist" / "Album" + nested.mkdir(parents=True) + (nested / "01.flac").write_bytes(b"") + (nested / "02.flac").write_bytes(b"") + (nested / "03.flac").write_bytes(b"") + (nested / "cover.jpg").write_bytes(b"") + + assert _count_staged_audio(str(tmp_path)) == 3 + + +def test_count_staged_audio_empty_dir_is_zero(tmp_path): + assert _count_staged_audio(str(tmp_path)) == 0 + + +def test_count_staged_audio_missing_dir_is_zero(): + assert _count_staged_audio("/nonexistent/path/for/sure") == 0 diff --git a/worker/tests/test_pipeline.py b/worker/tests/test_pipeline.py index 0a5bc28..f2245ff 100644 --- a/worker/tests/test_pipeline.py +++ b/worker/tests/test_pipeline.py @@ -19,6 +19,12 @@ def _request_status(conn, job_id): return cur.fetchone()[0] +def _download_progress(conn, job_id): + with conn.cursor() as cur: + cur.execute('SELECT "downloadProgress" FROM "Job" WHERE id = %s', (job_id,)) + return cur.fetchone()[0] + + def test_success_picks_best_source_and_imports(conn): job_id = insert_request(conn, artist="Radiohead", album="In Rainbows") claim_next(conn) @@ -26,6 +32,7 @@ def test_success_picks_best_source_and_imports(conn): assert _job_state(conn, job_id) == ("imported", "import") assert _request_status(conn, job_id) == "completed" + assert _download_progress(conn, job_id) == 1.0 # a successful download completes at 100% with conn.cursor() as cur: cur.execute('SELECT source FROM "Candidate" WHERE "jobId" = %s AND chosen = true', (job_id,)) assert cur.fetchone()[0] == "qobuz" # highest quality won @@ -41,6 +48,7 @@ def test_falls_through_when_best_download_fails(conn): run_pipeline(conn, job_id, [FailingAdapter(), FakeSoulseek()], dest_root="/tmp/lib") assert _job_state(conn, job_id) == ("imported", "import") + assert _download_progress(conn, job_id) == 1.0 # falls through, but still completes at 100% with conn.cursor() as cur: cur.execute('SELECT source FROM "Candidate" WHERE "jobId" = %s AND chosen = true', (job_id,)) assert cur.fetchone()[0] == "soulseek" @@ -75,6 +83,7 @@ def test_all_downloads_fail_goes_to_needs_attention(conn): run_pipeline(conn, job_id, [FailingAdapter()], dest_root="/tmp/lib") assert _job_state(conn, job_id)[0] == "needs_attention" + assert _download_progress(conn, job_id) == 0.0 # never reached 1.0 — all attempts failed def test_incomplete_download_goes_to_needs_attention(conn):