feat(stats): per-user API quota attribution + admin usage page

Track who burned how much YouTube API quota. A QuotaEvent audit log (migration
0009) records every spend with the triggering user (NULL = background/system) and
an action label, set via a request/job-scoped contextvar (quota.attribute) so no
call signatures change. User-initiated work (sync subscriptions, unsubscribe,
opt-in recent backfill, manual enrich) attributes to the user; scheduler work to
System, split by action.

- backend: QuotaEvent model + migration 0009; quota.attribute() contextvar;
  record_usage logs events; entry points wrapped (routes/sync, routes/channels,
  scheduler); GET /api/quota/my-usage + GET /api/quota/admin
- frontend: admin-only Stats page (header nav, page=stats) with daily bars +
  per-user breakdown by action and range picker; 'Your API usage' in Settings ->
  Sync for every user

Verified: attribution + endpoints compute correctly; events are per-user vs System.

backend/app/routes/channels.py

@@ -5,6 +5,7 @@ from fastapi import APIRouter, Depends, HTTPException
 from sqlalchemy import and_, func, select
 from sqlalchemy.orm import Session
 
+from app import quota
 from app.auth import current_user, has_write_scope
 from app.db import get_db
 from app.models import Channel, ChannelTag, Subscription, Tag, User, Video
@@ -121,7 +122,8 @@ def update_channel(
     # fetched its recent uploads yet, do that one channel now (cheap) instead of waiting for
     # the scheduler. Deep paging still follows on the scheduler's next run (recent-then-deep).
     if deep_turned_on and channel is not None and channel.recent_synced_at is None:
-        run_recent_backfill(db, [channel], max_channels=1)
+        with quota.attribute(user.id, "backfill_recent"):
+            run_recent_backfill(db, [channel], max_channels=1)
 
     return {
         "id": channel_id,
@@ -152,7 +154,7 @@ def unsubscribe(
             detail="No YouTube subscription id on record — sync subscriptions first.",
         )
     try:
-        with YouTubeClient(db, user) as yt:
+        with quota.attribute(user.id, "unsubscribe"), YouTubeClient(db, user) as yt:
             yt.delete_subscription(sub.yt_subscription_id)
     except YouTubeError as exc:
         raise HTTPException(status_code=502, detail=f"YouTube unsubscribe failed: {exc}")

backend/app/routes/quota.py

@@ -0,0 +1,106 @@
+"""Quota usage attribution: a user's own consumption + an admin breakdown across users.
+
+The aggregate daily budget lives in ApiQuotaUsage; here we read the per-spend audit log
+(QuotaEvent) to attribute units to the user who triggered them (NULL actor = system).
+"""
+from datetime import datetime, timedelta, timezone
+
+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 QuotaEvent, User
+from app.routes.admin import admin_user
+
+router = APIRouter(prefix="/api/quota", tags=["quota"])
+
+
+def _sum(db: Session, *conds) -> int:
+    return int(
+        db.scalar(
+            select(func.coalesce(func.sum(QuotaEvent.units), 0)).where(*conds)
+        )
+        or 0
+    )
+
+
+@router.get("/my-usage")
+def my_usage(
+    user: User = Depends(current_user), db: Session = Depends(get_db)
+) -> dict:
+    now = datetime.now(timezone.utc)
+    mine = QuotaEvent.user_id == user.id
+    by_action = {
+        action: int(units)
+        for action, units in db.execute(
+            select(QuotaEvent.action, func.sum(QuotaEvent.units))
+            .where(mine, QuotaEvent.created_at >= now - timedelta(days=30))
+            .group_by(QuotaEvent.action)
+        ).all()
+    }
+    return {
+        "today": _sum(db, mine, QuotaEvent.created_at >= quota.pacific_day_start_utc()),
+        "last_7d": _sum(db, mine, QuotaEvent.created_at >= now - timedelta(days=7)),
+        "last_30d": _sum(db, mine, QuotaEvent.created_at >= now - timedelta(days=30)),
+        "all_time": _sum(db, mine),
+        "by_action": by_action,
+    }
+
+
+@router.get("/admin")
+def admin_usage(
+    days: int = 30,
+    _: User = Depends(admin_user),
+    db: Session = Depends(get_db),
+) -> dict:
+    days = max(1, min(days, 365))
+    since = datetime.now(timezone.utc) - timedelta(days=days)
+
+    totals = db.execute(
+        select(QuotaEvent.user_id, func.sum(QuotaEvent.units))
+        .where(QuotaEvent.created_at >= since)
+        .group_by(QuotaEvent.user_id)
+    ).all()
+
+    per_action: dict[int | None, dict[str, int]] = {}
+    for uid, action, units in db.execute(
+        select(QuotaEvent.user_id, QuotaEvent.action, func.sum(QuotaEvent.units))
+        .where(QuotaEvent.created_at >= since)
+        .group_by(QuotaEvent.user_id, QuotaEvent.action)
+    ).all():
+        per_action.setdefault(uid, {})[action] = int(units)
+
+    uids = [uid for uid, _ in totals if uid is not None]
+    emails = (
+        {u.id: u.email for u in db.execute(select(User).where(User.id.in_(uids))).scalars()}
+        if uids
+        else {}
+    )
+
+    rows = [
+        {
+            "user_id": uid,
+            "email": emails.get(uid, "(deleted user)") if uid is not None else "System",
+            "total": int(total),
+            "by_action": per_action.get(uid, {}),
+        }
+        for uid, total in totals
+    ]
+    rows.sort(key=lambda r: r["total"], reverse=True)
+
+    # Instance-wide daily series in Pacific days (matches the budget reset).
+    day_expr = func.date(func.timezone("America/Los_Angeles", QuotaEvent.created_at))
+    daily = [
+        {"day": str(day), "total": int(units)}
+        for day, units in db.execute(
+            select(day_expr, func.sum(QuotaEvent.units))
+            .where(QuotaEvent.created_at >= since)
+            .group_by(day_expr)
+            .order_by(day_expr)
+        ).all()
+    ]
+
+    return {"range_days": days, "rows": rows, "daily": daily}

backend/app/routes/sync.py

@@ -34,7 +34,8 @@ 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)
+    with quota.attribute(user.id, "sync_subscriptions"):
+        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
@@ -55,7 +56,8 @@ def sync_backfill(
     max_channels: int = 25,
 ) -> dict:
     before = quota.units_used_today(db)
-    result = run_recent_backfill(db, _user_channels(db, user), max_channels=max_channels)
+    with quota.attribute(user.id, "backfill_recent"):
+        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
 
@@ -65,7 +67,8 @@ def sync_enrich(
     user: User = Depends(current_user), db: Session = Depends(get_db)
 ) -> dict:
     before = quota.units_used_today(db)
-    enriched = run_enrich(db)
+    with quota.attribute(user.id, "enrich"):
+        enriched = run_enrich(db)
     return {
         "enriched": enriched,
         "quota_used_estimate": quota.units_used_today(db) - before,