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/alembic/versions/0007_deep_requested.py

@@ -0,0 +1,37 @@
+"""per-user opt-in to full-history (deep) backfill
+
+Revision ID: 0007_deep_requested
+Revises: 0006_app_state
+Create Date: 2026-06-11
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+revision: str = "0007_deep_requested"
+down_revision: Union[str, None] = "0006_app_state"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "subscriptions",
+        sa.Column(
+            "deep_requested",
+            sa.Boolean(),
+            nullable=False,
+            server_default="false",
+        ),
+    )
+    # Preserve today's behaviour: admins already had every channel deep-backfilling
+    # globally, so opt their existing subscriptions in. Other users start recent-only.
+    op.execute(
+        "UPDATE subscriptions SET deep_requested = true "
+        "WHERE user_id IN (SELECT id FROM users WHERE role = 'admin')"
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("subscriptions", "deep_requested")

backend/app/models.py

@@ -112,6 +112,12 @@ class Subscription(Base):
     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")
+    # Per-user opt-in to full-history (deep) backfill of this channel. Default off so a
+    # new user's unique channels don't trigger a big shared-quota burst; the deep
+    # scheduler only picks up channels at least one user has requested.
+    deep_requested: 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()

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)

backend/app/sync/runner.py

@@ -13,7 +13,7 @@ from app import quota
 from app.config import settings
 
 log = logging.getLogger("subfeed.sync")
-from app.models import Channel, OAuthToken, User
+from app.models import Channel, OAuthToken, Subscription, User
 from app.sync.subscriptions import import_subscriptions
 from app.sync.videos import (
     backfill_channel_deep,
@@ -125,11 +125,22 @@ def run_deep_backfill(db: Session, max_channels: int = 10, max_pages: int = 5) -
     user = get_service_user(db)
     if user is None:
         return {"channels_processed": 0, "videos_added": 0, "reason": "no service user"}
+    # Demand-driven: only deep-backfill channels at least one user has opted into
+    # (Subscription.deep_requested). Recent backfill stays unconditional (cheap).
+    deep_requested = (
+        select(Subscription.id)
+        .where(
+            Subscription.channel_id == Channel.id,
+            Subscription.deep_requested.is_(True),
+        )
+        .exists()
+    )
     channels = (
         db.execute(
             select(Channel).where(
                 Channel.backfill_done.is_(False),
                 Channel.recent_synced_at.is_not(None),
+                deep_requested,
             )
         )
         .scalars()
@@ -148,3 +159,26 @@ def run_deep_backfill(db: Session, max_channels: int = 10, max_pages: int = 5) -
                 db.rollback()
                 log.exception("Deep backfill failed for channel %s", channel.id)
     return {"channels_processed": processed, "videos_added": videos_added}
+
+
+def estimate_deep_backfill(db: Session, channels: list[Channel]) -> dict:
+    """Rough ETA to finish full-history backfill of `channels`. Quota-bound: total
+    estimated API units / the daily budget left for backfill. Deliberately approximate —
+    a channel's cost ~= playlistItems pages (1 unit / 50 videos) + enrichment (likewise)."""
+    pending = [c for c in channels if not c.backfill_done]
+
+    def units(c: Channel) -> int:
+        videos = c.video_count or 0
+        pages = max(1, (videos + 49) // 50)
+        return 2 * pages
+
+    total_units = sum(units(c) for c in pending)
+    daily_budget = max(1, settings.quota_daily_budget - settings.backfill_quota_reserve)
+    eta_seconds = int(total_units / daily_budget * 86_400) if pending else 0
+    return {
+        "channels_pending": len(pending),
+        "videos_pending_est": sum(c.video_count or 0 for c in pending),
+        "units_est": total_units,
+        "daily_budget": daily_budget,
+        "eta_seconds": eta_seconds,
+    }