feat(m5d): demand-driven deep backfill + per-user ETA

Per-user opt-in to full-history (deep) backfill so a new user's unique
channels no longer trigger a big shared-quota burst.

- migration 0007: Subscription.deep_requested (default false); seed admins'
  existing subscriptions to preserve today's global-backfill behaviour
- run_deep_backfill is now demand-driven: only channels at least one user has
  requested are deep-backfilled; recent backfill stays unconditional (cheap)
- estimate_deep_backfill ETA helper (quota-bound) surfaced in /api/sync/my-status
- POST /api/sync/deep-all to opt all my channels in; PATCH channels accepts
  deep_requested
- UI: per-channel Full history toggle, Backfill everything action, deep
  progress + ETA in Channels header and Settings - Sync

backend/app/routes/channels.py

@@ -54,6 +54,7 @@ def list_channels(
             "stored_videos": stored.get(ch.id, 0),
             "priority": sub.priority,
             "hidden": sub.hidden,
+            "deep_requested": sub.deep_requested,
             "tag_ids": tags_by_channel.get(ch.id, []),
             "details_synced": ch.details_synced_at is not None,
             "recent_synced": ch.recent_synced_at is not None,
@@ -86,8 +87,18 @@ def update_channel(
         sub.priority = int(payload["priority"])
     if "hidden" in payload:
         sub.hidden = bool(payload["hidden"])
+    if "deep_requested" in payload:
+        # Opt this channel into (or out of) full-history backfill. The deep scheduler
+        # picks the flag up on its next run; turning it off won't undo already-fetched
+        # videos, it just stops further deep paging if nobody else still wants it.
+        sub.deep_requested = bool(payload["deep_requested"])
     db.commit()
-    return {"id": channel_id, "priority": sub.priority, "hidden": sub.hidden}
+    return {
+        "id": channel_id,
+        "priority": sub.priority,
+        "hidden": sub.hidden,
+        "deep_requested": sub.deep_requested,
+    }
 
 
 @router.post("/{channel_id}/tags")

backend/app/routes/sync.py

@@ -1,12 +1,17 @@
 from fastapi import APIRouter, Depends, HTTPException
-from sqlalchemy import and_, case, func, select
+from sqlalchemy import and_, case, func, select, update
 from sqlalchemy.orm import Session
 
 from app import quota, state
 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.runner import (
+    estimate_deep_backfill,
+    run_enrich,
+    run_recent_backfill,
+    run_rss_poll,
+)
 from app.sync.subscriptions import import_subscriptions
 
 router = APIRouter(prefix="/api/sync", tags=["sync"])
@@ -113,6 +118,21 @@ def my_status(
     ).one()
     total = total or 0
     deep_done = int(deep_done or 0)
+
+    # Per-user full-history (deep) backfill: channels this user has opted in, and an ETA
+    # for the still-pending ones (quota-bound estimate).
+    deep_requested_channels = (
+        db.execute(
+            select(Channel).join(
+                Subscription,
+                and_(sub_join, Subscription.deep_requested.is_(True)),
+            )
+        )
+        .scalars()
+        .all()
+    )
+    deep_eta = estimate_deep_backfill(db, deep_requested_channels)
+
     my_videos = db.scalar(
         select(func.count(Video.id)).join(
             Subscription,
@@ -130,6 +150,9 @@ def my_status(
         "channels_deep_done": deep_done,
         "channels_recent_pending": total - recent_synced,
         "channels_deep_pending": total - deep_done,
+        "channels_deep_requested": len(deep_requested_channels),
+        "deep_pending_count": deep_eta["channels_pending"],
+        "deep_eta_seconds": deep_eta["eta_seconds"],
         "my_videos": my_videos or 0,
         "quota_used_today": quota.units_used_today(db),
         "quota_remaining_today": quota.remaining_today(db),
@@ -137,6 +160,22 @@ def my_status(
     }
 
 
+@router.post("/deep-all")
+def deep_all(
+    user: User = Depends(current_user),
+    db: Session = Depends(get_db),
+    on: bool = True,
+) -> dict:
+    """Opt every one of the user's channels into (or out of) full-history backfill."""
+    result = db.execute(
+        update(Subscription)
+        .where(Subscription.user_id == user.id)
+        .values(deep_requested=on)
+    )
+    db.commit()
+    return {"updated": result.rowcount, "deep_requested": on}
+
+
 @router.post("/pause")
 def pause_sync(
     user: User = Depends(current_user), db: Session = Depends(get_db)