feat(downloads): live per-file progress on running events — #709

Now that we own the walk, surface live counts on the in-flight download in the
Downloads view. ingest_core.run takes an event_id and does a TIME-THROTTLED
write (~5s, decoupled from page boundaries so it ticks steadily regardless of
how big/slow a page is) of {downloaded, skipped, errors, quarantined, posts} to
the running download_event's metadata.live (jsonb_set; short session; status
guard so a finalized event isn't clobbered). download_backends threads
event_id from ctx; the /api/downloads list surfaces `live`; ActiveDownloadsPanel
renders it beside the elapsed timer. Native (Patreon) only — gallery-dl is an
opaque subprocess; the row only shows when `live` is present. Phase 3 overwrites
metadata with run_stats on finish, dropping `live`.

Test: _write_live_progress updates a running event's metadata.live and leaves a
finalized (status != running) event alone.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

backend/app/api/downloads.py

@@ -44,6 +44,9 @@ def _list_record(event: DownloadEvent, source: Source | None, artist: Artist | N
         "bytes_downloaded": event.bytes_downloaded,
         "error": event.error,
         "summary": _summary_from_metadata(event.metadata_),
+        # plan #709: mid-walk live counts for a RUNNING native-ingester event
+        # (None otherwise; phase 3 overwrites metadata with run_stats on finish).
+        "live": (event.metadata_ or {}).get("live"),
     }
 
 

backend/app/services/download_backends.py

@@ -139,6 +139,8 @@ async def _run_native_ingester(
             resume_cursor=source_config.resume_cursor,
             time_budget_seconds=source_config.timeout,
             posts_base=int(overrides.get("_backfill_posts", 0)),
+            # plan #709: live progress writes to this running event mid-walk.
+            event_id=ctx.get("event_id"),
         ),
     )
     return dl_result, resolved_campaign_id

backend/app/services/ingest_core.py

@@ -25,6 +25,7 @@ Plain-HTTP homelab: no secure-context Web API.
 
 from __future__ import annotations
 
+import json
 import logging
 import time
 from collections.abc import Callable
@@ -48,6 +49,12 @@ DEAD_LETTER_THRESHOLD = 3
 # last_error is Text but bound it so a giant traceback doesn't bloat the row.
 _ERROR_MAX = 1000
 
+# plan #709: throttle the live-progress write to the running DownloadEvent to one
+# every ~5s — a steady cadence for the Downloads view regardless of how big/slow a
+# page is (page boundaries can be minutes apart on image-dense backfills, so a
+# page-tied update would lurch). Trivial churn (~one single-row UPDATE / 5s).
+_LIVE_PROGRESS_INTERVAL = 5.0
+
 
 class Ingester:
     """Generic native-ingest orchestration. Subclass with a platform adapter
@@ -92,6 +99,7 @@ class Ingester:
         time_budget_seconds: float = 870.0,
         seen_threshold: int = _TICK_SEEN_THRESHOLD,
         posts_base: int = 0,
+        event_id: int | None = None,
     ) -> DownloadResult:
         """Walk + download for one source, returning a gallery-dl-shaped result.
 
@@ -109,6 +117,7 @@ class Ingester:
         checkpoint = mode in ("backfill", "recovery")
         ledger_key = self._ledger_key
         start = time.monotonic()
+        last_live = start  # plan #709: last live-progress write timestamp
         log_lines: list[str] = []
         written: list[str] = []
         quarantined_paths: list[str] = []
@@ -288,6 +297,19 @@ class Ingester:
                 if to_fail:
                     self._record_failures(source_id, to_fail)
 
+                # plan #709: time-throttled live progress to the running event so
+                # the Downloads view ticks ~every 5s, independent of page size.
+                now = time.monotonic()
+                if event_id is not None and (now - last_live) >= _LIVE_PROGRESS_INTERVAL:
+                    last_live = now
+                    self._write_live_progress(event_id, {
+                        "downloaded": downloaded,
+                        "skipped": skipped_count,
+                        "errors": errors,
+                        "quarantined": quarantined,
+                        "posts": posts_processed,
+                    })
+
                 if early_out:
                     break
             else:
@@ -474,6 +496,25 @@ class Ingester:
             )
             session.commit()
 
+    def _write_live_progress(self, event_id: int, counts: dict) -> None:
+        """Throttled mid-walk write of live counts to the RUNNING download_event
+        (plan #709) so the Downloads view shows progress before the chunk
+        finishes. A short session (never held across the walk); the `status =
+        'running'` guard avoids clobbering an event phase 3 already finalized.
+        `metadata` is JSONB — jsonb_set sets just the `live` key, leaving the rest
+        for phase 3 to overwrite with the final run_stats."""
+        with self.session_factory() as session:
+            session.execute(
+                text(
+                    "UPDATE download_event SET metadata = jsonb_set("
+                    "  coalesce(metadata, '{}'::jsonb), '{live}',"
+                    "  cast(:live AS jsonb)) "
+                    "WHERE id = :eid AND status = 'running'"
+                ),
+                {"live": json.dumps(counts), "eid": event_id},
+            )
+            session.commit()
+
     def _still_running(self, source_id: int) -> bool:
         """True while the source is armed for a deep walk (plan #708 B4).