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/alembic/versions/0009_quota_events.py

@@ -0,0 +1,46 @@
+"""quota_events: per-spend audit log for per-user quota attribution
+
+Revision ID: 0009_quota_events
+Revises: 0008_invites
+Create Date: 2026-06-12
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+revision: str = "0009_quota_events"
+down_revision: Union[str, None] = "0008_invites"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "quota_events",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column(
+            "user_id",
+            sa.Integer(),
+            sa.ForeignKey("users.id", ondelete="SET NULL"),
+            nullable=True,
+        ),
+        sa.Column("action", sa.String(length=40), nullable=False),
+        sa.Column("units", sa.Integer(), nullable=False),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.func.now(),
+            nullable=False,
+        ),
+    )
+    op.create_index("ix_quota_events_user_id", "quota_events", ["user_id"])
+    op.create_index("ix_quota_events_action", "quota_events", ["action"])
+    op.create_index("ix_quota_events_created_at", "quota_events", ["created_at"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_quota_events_created_at", table_name="quota_events")
+    op.drop_index("ix_quota_events_action", table_name="quota_events")
+    op.drop_index("ix_quota_events_user_id", table_name="quota_events")
+    op.drop_table("quota_events")

backend/app/main.py

@@ -26,7 +26,7 @@ from starlette.middleware.sessions import SessionMiddleware
 
 from app import auth
 from app.config import settings
-from app.routes import admin, channels, feed, health, me, sync, tags
+from app.routes import admin, channels, feed, health, me, quota, sync, tags
 from app.scheduler import shutdown_scheduler, start_scheduler
 
 
@@ -67,6 +67,7 @@ app.include_router(feed.router)
 app.include_router(me.router)
 app.include_router(channels.router)
 app.include_router(admin.router)
+app.include_router(quota.router)
 
 # The built SPA (populated by the Docker frontend build stage).
 STATIC_DIR = Path(__file__).parent / "static_spa"

backend/app/models.py

@@ -265,6 +265,25 @@ class ApiQuotaUsage(Base):
     units_used: Mapped[int] = mapped_column(Integer, default=0, server_default="0")
 
 
+class QuotaEvent(Base):
+    """Per-spend audit log for YouTube API quota attribution. `user_id` is the end-user
+    whose action caused the spend; NULL means background/system work (scheduler backfill,
+    enrichment, resync). The aggregate budget still lives in ApiQuotaUsage; this is detail."""
+
+    __tablename__ = "quota_events"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    # SET NULL on user delete: keep the audit trail (it just becomes system-attributed).
+    user_id: Mapped[int | None] = mapped_column(
+        ForeignKey("users.id", ondelete="SET NULL"), index=True
+    )
+    action: Mapped[str] = mapped_column(String(40), index=True)
+    units: Mapped[int] = mapped_column(Integer)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), index=True
+    )
+
+
 class AppState(Base):
     """Single-row global app state (admin-controlled)."""
 

backend/app/quota.py

@@ -4,22 +4,52 @@ The whole app shares one daily budget (the API resets at midnight US Pacific tim
 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
+import contextlib
+import contextvars
+from datetime import datetime, timezone
 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
+from app.models import ApiQuotaUsage, QuotaEvent
 
 _PACIFIC = ZoneInfo("America/Los_Angeles")
 
+# Request/job-scoped attribution for quota spend: who triggered it and what kind of work.
+# Set at entry points (route handlers, scheduler jobs) via attribute(); read by
+# record_usage. Default = background/system, generic action.
+_actor_id: contextvars.ContextVar[int | None] = contextvars.ContextVar(
+    "quota_actor_id", default=None
+)
+_action: contextvars.ContextVar[str] = contextvars.ContextVar(
+    "quota_action", default="api"
+)
+
+
+@contextlib.contextmanager
+def attribute(actor_id: int | None, action: str):
+    """Attribute any quota spent in this block to `actor_id` (None = system) under `action`."""
+    t1 = _actor_id.set(actor_id)
+    t2 = _action.set(action)
+    try:
+        yield
+    finally:
+        _actor_id.reset(t1)
+        _action.reset(t2)
+
 
 def pacific_today():
     return datetime.now(_PACIFIC).date()
 
 
+def pacific_day_start_utc() -> datetime:
+    """UTC instant of the current Pacific day's midnight — aligns 'today' with the budget reset."""
+    start = datetime.now(_PACIFIC).replace(hour=0, minute=0, second=0, microsecond=0)
+    return start.astimezone(timezone.utc)
+
+
 def units_used_today(db: Session) -> int:
     row = db.get(ApiQuotaUsage, pacific_today())
     return row.units_used if row else 0
@@ -34,7 +64,7 @@ def can_spend(db: Session, units: int) -> bool:
 
 
 def record_usage(db: Session, units: int) -> None:
-    """Atomically add `units` to today's counter (upsert)."""
+    """Atomically add `units` to today's counter (upsert) and log an attribution event."""
     if units <= 0:
         return
     day = pacific_today()
@@ -47,4 +77,6 @@ def record_usage(db: Session, units: int) -> None:
         )
     )
     db.execute(stmt)
+    # Audit detail: who/what spent it (per-user attribution; NULL actor = system).
+    db.add(QuotaEvent(user_id=_actor_id.get(), action=_action.get(), units=units))
     db.commit()

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,

backend/app/scheduler.py

@@ -4,6 +4,7 @@ import logging
 
 from apscheduler.schedulers.background import BackgroundScheduler
 
+from app import quota
 from app.config import settings
 from app.db import SessionLocal
 from app.state import is_sync_paused
@@ -42,14 +43,21 @@ def _rss_job() -> None:
 
 
 def _enrich_job() -> None:
-    _job("enrich", lambda db: run_enrich(db))
+    def work(db):
+        with quota.attribute(None, "enrich"):
+            return run_enrich(db)
+
+    _job("enrich", work)
 
 
 def _backfill_job() -> None:
-    # Recent-first for not-yet-synced channels, then deep backfill for the rest.
+    # Recent-first for not-yet-synced channels, then deep backfill for the rest. All
+    # background spend is attributed to the system (no actor), split by action.
     def work(db):
-        recent = run_recent_backfill(db, max_channels=25)
-        deep = run_deep_backfill(db, max_channels=10)
+        with quota.attribute(None, "backfill_recent"):
+            recent = run_recent_backfill(db, max_channels=25)
+        with quota.attribute(None, "backfill_deep"):
+            deep = run_deep_backfill(db, max_channels=10)
         return {"recent": recent, "deep": deep}
 
     _job("backfill", work)
@@ -64,7 +72,11 @@ def _shorts_job() -> None:
 
 
 def _subscriptions_job() -> None:
-    _job("subscriptions", run_subscription_resync)
+    def work(db):
+        with quota.attribute(None, "subscription_resync"):
+            return run_subscription_resync(db)
+
+    _job("subscriptions", work)
 
 
 def start_scheduler() -> None: