diff --git a/backend/app/sync/autotag.py b/backend/app/sync/autotag.py index e10db9e..09cce68 100644 --- a/backend/app/sync/autotag.py +++ b/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} diff --git a/backend/app/sync/playlists.py b/backend/app/sync/playlists.py index bb60489..f21ef4f 100644 --- a/backend/app/sync/playlists.py +++ b/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} diff --git a/backend/app/sync/runner.py b/backend/app/sync/runner.py index 2a1a239..ed76ebc 100644 --- a/backend/app/sync/runner.py +++ b/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)} diff --git a/backend/app/sync/videos.py b/backend/app/sync/videos.py index 0292390..4b64c51 100644 --- a/backend/app/sync/videos.py +++ b/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", {})