merge: M2 ingest core

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/alembic/versions/0002_ingest_core.py

@@ -0,0 +1,127 @@
+"""ingest core: channels, subscriptions, videos, api_quota_usage
+
+Revision ID: 0002_ingest_core
+Revises: 0001_initial
+Create Date: 2026-06-11
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+revision: str = "0002_ingest_core"
+down_revision: Union[str, None] = "0001_initial"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "channels",
+        sa.Column("id", sa.String(length=32), primary_key=True),
+        sa.Column("title", sa.String(length=255), nullable=True),
+        sa.Column("handle", sa.String(length=255), nullable=True),
+        sa.Column("description", sa.Text(), nullable=True),
+        sa.Column("thumbnail_url", sa.String(length=1024), nullable=True),
+        sa.Column("uploads_playlist_id", sa.String(length=64), nullable=True),
+        sa.Column("subscriber_count", sa.BigInteger(), nullable=True),
+        sa.Column("video_count", sa.BigInteger(), nullable=True),
+        sa.Column("topic_categories", sa.JSON(), nullable=True),
+        sa.Column("default_language", sa.String(length=16), nullable=True),
+        sa.Column("country", sa.String(length=8), nullable=True),
+        sa.Column("details_synced_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("recent_synced_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("last_rss_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("backfill_cursor", sa.String(length=128), nullable=True),
+        sa.Column("backfill_done", sa.Boolean(), nullable=False, server_default="false"),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.func.now(),
+            nullable=False,
+        ),
+    )
+
+    op.create_table(
+        "subscriptions",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column(
+            "user_id",
+            sa.Integer(),
+            sa.ForeignKey("users.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column(
+            "channel_id",
+            sa.String(length=32),
+            sa.ForeignKey("channels.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column("yt_subscription_id", sa.String(length=128), nullable=True),
+        sa.Column("priority", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("hidden", sa.Boolean(), nullable=False, server_default="false"),
+        sa.Column("subscribed_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.func.now(),
+            nullable=False,
+        ),
+        sa.UniqueConstraint("user_id", "channel_id", name="uq_user_channel"),
+    )
+    op.create_index("ix_subscriptions_user_id", "subscriptions", ["user_id"])
+    op.create_index("ix_subscriptions_channel_id", "subscriptions", ["channel_id"])
+
+    op.create_table(
+        "videos",
+        sa.Column("id", sa.String(length=16), primary_key=True),
+        sa.Column(
+            "channel_id",
+            sa.String(length=32),
+            sa.ForeignKey("channels.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column("title", sa.Text(), nullable=True),
+        sa.Column("description", sa.Text(), nullable=True),
+        sa.Column("published_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("thumbnail_url", sa.String(length=1024), nullable=True),
+        sa.Column("duration_seconds", sa.Integer(), nullable=True),
+        sa.Column("view_count", sa.BigInteger(), nullable=True),
+        sa.Column("like_count", sa.BigInteger(), nullable=True),
+        sa.Column("category_id", sa.Integer(), nullable=True),
+        sa.Column("topic_categories", sa.JSON(), nullable=True),
+        sa.Column("default_language", sa.String(length=16), nullable=True),
+        sa.Column("detected_language", sa.String(length=16), nullable=True),
+        sa.Column("is_short", sa.Boolean(), nullable=False, server_default="false"),
+        sa.Column("live_status", sa.String(length=16), nullable=False, server_default="none"),
+        sa.Column("enriched_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.func.now(),
+            nullable=False,
+        ),
+    )
+    op.create_index("ix_videos_channel_id", "videos", ["channel_id"])
+    op.create_index("ix_videos_published_at", "videos", ["published_at"])
+    op.create_index("ix_videos_is_short", "videos", ["is_short"])
+    op.create_index("ix_videos_live_status", "videos", ["live_status"])
+
+    op.create_table(
+        "api_quota_usage",
+        sa.Column("day", sa.Date(), primary_key=True),
+        sa.Column("units_used", sa.Integer(), nullable=False, server_default="0"),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("api_quota_usage")
+    op.drop_index("ix_videos_live_status", table_name="videos")
+    op.drop_index("ix_videos_is_short", table_name="videos")
+    op.drop_index("ix_videos_published_at", table_name="videos")
+    op.drop_index("ix_videos_channel_id", table_name="videos")
+    op.drop_table("videos")
+    op.drop_index("ix_subscriptions_channel_id", table_name="subscriptions")
+    op.drop_index("ix_subscriptions_user_id", table_name="subscriptions")
+    op.drop_table("subscriptions")
+    op.drop_table("channels")

backend/app/config.py

@@ -24,6 +24,30 @@ class Settings(BaseSettings):
     # Origin of a separately served frontend dev server (enables CORS). Empty in production.
     frontend_origin: str = ""
 
+    # --- Sync / YouTube Data API ---
+    # Optional API key for public reads (channels/videos/playlistItems). When set it is
+    # preferred for shared backfill/enrichment so it doesn't depend on a specific user.
+    youtube_api_key: str = ""
+    # 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 = 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
@@ -8,9 +9,20 @@ from starlette.middleware.sessions import SessionMiddleware
 
 from app import auth
 from app.config import settings
-from app.routes import health
+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,
@@ -30,6 +42,7 @@ if settings.frontend_origin:
 
 app.include_router(health.router)
 app.include_router(auth.router)
+app.include_router(sync.router)
 
 STATIC_DIR = Path(__file__).parent / "static"
 app.mount("/assets", StaticFiles(directory=STATIC_DIR / "assets"), name="assets")

backend/app/models.py

@@ -1,6 +1,18 @@
-from datetime import datetime
+from datetime import date, datetime
 
-from sqlalchemy import DateTime, ForeignKey, String, func
+from sqlalchemy import (
+    BigInteger,
+    Boolean,
+    Date,
+    DateTime,
+    ForeignKey,
+    Integer,
+    JSON,
+    String,
+    Text,
+    UniqueConstraint,
+    func,
+)
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from app.db import Base
@@ -22,6 +34,9 @@ class User(Base):
     token: Mapped["OAuthToken | None"] = relationship(
         back_populates="user", uselist=False, cascade="all, delete-orphan"
     )
+    subscriptions: Mapped[list["Subscription"]] = relationship(
+        back_populates="user", cascade="all, delete-orphan"
+    )
 
 
 class OAuthToken(Base):
@@ -41,3 +56,110 @@ class OAuthToken(Base):
     )
 
     user: Mapped["User"] = relationship(back_populates="token")
+
+
+class Channel(Base):
+    """A YouTube channel. Shared across all users (one channel's videos are the same
+    for everyone), so its expensive metadata is fetched and stored only once."""
+
+    __tablename__ = "channels"
+
+    id: Mapped[str] = mapped_column(String(32), primary_key=True)  # "UC..." channel id
+    title: Mapped[str | None] = mapped_column(String(255))
+    handle: Mapped[str | None] = mapped_column(String(255))
+    description: Mapped[str | None] = mapped_column(Text)
+    thumbnail_url: Mapped[str | None] = mapped_column(String(1024))
+    uploads_playlist_id: Mapped[str | None] = mapped_column(String(64))
+    subscriber_count: Mapped[int | None] = mapped_column(BigInteger)
+    video_count: Mapped[int | None] = mapped_column(BigInteger)
+    topic_categories: Mapped[list | None] = mapped_column(JSON)
+    default_language: Mapped[str | None] = mapped_column(String(16))
+    country: Mapped[str | None] = mapped_column(String(8))
+
+    # Sync bookkeeping.
+    details_synced_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    recent_synced_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    last_rss_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    backfill_cursor: Mapped[str | None] = mapped_column(String(128))
+    backfill_done: Mapped[bool] = mapped_column(
+        Boolean, default=False, server_default="false"
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now()
+    )
+
+    videos: Mapped[list["Video"]] = relationship(back_populates="channel")
+    subscriptions: Mapped[list["Subscription"]] = relationship(back_populates="channel")
+
+
+class Subscription(Base):
+    """Per-user link to a channel, with the user's own priority/visibility overrides."""
+
+    __tablename__ = "subscriptions"
+    __table_args__ = (UniqueConstraint("user_id", "channel_id", name="uq_user_channel"),)
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    user_id: Mapped[int] = mapped_column(
+        ForeignKey("users.id", ondelete="CASCADE"), index=True
+    )
+    channel_id: Mapped[str] = mapped_column(
+        ForeignKey("channels.id", ondelete="CASCADE"), index=True
+    )
+    # YouTube's subscription resource id, needed to unsubscribe via the API.
+    yt_subscription_id: Mapped[str | None] = mapped_column(String(128))
+    priority: Mapped[int] = mapped_column(Integer, default=0, server_default="0")
+    hidden: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false")
+    subscribed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now()
+    )
+
+    user: Mapped["User"] = relationship(back_populates="subscriptions")
+    channel: Mapped["Channel"] = relationship(back_populates="subscriptions")
+
+
+class Video(Base):
+    """A YouTube video. Shared across users; per-user state lives elsewhere."""
+
+    __tablename__ = "videos"
+
+    id: Mapped[str] = mapped_column(String(16), primary_key=True)  # video id
+    channel_id: Mapped[str] = mapped_column(
+        ForeignKey("channels.id", ondelete="CASCADE"), index=True
+    )
+    title: Mapped[str | None] = mapped_column(Text)
+    description: Mapped[str | None] = mapped_column(Text)
+    published_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), index=True
+    )
+    thumbnail_url: Mapped[str | None] = mapped_column(String(1024))
+    duration_seconds: Mapped[int | None] = mapped_column(Integer)
+    view_count: Mapped[int | None] = mapped_column(BigInteger)
+    like_count: Mapped[int | None] = mapped_column(BigInteger)
+    category_id: Mapped[int | None] = mapped_column(Integer)
+    topic_categories: Mapped[list | None] = mapped_column(JSON)
+    default_language: Mapped[str | None] = mapped_column(String(16))
+    detected_language: Mapped[str | None] = mapped_column(String(16))
+    is_short: Mapped[bool] = mapped_column(
+        Boolean, default=False, server_default="false", index=True
+    )
+    # none | live | upcoming | premiere | was_live
+    live_status: Mapped[str] = mapped_column(
+        String(16), default="none", server_default="none", index=True
+    )
+    enriched_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now()
+    )
+
+    channel: Mapped["Channel"] = relationship(back_populates="videos")
+
+
+class ApiQuotaUsage(Base):
+    """Tracks YouTube Data API units spent per Pacific-time day (the quota resets at
+    midnight Pacific). The whole app shares one daily budget."""
+
+    __tablename__ = "api_quota_usage"
+
+    day: Mapped[date] = mapped_column(Date, primary_key=True)
+    units_used: Mapped[int] = mapped_column(Integer, default=0, server_default="0")

backend/app/quota.py

@@ -0,0 +1,50 @@
+"""Central YouTube Data API quota guard.
+
+The whole app shares one daily budget (the API resets at midnight US Pacific time).
+Steady-state work (RSS detection is free; enrichment of new videos is cheap) should
+always be allowed; expensive backfill must yield to the remaining budget.
+"""
+from datetime import datetime
+from zoneinfo import ZoneInfo
+
+from sqlalchemy.dialects.postgresql import insert as pg_insert
+from sqlalchemy.orm import Session
+
+from app.config import settings
+from app.models import ApiQuotaUsage
+
+_PACIFIC = ZoneInfo("America/Los_Angeles")
+
+
+def pacific_today():
+    return datetime.now(_PACIFIC).date()
+
+
+def units_used_today(db: Session) -> int:
+    row = db.get(ApiQuotaUsage, pacific_today())
+    return row.units_used if row else 0
+
+
+def remaining_today(db: Session) -> int:
+    return max(0, settings.quota_daily_budget - units_used_today(db))
+
+
+def can_spend(db: Session, units: int) -> bool:
+    return remaining_today(db) >= units
+
+
+def record_usage(db: Session, units: int) -> None:
+    """Atomically add `units` to today's counter (upsert)."""
+    if units <= 0:
+        return
+    day = pacific_today()
+    stmt = (
+        pg_insert(ApiQuotaUsage)
+        .values(day=day, units_used=units)
+        .on_conflict_do_update(
+            index_elements=[ApiQuotaUsage.day],
+            set_={"units_used": ApiQuotaUsage.units_used + units},
+        )
+    )
+    db.execute(stmt)
+    db.commit()

backend/app/routes/sync.py

@@ -0,0 +1,85 @@
+from fastapi import APIRouter, Depends
+from sqlalchemy import func, select
+from sqlalchemy.orm import Session
+
+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)
+) -> dict:
+    before = quota.units_used_today(db)
+    result = import_subscriptions(db, user)
+    result["quota_used_estimate"] = quota.units_used_today(db) - before
+    result["quota_remaining_today"] = quota.remaining_today(db)
+    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)
+) -> dict:
+    sub_count = db.scalar(
+        select(func.count())
+        .select_from(Subscription)
+        .where(Subscription.user_id == user.id)
+    )
+    return {
+        "subscriptions": sub_count,
+        "channels_total": db.scalar(select(func.count()).select_from(Channel)),
+        "videos_total": db.scalar(select(func.count()).select_from(Video)),
+        "quota_used_today": quota.units_used_today(db),
+        "quota_remaining_today": quota.remaining_today(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/subscriptions.py

@@ -0,0 +1,115 @@
+"""Import the authenticated user's YouTube subscriptions and channel metadata."""
+from datetime import datetime, timezone
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.models import Channel, Subscription, User
+from app.youtube.client import YouTubeClient, best_thumbnail
+
+
+def _to_int(value) -> int | None:
+    try:
+        return int(value) if value is not None else None
+    except (TypeError, ValueError):
+        return None
+
+
+def apply_channel_details(db: Session, items: list[dict]) -> None:
+    now = datetime.now(timezone.utc)
+    for item in items:
+        channel = db.get(Channel, item["id"])
+        if channel is None:
+            continue
+        snippet = item.get("snippet", {})
+        content = item.get("contentDetails", {})
+        stats = item.get("statistics", {})
+        topics = item.get("topicDetails", {})
+
+        channel.title = snippet.get("title") or channel.title
+        channel.handle = snippet.get("customUrl")
+        channel.description = snippet.get("description")
+        channel.thumbnail_url = (
+            best_thumbnail(snippet.get("thumbnails")) or channel.thumbnail_url
+        )
+        channel.default_language = snippet.get("defaultLanguage")
+        channel.country = snippet.get("country")
+        channel.uploads_playlist_id = content.get("relatedPlaylists", {}).get("uploads")
+        channel.subscriber_count = _to_int(stats.get("subscriberCount"))
+        channel.video_count = _to_int(stats.get("videoCount"))
+        channel.topic_categories = topics.get("topicCategories")
+        channel.details_synced_at = now
+
+
+def import_subscriptions(db: Session, user: User) -> dict:
+    """Pull all subscriptions for `user`, upsert channels + subscription links, prune
+    channels the user has unsubscribed from on YouTube, and fetch channel details for
+    any channel we don't have details for yet."""
+    fetched: dict[str, dict] = {}
+    new_channels = 0
+
+    with YouTubeClient(db, user) as yt:
+        for sub in yt.iter_subscriptions():
+            cid = sub.get("channel_id")
+            if cid:
+                fetched[cid] = sub
+
+        for cid, sub in fetched.items():
+            channel = db.get(Channel, cid)
+            if channel is None:
+                channel = Channel(
+                    id=cid,
+                    title=sub.get("title"),
+                    description=sub.get("description"),
+                    thumbnail_url=sub.get("thumbnail_url"),
+                )
+                db.add(channel)
+                new_channels += 1
+            sub_row = db.execute(
+                select(Subscription).where(
+                    Subscription.user_id == user.id, Subscription.channel_id == cid
+                )
+            ).scalar_one_or_none()
+            if sub_row is None:
+                sub_row = Subscription(user_id=user.id, channel_id=cid)
+                db.add(sub_row)
+            sub_row.yt_subscription_id = sub.get("yt_subscription_id")
+        db.flush()
+
+        # Prune subscriptions removed on YouTube.
+        removed = 0
+        existing = (
+            db.execute(select(Subscription).where(Subscription.user_id == user.id))
+            .scalars()
+            .all()
+        )
+        for sub_row in existing:
+            if sub_row.channel_id not in fetched:
+                db.delete(sub_row)
+                removed += 1
+        db.commit()
+
+        # Fetch details for channels that don't have them yet.
+        need_details = (
+            db.execute(
+                select(Channel).where(
+                    Channel.id.in_(fetched.keys()),
+                    Channel.details_synced_at.is_(None),
+                )
+            )
+            .scalars()
+            .all()
+        )
+        detailed = 0
+        if need_details:
+            items = yt.get_channels([c.id for c in need_details])
+            apply_channel_details(db, items)
+            detailed = len(items)
+        db.commit()
+
+    return {
+        "subscriptions": len(fetched),
+        "channels_new": new_channels,
+        "channels_detailed": detailed,
+        "removed_stale": removed,
+    }

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

@@ -0,0 +1,161 @@
+"""Thin synchronous YouTube Data API v3 client bound to a user's OAuth token.
+
+Each list call costs 1 quota unit and is recorded via the central quota guard.
+Public reads (channels/videos/playlistItems) use the configured API key when available
+so they don't depend on a specific user's token; subscriptions.list?mine=true always
+uses OAuth.
+"""
+from collections.abc import Iterator
+from datetime import datetime, timedelta, timezone
+
+import httpx
+
+from app import quota
+from app.config import settings
+from app.models import User
+from app.security import decrypt
+
+GOOGLE_TOKEN_URL = "https://oauth2.googleapis.com/token"
+API_BASE = "https://www.googleapis.com/youtube/v3"
+
+
+class YouTubeError(Exception):
+    pass
+
+
+def _chunks(items: list, size: int) -> Iterator[list]:
+    for i in range(0, len(items), size):
+        yield items[i : i + size]
+
+
+def best_thumbnail(thumbnails: dict | None) -> str | None:
+    if not thumbnails:
+        return None
+    for key in ("maxres", "standard", "high", "medium", "default"):
+        if key in thumbnails and thumbnails[key].get("url"):
+            return thumbnails[key]["url"]
+    return None
+
+
+class YouTubeClient:
+    def __init__(self, db, user: User):
+        self.db = db
+        self.user = user
+        self._http = httpx.Client(timeout=30.0)
+
+    def __enter__(self) -> "YouTubeClient":
+        return self
+
+    def __exit__(self, *exc) -> None:
+        self._http.close()
+
+    # --- auth ---
+    def _access_token(self) -> str:
+        tok = self.user.token
+        if tok is None:
+            raise YouTubeError("User has no stored OAuth token")
+        now = datetime.now(timezone.utc)
+        if tok.access_token and tok.expiry and tok.expiry > now + timedelta(seconds=60):
+            return tok.access_token
+        refresh = decrypt(tok.refresh_token_enc)
+        if not refresh:
+            raise YouTubeError("No refresh token; user must re-authenticate")
+        resp = self._http.post(
+            GOOGLE_TOKEN_URL,
+            data={
+                "client_id": settings.google_client_id,
+                "client_secret": settings.google_client_secret,
+                "refresh_token": refresh,
+                "grant_type": "refresh_token",
+            },
+        )
+        if resp.status_code != 200:
+            raise YouTubeError(f"Token refresh failed: {resp.status_code} {resp.text[:200]}")
+        data = resp.json()
+        tok.access_token = data["access_token"]
+        tok.expiry = now + timedelta(seconds=int(data.get("expires_in", 3600)))
+        self.db.add(tok)
+        self.db.commit()
+        return tok.access_token
+
+    # --- core request ---
+    def _get(self, path: str, params: dict, cost: int = 1, allow_key: bool = True) -> dict:
+        p = dict(params)
+        headers = {}
+        if allow_key and settings.youtube_api_key:
+            p["key"] = settings.youtube_api_key
+        else:
+            headers["Authorization"] = f"Bearer {self._access_token()}"
+        resp = self._http.get(f"{API_BASE}/{path}", params=p, headers=headers)
+        quota.record_usage(self.db, cost)
+        if resp.status_code != 200:
+            raise YouTubeError(f"GET {path} -> {resp.status_code}: {resp.text[:300]}")
+        return resp.json()
+
+    # --- endpoints ---
+    def iter_subscriptions(self) -> Iterator[dict]:
+        """Yield the authenticated user's subscriptions (requires OAuth)."""
+        page_token = None
+        while True:
+            params = {
+                "part": "snippet",
+                "mine": "true",
+                "maxResults": 50,
+                "order": "alphabetical",
+            }
+            if page_token:
+                params["pageToken"] = page_token
+            data = self._get("subscriptions", params, cost=1, allow_key=False)
+            for item in data.get("items", []):
+                snippet = item.get("snippet", {})
+                resource = snippet.get("resourceId", {})
+                yield {
+                    "channel_id": resource.get("channelId"),
+                    "title": snippet.get("title"),
+                    "description": snippet.get("description"),
+                    "thumbnail_url": best_thumbnail(snippet.get("thumbnails")),
+                    "yt_subscription_id": item.get("id"),
+                }
+            page_token = data.get("nextPageToken")
+            if not page_token:
+                break
+
+    def get_channels(self, channel_ids: list[str]) -> list[dict]:
+        items: list[dict] = []
+        for batch in _chunks(channel_ids, 50):
+            data = self._get(
+                "channels",
+                {
+                    "part": "snippet,contentDetails,statistics,topicDetails",
+                    "id": ",".join(batch),
+                    "maxResults": 50,
+                },
+            )
+            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