feat(scheduler): report progress from every job + parallelize RSS polling

Every background job now reports progress, so the dashboard can show a live
bar for any run (not only enrich/backfill/shorts/maintenance): rss_poll and
subscriptions (runner), autotag, and playlist_sync gained progress.report
calls over their loops.

RSS polling now fetches the channel feeds concurrently (16 workers) instead
of one-at-a-time — a slow/unreachable feed no longer blocks the rest, cutting
a full poll of the catalogue from minutes to seconds. Feed fetches are
network-bound and run in the pool; DB inserts stay on the session's thread
(SQLAlchemy sessions aren't thread-safe), applied as each fetch completes.
Split the DB-write half of poll_rss_channel into apply_rss_feed so the fetch
and apply phases compose cleanly.

backend/app/sync/autotag.py

@@ -13,6 +13,7 @@ from sqlalchemy.orm import Session
 
 log = logging.getLogger("subfeed.autotag")
 
+from app import progress
 from app.config import settings
 from app.models import Channel, ChannelTag, Tag, Video
 
@@ -270,13 +271,14 @@ def run_autotag_all(db: Session, only_missing: bool = False) -> dict:
         )
     channels = db.execute(query).scalars().all()
     tagged = 0
-    for channel in channels:
+    for i, channel in enumerate(channels, 1):
         try:
             apply_channel_autotags(db, channel)
             tagged += 1
         except Exception:
             db.rollback()
             log.exception("Auto-tagging failed for channel %s", channel.id)
+        progress.report(i, len(channels), "autotag")
     removed = _cleanup_orphan_system_tags(db)
     return {"channels_tagged": tagged, "orphan_tags_removed": removed}
 

backend/app/sync/playlists.py

@@ -12,7 +12,7 @@ from datetime import datetime, timezone
 from sqlalchemy import delete, select
 from sqlalchemy.orm import Session
 
-from app import quota
+from app import progress, quota
 from app.auth import has_read_scope, has_write_scope
 from app.models import Channel, OAuthToken, Playlist, PlaylistItem, User, Video
 from app.sync.videos import apply_video_details, parse_dt
@@ -382,8 +382,9 @@ def sync_all_playlists(db: Session) -> dict:
         .all()
     )
     total = 0
-    for user in users:
+    for i, user in enumerate(users, 1):
         if not has_read_scope(user):
+            progress.report(i, len(users), "playlist_sync")
             continue
         try:
             with quota.attribute(user.id, "playlist_sync"):
@@ -394,4 +395,5 @@ def sync_all_playlists(db: Session) -> dict:
         except Exception:
             db.rollback()
             log.exception("Playlist sync crashed for user %s", user.id)
+        progress.report(i, len(users), "playlist_sync")
     return {"users": len(users), "playlists_synced": total}

backend/app/sync/runner.py

@@ -5,6 +5,7 @@ user" (any user with a stored refresh token) unless a YOUTUBE_API_KEY is configu
 public reads.
 """
 import logging
+from concurrent.futures import ThreadPoolExecutor, as_completed
 
 from sqlalchemy import select
 from sqlalchemy.orm import Session
@@ -16,15 +17,19 @@ log = logging.getLogger("subfeed.sync")
 from app.models import Channel, OAuthToken, Subscription, User
 from app.sync.subscriptions import import_subscriptions
 from app.sync.videos import (
+    apply_rss_feed,
     backfill_channel_deep,
     backfill_channel_recent,
     enrich_pending,
-    poll_rss_channel,
     reconcile_full_history,
     refresh_live,
     run_shorts_classification,
 )
 from app.youtube.client import YouTubeClient
+from app.youtube.rss import fetch_channel_feed
+
+# RSS feeds are free and network-bound, so fetch many at once; DB writes stay single-threaded.
+RSS_POLL_WORKERS = 16
 
 
 def get_service_user(db: Session) -> User | None:
@@ -43,13 +48,30 @@ def get_service_user(db: Session) -> User | None:
 def run_rss_poll(db: Session, channels: list[Channel] | None = None) -> int:
     if channels is None:
         channels = db.execute(select(Channel)).scalars().all()
+    total = len(channels)
     new = 0
-    for channel in channels:
-        try:
-            new += poll_rss_channel(db, channel)
-        except Exception:
-            db.rollback()
-            log.exception("RSS poll failed for channel %s", channel.id)
+    done = 0
+    # Fetch feeds concurrently (network-bound, no DB), then apply inserts on this thread as
+    # each fetch returns — a SQLAlchemy session isn't thread-safe. Pass the channel id (a
+    # plain str) to the workers; the ORM instance is touched only here on the main thread.
+    futures = {}
+    with ThreadPoolExecutor(max_workers=RSS_POLL_WORKERS) as pool:
+        for channel in channels:
+            futures[pool.submit(fetch_channel_feed, channel.id)] = channel
+        for fut in as_completed(futures):
+            channel = futures[fut]
+            try:
+                entries = fut.result()
+            except Exception:
+                log.exception("RSS fetch failed for channel %s", channel.id)
+                entries = []
+            try:
+                new += apply_rss_feed(db, channel, entries)
+            except Exception:
+                db.rollback()
+                log.exception("RSS apply failed for channel %s", channel.id)
+            done += 1
+            progress.report(done, total, "rss_poll")
     return new
 
 
@@ -90,12 +112,13 @@ def run_subscription_resync(db: Session) -> dict:
         .scalars()
         .all()
     )
-    for user in users:
+    for i, user in enumerate(users, 1):
         try:
             import_subscriptions(db, user)
         except Exception:
             db.rollback()
             log.exception("Subscription resync failed for user %s", user.id)
+        progress.report(i, len(users), "subscriptions")
     return {"users": len(users)}
 
 

backend/app/sync/videos.py

@@ -67,8 +67,10 @@ def _insert_stubs(db: Session, rows: list[dict]) -> int:
 
 
 # --- RSS (free) ---
-def poll_rss_channel(db: Session, channel: Channel) -> int:
-    entries = fetch_channel_feed(channel.id)
+def apply_rss_feed(db: Session, channel: Channel, entries: list[dict]) -> int:
+    """Insert new video stubs from an already-fetched feed and stamp the channel's last RSS
+    time. DB-only — must run on the session's own thread; the network fetch
+    (fetch_channel_feed) is kept separate so callers like run_rss_poll can parallelise it."""
     inserted = _insert_stubs(db, entries)
     channel.last_rss_at = _now()
     db.add(channel)
@@ -76,6 +78,10 @@ def poll_rss_channel(db: Session, channel: Channel) -> int:
     return inserted
 
 
+def poll_rss_channel(db: Session, channel: Channel) -> int:
+    return apply_rss_feed(db, channel, fetch_channel_feed(channel.id))
+
+
 # --- Backfill (uploads playlist; costs quota) ---
 def _stub_from_playlist_item(item: dict, channel_id: str) -> dict | None:
     content = item.get("contentDetails", {})