feat: M2 (part 2) — RSS poller, backfill, enrichment, scheduler

- Free per-channel RSS reader for quota-less fresh-video detection
- Recent-first backfill (configurable: 100 videos / 1 year) plus resumable deep
  backfill from the uploads playlist
- Enrichment via videos.list: duration, view/like counts, category, topics,
  language, Shorts heuristic and livestream/premiere classification
- Reusable sync runners + APScheduler jobs (rss / enrich / backfill), all
  quota-aware with a reserve so backfill never starves fresh enrichment
- Manual triggers: POST /api/sync/{rss,backfill,enrich}
- Exact insert counting via RETURNING with in-batch de-duplication

README.md

@@ -9,8 +9,12 @@ Each user signs in with their own Google account (invite-only) and sees only the
 subscriptions. All the expensive data (channels, videos, metadata) is fetched from YouTube
 once and stored locally, so filtering/searching/sorting are instant and don't burn API quota.
 
-> Status: early development. Milestone **M1** (foundation) is in place: docker-compose stack,
-> FastAPI backend, Google OAuth login with an email invite-list, and encrypted token storage.
+> Status: early development.
+> - **M1** (foundation): docker-compose stack, FastAPI backend, Google OAuth login with an
+>   email invite-list, encrypted token storage.
+> - **M2** (ingest core): subscription import, free RSS detection, recent-first + deep backfill
+>   from the uploads playlist, enrichment (duration/stats/category, Shorts & livestream
+>   classification), a shared daily quota guard, and a background scheduler.
 
 ## Requirements
 

backend/app/config.py

@@ -31,9 +31,23 @@ class Settings(BaseSettings):
     # Daily quota budget (units). Default leaves headroom under the 10,000/day free limit.
     quota_daily_budget: int = 9000
     # Recent-first backfill: how far back to fetch on the first pass per channel.
-    backfill_recent_max_videos: int = 200
+    backfill_recent_max_videos: int = 100
     backfill_recent_max_days: int = 365
 
+    # Videos at or below this duration (seconds) are treated as Shorts.
+    shorts_max_seconds: int = 60
+    # videos.list accepts up to 50 ids per call.
+    enrich_batch_size: int = 50
+
+    # --- Background scheduler ---
+    scheduler_enabled: bool = True
+    rss_poll_minutes: int = 20
+    enrich_interval_minutes: int = 3
+    backfill_interval_minutes: int = 10
+    # Keep this many quota units in reserve so scheduled backfill never starves
+    # interactive syncs / enrichment of fresh videos.
+    backfill_quota_reserve: int = 2000
+
     @property
     def allowed_email_set(self) -> set[str]:
         return {e.strip().lower() for e in self.allowed_emails.split(",") if e.strip()}

backend/app/main.py

@@ -1,3 +1,4 @@
+from contextlib import asynccontextmanager
 from pathlib import Path
 
 from fastapi import FastAPI
@@ -9,8 +10,19 @@ from starlette.middleware.sessions import SessionMiddleware
 from app import auth
 from app.config import settings
 from app.routes import health, sync
+from app.scheduler import shutdown_scheduler, start_scheduler
 
-app = FastAPI(title=settings.app_name)
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    start_scheduler()
+    try:
+        yield
+    finally:
+        shutdown_scheduler()
+
+
+app = FastAPI(title=settings.app_name, lifespan=lifespan)
 
 app.add_middleware(
     SessionMiddleware,

backend/app/routes/sync.py

@@ -6,11 +6,24 @@ from app import quota
 from app.auth import current_user
 from app.db import get_db
 from app.models import Channel, Subscription, User, Video
+from app.sync.runner import run_enrich, run_recent_backfill, run_rss_poll
 from app.sync.subscriptions import import_subscriptions
 
 router = APIRouter(prefix="/api/sync", tags=["sync"])
 
 
+def _user_channels(db: Session, user: User) -> list[Channel]:
+    return (
+        db.execute(
+            select(Channel)
+            .join(Subscription, Subscription.channel_id == Channel.id)
+            .where(Subscription.user_id == user.id)
+        )
+        .scalars()
+        .all()
+    )
+
+
 @router.post("/subscriptions")
 def sync_subscriptions(
     user: User = Depends(current_user), db: Session = Depends(get_db)
@@ -22,6 +35,38 @@ def sync_subscriptions(
     return result
 
 
+@router.post("/rss")
+def sync_rss(
+    user: User = Depends(current_user), db: Session = Depends(get_db)
+) -> dict:
+    new = run_rss_poll(db, _user_channels(db, user))
+    return {"new_videos": new}
+
+
+@router.post("/backfill")
+def sync_backfill(
+    user: User = Depends(current_user),
+    db: Session = Depends(get_db),
+    max_channels: int = 25,
+) -> dict:
+    before = quota.units_used_today(db)
+    result = run_recent_backfill(db, _user_channels(db, user), max_channels=max_channels)
+    result["quota_used_estimate"] = quota.units_used_today(db) - before
+    return result
+
+
+@router.post("/enrich")
+def sync_enrich(
+    user: User = Depends(current_user), db: Session = Depends(get_db)
+) -> dict:
+    before = quota.units_used_today(db)
+    enriched = run_enrich(db)
+    return {
+        "enriched": enriched,
+        "quota_used_estimate": quota.units_used_today(db) - before,
+    }
+
+
 @router.get("/status")
 def sync_status(
     user: User = Depends(current_user), db: Session = Depends(get_db)

backend/app/scheduler.py

@@ -0,0 +1,77 @@
+"""Background scheduler: free RSS detection, enrichment of new videos, and quota-aware
+recent-first / deep backfill. Runs inside the API process (single worker)."""
+import logging
+
+from apscheduler.schedulers.background import BackgroundScheduler
+
+from app.config import settings
+from app.db import SessionLocal
+from app.sync.runner import (
+    run_deep_backfill,
+    run_enrich,
+    run_recent_backfill,
+    run_rss_poll,
+)
+
+logger = logging.getLogger("subfeed.scheduler")
+
+_scheduler: BackgroundScheduler | None = None
+
+
+def _job(name: str, fn) -> None:
+    db = SessionLocal()
+    try:
+        result = fn(db)
+        logger.info("job %s -> %s", name, result)
+    except Exception:
+        db.rollback()
+        logger.exception("job %s failed", name)
+    finally:
+        db.close()
+
+
+def _rss_job() -> None:
+    _job("rss_poll", lambda db: run_rss_poll(db))
+
+
+def _enrich_job() -> None:
+    _job("enrich", lambda db: run_enrich(db))
+
+
+def _backfill_job() -> None:
+    # Recent-first for not-yet-synced channels, then deep backfill for the rest.
+    def work(db):
+        recent = run_recent_backfill(db, max_channels=25)
+        deep = run_deep_backfill(db, max_channels=10)
+        return {"recent": recent, "deep": deep}
+
+    _job("backfill", work)
+
+
+def start_scheduler() -> None:
+    global _scheduler
+    if not settings.scheduler_enabled or _scheduler is not None:
+        return
+    scheduler = BackgroundScheduler(timezone="UTC")
+    scheduler.add_job(
+        _rss_job, "interval", minutes=settings.rss_poll_minutes, id="rss_poll"
+    )
+    scheduler.add_job(
+        _enrich_job, "interval", minutes=settings.enrich_interval_minutes, id="enrich"
+    )
+    scheduler.add_job(
+        _backfill_job,
+        "interval",
+        minutes=settings.backfill_interval_minutes,
+        id="backfill",
+    )
+    scheduler.start()
+    _scheduler = scheduler
+    logger.info("scheduler started")
+
+
+def shutdown_scheduler() -> None:
+    global _scheduler
+    if _scheduler is not None:
+        _scheduler.shutdown(wait=False)
+        _scheduler = None

backend/app/sync/runner.py

@@ -0,0 +1,116 @@
+"""Reusable sync routines shared by the scheduler and the manual /api/sync endpoints.
+
+Channel/video data is shared across users, so background work acts through a "service
+user" (any user with a stored refresh token) unless a YOUTUBE_API_KEY is configured for
+public reads.
+"""
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app import quota
+from app.config import settings
+from app.models import Channel, OAuthToken, User
+from app.sync.videos import (
+    backfill_channel_deep,
+    backfill_channel_recent,
+    enrich_pending,
+    poll_rss_channel,
+)
+from app.youtube.client import YouTubeClient
+
+
+def get_service_user(db: Session) -> User | None:
+    return (
+        db.execute(
+            select(User)
+            .join(OAuthToken)
+            .where(OAuthToken.refresh_token_enc.is_not(None))
+            .order_by(User.id)
+        )
+        .scalars()
+        .first()
+    )
+
+
+def run_rss_poll(db: Session, channels: list[Channel] | None = None) -> int:
+    if channels is None:
+        channels = db.execute(select(Channel)).scalars().all()
+    new = 0
+    for channel in channels:
+        try:
+            new += poll_rss_channel(db, channel)
+        except Exception:
+            db.rollback()
+    return new
+
+
+def run_enrich(db: Session, max_batches: int = 200, floor: int = 200) -> int:
+    user = get_service_user(db)
+    if user is None:
+        return 0
+    total = 0
+    with YouTubeClient(db, user) as yt:
+        for _ in range(max_batches):
+            if quota.remaining_today(db) <= floor:
+                break
+            n = enrich_pending(db, yt)
+            total += n
+            if n == 0:
+                break
+    return total
+
+
+def run_recent_backfill(
+    db: Session, channels: list[Channel] | None = None, max_channels: int | None = None
+) -> dict:
+    user = get_service_user(db)
+    if user is None:
+        return {"channels_processed": 0, "videos_added": 0, "reason": "no service user"}
+    if channels is None:
+        channels = (
+            db.execute(select(Channel).where(Channel.recent_synced_at.is_(None)))
+            .scalars()
+            .all()
+        )
+    if max_channels:
+        channels = channels[:max_channels]
+    processed = 0
+    videos_added = 0
+    with YouTubeClient(db, user) as yt:
+        for channel in channels:
+            if quota.remaining_today(db) <= settings.backfill_quota_reserve:
+                break
+            try:
+                videos_added += backfill_channel_recent(db, yt, channel)
+                processed += 1
+            except Exception:
+                db.rollback()
+    return {"channels_processed": processed, "videos_added": videos_added}
+
+
+def run_deep_backfill(db: Session, max_channels: int = 10, max_pages: int = 5) -> dict:
+    user = get_service_user(db)
+    if user is None:
+        return {"channels_processed": 0, "videos_added": 0, "reason": "no service user"}
+    channels = (
+        db.execute(
+            select(Channel).where(
+                Channel.backfill_done.is_(False),
+                Channel.recent_synced_at.is_not(None),
+            )
+        )
+        .scalars()
+        .all()
+    )[:max_channels]
+    processed = 0
+    videos_added = 0
+    with YouTubeClient(db, user) as yt:
+        for channel in channels:
+            if quota.remaining_today(db) <= settings.backfill_quota_reserve:
+                break
+            try:
+                videos_added += backfill_channel_deep(db, yt, channel, max_pages)
+                processed += 1
+            except Exception:
+                db.rollback()
+    return {"channels_processed": processed, "videos_added": videos_added}

backend/app/sync/videos.py

@@ -0,0 +1,227 @@
+"""Video ingestion: free RSS detection, recent-first (and deep) backfill from the
+uploads playlist, and enrichment via videos.list (duration, stats, category, Shorts
+and livestream classification)."""
+import re
+from datetime import datetime, timedelta, timezone
+
+from sqlalchemy import select
+from sqlalchemy.dialects.postgresql import insert as pg_insert
+from sqlalchemy.orm import Session
+
+from app.config import settings
+from app.models import Channel, Video
+from app.youtube.client import YouTubeClient, best_thumbnail
+from app.youtube.rss import fetch_channel_feed
+
+_DURATION_RE = re.compile(
+    r"P(?:(\d+)D)?T(?:(\d+)H)?(?:(\d+)M)?(?:(\d+)S)?", re.IGNORECASE
+)
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def parse_iso8601_duration(value: str | None) -> int | None:
+    if not value:
+        return None
+    match = _DURATION_RE.fullmatch(value)
+    if not match:
+        return None
+    days, hours, minutes, seconds = (int(g) if g else 0 for g in match.groups())
+    return days * 86400 + hours * 3600 + minutes * 60 + seconds
+
+
+def parse_dt(value: str | None) -> datetime | None:
+    if not value:
+        return None
+    try:
+        return datetime.fromisoformat(value.replace("Z", "+00:00"))
+    except ValueError:
+        return None
+
+
+def _insert_stubs(db: Session, rows: list[dict]) -> int:
+    """Idempotently insert video stubs; returns the number of newly inserted rows.
+
+    Uses RETURNING so the count is exact (ON CONFLICT DO NOTHING leaves rowcount
+    unreliable across drivers). Duplicates within the batch are dropped first so a
+    single INSERT can't raise CardinalityViolation on the conflict target."""
+    if not rows:
+        return 0
+    seen: dict[str, dict] = {}
+    for row in rows:
+        seen[row["id"]] = row
+    stmt = (
+        pg_insert(Video)
+        .values(list(seen.values()))
+        .on_conflict_do_nothing(index_elements=[Video.id])
+        .returning(Video.id)
+    )
+    inserted = len(db.execute(stmt).fetchall())
+    db.commit()
+    return inserted
+
+
+# --- RSS (free) ---
+def poll_rss_channel(db: Session, channel: Channel) -> int:
+    entries = fetch_channel_feed(channel.id)
+    inserted = _insert_stubs(db, entries)
+    channel.last_rss_at = _now()
+    db.add(channel)
+    db.commit()
+    return inserted
+
+
+# --- Backfill (uploads playlist; costs quota) ---
+def _stub_from_playlist_item(item: dict, channel_id: str) -> dict | None:
+    content = item.get("contentDetails", {})
+    snippet = item.get("snippet", {})
+    video_id = content.get("videoId")
+    if not video_id:
+        return None
+    published = parse_dt(content.get("videoPublishedAt") or snippet.get("publishedAt"))
+    return {
+        "id": video_id,
+        "channel_id": channel_id,
+        "title": snippet.get("title"),
+        "published_at": published,
+        "thumbnail_url": best_thumbnail(snippet.get("thumbnails")),
+    }
+
+
+def backfill_channel_recent(db: Session, yt: YouTubeClient, channel: Channel) -> int:
+    """First pass: newest videos up to the configured count or age cutoff. Records a
+    resume cursor for the later deep backfill."""
+    if not channel.uploads_playlist_id:
+        channel.recent_synced_at = _now()
+        db.add(channel)
+        db.commit()
+        return 0
+
+    cutoff = _now() - timedelta(days=settings.backfill_recent_max_days)
+    collected: list[dict] = []
+    page_token: str | None = None
+    next_cursor: str | None = None
+    done = False
+
+    while len(collected) < settings.backfill_recent_max_videos:
+        data = yt.get_playlist_items(channel.uploads_playlist_id, page_token)
+        reached_cutoff = False
+        for item in data.get("items", []):
+            stub = _stub_from_playlist_item(item, channel.id)
+            if stub is None:
+                continue
+            if stub["published_at"] and stub["published_at"] < cutoff:
+                reached_cutoff = True
+                break
+            collected.append(stub)
+            if len(collected) >= settings.backfill_recent_max_videos:
+                break
+        page_token = data.get("nextPageToken")
+        if reached_cutoff or not page_token:
+            next_cursor = page_token
+            done = not page_token
+            break
+        next_cursor = page_token
+
+    channel.recent_synced_at = _now()
+    channel.backfill_cursor = next_cursor
+    channel.backfill_done = done
+    inserted = _insert_stubs(db, collected)
+    db.add(channel)
+    db.commit()
+    return inserted
+
+
+def backfill_channel_deep(
+    db: Session, yt: YouTubeClient, channel: Channel, max_pages: int = 5
+) -> int:
+    """Continue paging older uploads from the stored cursor (bounded per run)."""
+    if channel.backfill_done or not channel.uploads_playlist_id:
+        return 0
+    page_token = channel.backfill_cursor
+    total = 0
+    for _ in range(max_pages):
+        data = yt.get_playlist_items(channel.uploads_playlist_id, page_token)
+        rows = [
+            s
+            for s in (
+                _stub_from_playlist_item(it, channel.id) for it in data.get("items", [])
+            )
+            if s is not None
+        ]
+        total += _insert_stubs(db, rows)
+        page_token = data.get("nextPageToken")
+        channel.backfill_cursor = page_token
+        if not page_token:
+            channel.backfill_done = True
+            break
+    db.add(channel)
+    db.commit()
+    return total
+
+
+# --- Enrichment (videos.list) ---
+def _live_status(snippet: dict, live_details: dict | None) -> str:
+    broadcast = snippet.get("liveBroadcastContent")
+    if broadcast == "live":
+        return "live"
+    if broadcast == "upcoming":
+        return "upcoming"
+    if live_details and live_details.get("actualEndTime"):
+        return "was_live"
+    return "none"
+
+
+def apply_video_details(video: Video, item: dict) -> None:
+    snippet = item.get("snippet", {})
+    content = item.get("contentDetails", {})
+    stats = item.get("statistics", {})
+    topics = item.get("topicDetails", {})
+    live = item.get("liveStreamingDetails")
+
+    video.title = snippet.get("title") or video.title
+    video.description = snippet.get("description")
+    if not video.published_at:
+        video.published_at = parse_dt(snippet.get("publishedAt"))
+    video.thumbnail_url = best_thumbnail(snippet.get("thumbnails")) or video.thumbnail_url
+    video.duration_seconds = parse_iso8601_duration(content.get("duration"))
+    video.view_count = int(stats["viewCount"]) if stats.get("viewCount") else None
+    video.like_count = int(stats["likeCount"]) if stats.get("likeCount") else None
+    video.category_id = int(snippet["categoryId"]) if snippet.get("categoryId") else None
+    video.topic_categories = topics.get("topicCategories")
+    video.default_language = snippet.get("defaultLanguage") or snippet.get(
+        "defaultAudioLanguage"
+    )
+    video.live_status = _live_status(snippet, live)
+    video.is_short = bool(
+        video.live_status == "none"
+        and video.duration_seconds
+        and 0 < video.duration_seconds <= settings.shorts_max_seconds
+    )
+
+
+def enrich_pending(db: Session, yt: YouTubeClient, limit: int | None = None) -> int:
+    limit = limit or settings.enrich_batch_size
+    videos = (
+        db.execute(select(Video).where(Video.enriched_at.is_(None)).limit(limit))
+        .scalars()
+        .all()
+    )
+    if not videos:
+        return 0
+    items = {it["id"]: it for it in yt.get_videos([v.id for v in videos])}
+    now = _now()
+    enriched = 0
+    for video in videos:
+        item = items.get(video.id)
+        if item is None:
+            # Unavailable (deleted/private): stamp so we don't keep retrying.
+            video.enriched_at = now
+            continue
+        apply_video_details(video, item)
+        video.enriched_at = now
+        enriched += 1
+    db.commit()
+    return enriched

backend/app/youtube/client.py

@@ -133,3 +133,29 @@ class YouTubeClient:
             )
             items.extend(data.get("items", []))
         return items
+
+    def get_playlist_items(self, playlist_id: str, page_token: str | None = None) -> dict:
+        """One page (up to 50) of a playlist's items, most-recent-first for uploads
+        playlists. Returns the raw response (items + nextPageToken)."""
+        params = {
+            "part": "contentDetails,snippet",
+            "playlistId": playlist_id,
+            "maxResults": 50,
+        }
+        if page_token:
+            params["pageToken"] = page_token
+        return self._get("playlistItems", params)
+
+    def get_videos(self, video_ids: list[str]) -> list[dict]:
+        items: list[dict] = []
+        for batch in _chunks(video_ids, 50):
+            data = self._get(
+                "videos",
+                {
+                    "part": "snippet,contentDetails,statistics,topicDetails,liveStreamingDetails",
+                    "id": ",".join(batch),
+                    "maxResults": 50,
+                },
+            )
+            items.extend(data.get("items", []))
+        return items

backend/app/youtube/rss.py

@@ -0,0 +1,42 @@
+"""Free per-channel RSS feed reader (no API quota). Returns the channel's most recent
+~15 uploads as lightweight stubs for fast fresh-video detection."""
+from datetime import datetime, timezone
+
+import feedparser
+import httpx
+
+RSS_URL = "https://www.youtube.com/feeds/videos.xml?channel_id={channel_id}"
+
+
+def fetch_channel_feed(channel_id: str) -> list[dict]:
+    url = RSS_URL.format(channel_id=channel_id)
+    try:
+        resp = httpx.get(url, timeout=20.0, headers={"User-Agent": "Subfeed/1.0"})
+    except httpx.HTTPError:
+        return []
+    if resp.status_code != 200:
+        return []
+
+    parsed = feedparser.parse(resp.content)
+    out: list[dict] = []
+    for entry in parsed.entries:
+        video_id = entry.get("yt_videoid")
+        if not video_id:
+            continue
+        published = None
+        if entry.get("published_parsed"):
+            published = datetime(*entry.published_parsed[:6], tzinfo=timezone.utc)
+        thumbnail = None
+        media = entry.get("media_thumbnail") or []
+        if media:
+            thumbnail = media[0].get("url")
+        out.append(
+            {
+                "id": video_id,
+                "channel_id": channel_id,
+                "title": entry.get("title"),
+                "published_at": published,
+                "thumbnail_url": thumbnail,
+            }
+        )
+    return out

backend/requirements.txt

@@ -7,5 +7,7 @@ pydantic>=2.7,<3.0
 pydantic-settings>=2.3,<3.0
 authlib>=1.3,<2.0
 httpx>=0.27,<1.0
+feedparser>=6.0,<7.0
+apscheduler>=3.10,<4.0
 itsdangerous>=2.1,<3.0
 cryptography>=42,<46