Merge improvement/backend-consolidation

Maintenance epic phase 1: backend duplication cleanup + one correctness fix.
- fix: maintenance no longer scans the removed status=="saved" (under-counted impact)
- DRY: shared feed-row projection, playlist mirror/position/add-item helpers,
  token hashing, email validation, admin_user dep + count_admins, quota.measured,
  is_messageable_user, Timestamp/UpdatedAt model mixins.
Verified: compile + full app import + zero alembic schema drift + localdev smoke
(demo login, feed, playlist CRUD, sync status all green).

backend/app/auth.py

@@ -1,6 +1,4 @@
-import hashlib
 import logging
-import re
 import secrets
 from datetime import datetime, timedelta, timezone
 
@@ -8,7 +6,7 @@ import httpx
 from authlib.integrations.starlette_client import OAuth, OAuthError
 from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request
 from fastapi.responses import JSONResponse, RedirectResponse
-from sqlalchemy import delete, select
+from sqlalchemy import delete, func, select
 from sqlalchemy.orm import Session
 
 from app import email as email_mod
@@ -17,7 +15,8 @@ from app.config import settings
 from app.db import get_db
 from app.models import AuthToken, DemoWhitelist, Invite, OAuthToken, User
 from app.ratelimit import RateLimiter
-from app.security import decrypt, encrypt, hash_password, verify_password
+from app.security import decrypt, encrypt, hash_password, hash_token, verify_password
+from app.utils import valid_email
 
 # Email+password auth tuning.
 PASSWORD_MIN_LEN = 10
@@ -43,8 +42,6 @@ def _notify_suspended(background: BackgroundTasks, email: str) -> None:
     if _suspend_email_limiter.allow(email):
         background.add_task(email_mod.send_account_suspended, email, operator_contact())
 
-_EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
-
 # The single shared demo account's stable identity. The row is created lazily on first
 # demo login (see get_or_create_demo_user); these are just its sentinel keys.
 DEMO_GOOGLE_SUB = "__demo__"
@@ -184,7 +181,7 @@ def upsert_pending_invite(db: Session, email: str) -> Invite | None:
     """Idempotently record an access request. Returns the Invite if a *new* pending row
     was created (so the caller can notify admins), else None for already-known emails."""
     email = (email or "").lower()
-    if not _EMAIL_RE.match(email):
+    if not valid_email(email):
         return None
     inv = db.execute(select(Invite).where(Invite.email == email)).scalar_one_or_none()
     if inv is not None:
@@ -443,7 +440,7 @@ async def request_access(
     """Public: ask for access. Idempotent — repeat requests for the same email don't
     duplicate or re-spam admins (upsert returns None for an already-pending row)."""
     email = (payload.get("email") or "").strip().lower()
-    if not _EMAIL_RE.match(email):
+    if not valid_email(email):
         raise HTTPException(status_code=400, detail="Enter a valid email address.")
     if is_allowed(db, email):
         return {"status": "approved"}  # already allowed — just sign in
@@ -465,7 +462,7 @@ def demo_login(payload: dict, request: Request, db: Session = Depends(get_db)) -
     if not _demo_limiter.allow(_client_ip(request)):
         return {"authenticated": False}
     email = (payload.get("email") or "").strip().lower()
-    if _EMAIL_RE.match(email) and is_demo_allowed(db, email):
+    if valid_email(email) and is_demo_allowed(db, email):
         demo = get_or_create_demo_user(db)
         request.session["user_id"] = demo.id
         log.info("Demo login (key=%s, demo_id=%s)", email, demo.id)
@@ -487,10 +484,6 @@ def _app_base() -> str:
     return settings.app_base
 
 
-def _hash_token(raw: str) -> str:
-    return hashlib.sha256(raw.encode()).hexdigest()
-
-
 def _issue_token(db: Session, user: User, kind: str, ttl: timedelta) -> str:
     """Create a single-use token of `kind`; only its hash is stored. Returns the raw token
     (goes only into the emailed link)."""
@@ -499,7 +492,7 @@ def _issue_token(db: Session, user: User, kind: str, ttl: timedelta) -> str:
         AuthToken(
             user_id=user.id,
             kind=kind,
-            token_hash=_hash_token(raw),
+            token_hash=hash_token(raw),
             expires_at=datetime.now(timezone.utc) + ttl,
         )
     )
@@ -513,7 +506,7 @@ def _consume_token(db: Session, raw: str | None, kind: str) -> User | None:
         return None
     row = db.execute(
         select(AuthToken).where(
-            AuthToken.token_hash == _hash_token(raw), AuthToken.kind == kind
+            AuthToken.token_hash == hash_token(raw), AuthToken.kind == kind
         )
     ).scalar_one_or_none()
     if row is None or row.used_at is not None or row.expires_at < datetime.now(timezone.utc):
@@ -537,7 +530,7 @@ def register(
     email = (payload.get("email") or "").strip().lower()
     password = payload.get("password") or ""
     # Input validation is safe to reveal (independent of whether the email exists).
-    if not _EMAIL_RE.match(email):
+    if not valid_email(email):
         raise HTTPException(status_code=400, detail="Enter a valid email address.")
     if len(password) < PASSWORD_MIN_LEN:
         raise HTTPException(
@@ -633,7 +626,7 @@ def password_reset_request(
     if not _reset_limiter.allow(_client_ip(request)):
         return {"status": "ok"}
     email = (payload.get("email") or "").strip().lower()
-    if _EMAIL_RE.match(email):
+    if valid_email(email):
         user = db.execute(select(User).where(User.email == email)).scalar_one_or_none()
         if user is not None and user.password_hash and not user.is_demo:
             raw = _issue_token(db, user, "reset", RESET_TTL)
@@ -694,6 +687,23 @@ def require_human(user: User = Depends(current_user)) -> User:
     return user
 
 
+def admin_user(user: User = Depends(current_user)) -> User:
+    """Dependency: require the signed-in user to be an admin. Single source of truth for the
+    admin gate — every admin route/action depends on this rather than re-checking the role."""
+    if user.role != "admin":
+        raise HTTPException(status_code=403, detail="Admin only")
+    return user
+
+
+def count_admins(db: Session, *, active_only: bool = False) -> int:
+    """How many admins exist (optionally only un-suspended ones). Used by the 'can't remove /
+    suspend / delete the last admin' guards so they don't each re-spell the count query."""
+    q = select(func.count()).select_from(User).where(User.role == "admin")
+    if active_only:
+        q = q.where(User.is_suspended.is_(False))
+    return db.scalar(q) or 0
+
+
 @router.get("/link")
 async def link_google(
     request: Request,

backend/app/models.py

@@ -19,7 +19,24 @@ from sqlalchemy.orm import Mapped, mapped_column, relationship
 from app.db import Base
 
 
-class User(Base):
+class TimestampMixin:
+    """Adds a server-defaulted `created_at`. Models that index created_at or order it
+    differently (QuotaEvent, Notification, Message) keep their own column on purpose."""
+
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now()
+    )
+
+
+class UpdatedAtMixin:
+    """Adds an `updated_at` that server-defaults on insert and bumps on every UPDATE."""
+
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), onupdate=func.now()
+    )
+
+
+class User(Base, TimestampMixin):
     __tablename__ = "users"
 
     id: Mapped[int] = mapped_column(primary_key=True)
@@ -57,9 +74,6 @@ class User(Base):
     )
     # Free-form UI preferences (theme, color scheme, font scale, default filters…).
     preferences: Mapped[dict | None] = mapped_column(JSON)
-    created_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), server_default=func.now()
-    )
 
     token: Mapped["OAuthToken | None"] = relationship(
         back_populates="user", uselist=False, cascade="all, delete-orphan"
@@ -69,7 +83,7 @@ class User(Base):
     )
 
 
-class OAuthToken(Base):
+class OAuthToken(Base, UpdatedAtMixin):
     __tablename__ = "oauth_tokens"
 
     id: Mapped[int] = mapped_column(primary_key=True)
@@ -81,14 +95,11 @@ class OAuthToken(Base):
     access_token: Mapped[str | None] = mapped_column(String)
     expiry: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
     scopes: Mapped[str | None] = mapped_column(String)
-    updated_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), server_default=func.now(), onupdate=func.now()
-    )
 
     user: Mapped["User"] = relationship(back_populates="token")
 
 
-class AuthToken(Base):
+class AuthToken(Base, TimestampMixin):
     """Single-use, expiring token for email verification or password reset. Only the SHA-256
     HASH of the token is stored, so a DB leak can't be used to verify/reset; the raw token
     lives only in the emailed link. Consumed (used_at set) on first valid use."""
@@ -103,9 +114,6 @@ class AuthToken(Base):
     token_hash: Mapped[str] = mapped_column(String(64), unique=True, index=True)
     expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True))
     used_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
-    created_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), server_default=func.now()
-    )
 
 
 class Invite(Base):
@@ -128,7 +136,7 @@ class Invite(Base):
     decided_by: Mapped[str | None] = mapped_column(String(320))  # admin email
 
 
-class DemoWhitelist(Base):
+class DemoWhitelist(Base, TimestampMixin):
     """Emails that may enter the shared demo account from the login page (no Google OAuth).
     A hidden, admin-curated list of "keys to the same door" — every entry logs into the one
     shared demo user. Distinct from Invite (which gates real Google sign-in)."""
@@ -139,12 +147,9 @@ class DemoWhitelist(Base):
     email: Mapped[str] = mapped_column(String(320), unique=True, index=True)
     note: Mapped[str | None] = mapped_column(Text)
     added_by: Mapped[str | None] = mapped_column(String(320))  # admin email
-    created_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), server_default=func.now()
-    )
 
 
-class Channel(Base):
+class Channel(Base, TimestampMixin):
     """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."""
 
@@ -170,15 +175,12 @@ class Channel(Base):
     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):
+class Subscription(Base, TimestampMixin):
     """Per-user link to a channel, with the user's own priority/visibility overrides."""
 
     __tablename__ = "subscriptions"
@@ -202,15 +204,12 @@ class Subscription(Base):
         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):
+class Video(Base, TimestampMixin):
     """A YouTube video. Shared across users; per-user state lives elsewhere."""
 
     __tablename__ = "videos"
@@ -260,14 +259,11 @@ class Video(Base):
         DateTime(timezone=True), index=True
     )
     unavailable_reason: Mapped[str | None] = mapped_column(String(24))
-    created_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), server_default=func.now()
-    )
 
     channel: Mapped["Channel"] = relationship(back_populates="videos")
 
 
-class Tag(Base):
+class Tag(Base, TimestampMixin):
     """A label. System tags (user_id NULL) are computed automatically and shared by all
     users; user tags (user_id set) are personal. category groups them in the UI."""
 
@@ -283,9 +279,6 @@ class Tag(Base):
     category: Mapped[str] = mapped_column(
         String(16), default="other", server_default="other"
     )
-    created_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), server_default=func.now()
-    )
 
 
 class ChannelTag(Base):
@@ -312,7 +305,7 @@ class ChannelTag(Base):
     confidence: Mapped[float | None] = mapped_column(Float)
 
 
-class VideoState(Base):
+class VideoState(Base, UpdatedAtMixin):
     """Per-user watch state for a video. Rows exist only for non-default states; the
     absence of a row means 'new/unseen'."""
 
@@ -326,7 +319,7 @@ class VideoState(Base):
     video_id: Mapped[str] = mapped_column(
         ForeignKey("videos.id", ondelete="CASCADE"), index=True
     )
-    # new | watched | saved | hidden
+    # new | watched | hidden  ("saved" is now Watch-later playlist membership, not a status)
     status: Mapped[str] = mapped_column(String(12), default="new", server_default="new")
     watched_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
     # Resume position (seconds into the video) for the in-app player. Orthogonal to
@@ -337,9 +330,6 @@ class VideoState(Base):
         Integer, default=0, server_default="0", nullable=False
     )
     progress_updated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
-    updated_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), server_default=func.now(), onupdate=func.now()
-    )
 
 
 class ApiQuotaUsage(Base):
@@ -422,7 +412,7 @@ class SystemConfig(Base):
     )
 
 
-class Playlist(Base):
+class Playlist(Base, TimestampMixin, UpdatedAtMixin):
     """A per-user named, ordered collection of videos from the shared catalog.
 
     `kind`: 'user' (normal) or 'watch_later' (built-in, undeletable, one per user).
@@ -450,12 +440,6 @@ class Playlist(Base):
     # baseline `dirty` is derived against. NULL = no baseline yet (treated as dirty).
     synced_fingerprint: Mapped[str | None] = mapped_column(String)
     position: Mapped[int] = mapped_column(Integer, default=0, server_default="0")
-    created_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), server_default=func.now()
-    )
-    updated_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), server_default=func.now(), onupdate=func.now()
-    )
 
     items: Mapped[list["PlaylistItem"]] = relationship(
         back_populates="playlist",
@@ -560,7 +544,7 @@ class Message(Base):
     )
 
 
-class MessageKey(Base):
+class MessageKey(Base, TimestampMixin):
     """A user's end-to-end-encryption key material for direct messaging (phase 2).
 
     One row per user. `public_key` (base64 SPKI, ECDH P-256) is shared with others so they can
@@ -581,6 +565,3 @@ class MessageKey(Base):
     salt: Mapped[str] = mapped_column(String(64))
     wrap_iv: Mapped[str] = mapped_column(String(32))
     key_check: Mapped[str] = mapped_column(Text)
-    created_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), server_default=func.now()
-    )

backend/app/quota.py

@@ -66,6 +66,26 @@ def attribute(actor_id: int | None, action: str):
         _action.reset(t2)
 
 
+class Measure:
+    """Result handle from `measured()` — read .used / .remaining after the block exits."""
+
+    used = 0
+    remaining = 0
+
+
+@contextlib.contextmanager
+def measured(db: Session, actor_id: int | None, action: str):
+    """Like `attribute()`, but also reports how much quota the block spent. Yields a `Measure`
+    whose `.used` (units charged inside the block) and `.remaining` (today's remaining after)
+    are populated on exit — so manual sync routes don't each re-spell the before/after diff."""
+    m = Measure()
+    before = units_used_today(db)
+    with attribute(actor_id, action):
+        yield m
+    m.used = units_used_today(db) - before
+    m.remaining = remaining_today(db)
+
+
 def pacific_today():
     return datetime.now(_PACIFIC).date()
 

backend/app/routes/admin.py

@@ -1,6 +1,5 @@
 """Admin-only onboarding: review access requests (Invites), approve/deny, manual add.
 Also the demo-account admin surface: the demo email whitelist + a manual state reset."""
-import re
 from datetime import datetime, timezone
 
 from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException
@@ -8,7 +7,13 @@ from sqlalchemy import delete, func, select
 from sqlalchemy.orm import Session
 
 from app import email as email_mod
-from app.auth import current_user, get_or_create_demo_user, operator_contact, purge_user
+from app.auth import (
+    admin_user,
+    count_admins,
+    get_or_create_demo_user,
+    operator_contact,
+    purge_user,
+)
 from app.db import get_db
 from app.models import (
     DemoWhitelist,
@@ -20,17 +25,10 @@ from app.models import (
     Video,
     VideoState,
 )
+from app.utils import valid_email
 
 router = APIRouter(prefix="/api/admin", tags=["admin"])
 
-_EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
-
-
-def admin_user(user: User = Depends(current_user)) -> User:
-    if user.role != "admin":
-        raise HTTPException(status_code=403, detail="Admin only")
-    return user
-
 
 def _serialize(inv: Invite) -> dict:
     return {
@@ -110,7 +108,7 @@ def add_invite(
 ) -> dict:
     """Manually whitelist an email (approved straight away). Convenience for the admin."""
     email = (payload.get("email") or "").strip().lower()
-    if "@" not in email:
+    if not valid_email(email):
         raise HTTPException(status_code=400, detail="Enter a valid email address.")
     inv = db.execute(select(Invite).where(Invite.email == email)).scalar_one_or_none()
     if inv is None:
@@ -168,10 +166,8 @@ def set_user_role(
         raise HTTPException(status_code=400, detail="The demo account's role can't be changed.")
     if target.id == admin.id:
         raise HTTPException(status_code=400, detail="You can't change your own role.")
-    if target.role == "admin" and role == "user":
+    if target.role == "admin" and role == "user" and count_admins(db) <= 1:
-        admin_count = db.scalar(select(func.count()).select_from(User).where(User.role == "admin"))
+        raise HTTPException(status_code=400, detail="Can't remove the last admin.")
-        if (admin_count or 0) <= 1:
-            raise HTTPException(status_code=400, detail="Can't remove the last admin.")
     changed = target.role != role
     target.role = role
     db.commit()
@@ -201,14 +197,13 @@ def set_user_suspended(
         raise HTTPException(status_code=400, detail="The demo account can't be suspended.")
     if target.id == admin.id:
         raise HTTPException(status_code=400, detail="You can't suspend your own account.")
-    if suspended and target.role == "admin" and not target.is_suspended:
+    if (
-        active_admins = db.scalar(
+        suspended
-            select(func.count())
+        and target.role == "admin"
-            .select_from(User)
+        and not target.is_suspended
-            .where(User.role == "admin", User.is_suspended.is_(False))
+        and count_admins(db, active_only=True) <= 1
-        )
+    ):
-        if (active_admins or 0) <= 1:
+        raise HTTPException(status_code=400, detail="Can't suspend the last active admin.")
-            raise HTTPException(status_code=400, detail="Can't suspend the last active admin.")
     was_suspended = target.is_suspended
     target.is_suspended = suspended
     db.commit()
@@ -238,12 +233,8 @@ def admin_delete_user(
         raise HTTPException(
             status_code=400, detail="Delete your own account from Settings → Account."
         )
-    if target.role == "admin":
+    if target.role == "admin" and count_admins(db) <= 1:
-        admin_count = db.scalar(
+        raise HTTPException(status_code=400, detail="Can't delete the last admin.")
-            select(func.count()).select_from(User).where(User.role == "admin")
-        )
-        if (admin_count or 0) <= 1:
-            raise HTTPException(status_code=400, detail="Can't delete the last admin.")
     purge_user(db, target, background)
     return {"deleted": user_id}
 
@@ -281,7 +272,7 @@ def add_demo_whitelist(
     """Add an email that may enter the shared demo account from the login page (no Google
     sign-in). Idempotent — re-adding an existing email just returns it."""
     email = (payload.get("email") or "").strip().lower()
-    if not _EMAIL_RE.match(email):
+    if not valid_email(email):
         raise HTTPException(status_code=400, detail="Enter a valid email address.")
     row = db.execute(
         select(DemoWhitelist).where(DemoWhitelist.email == email)

backend/app/routes/feed.py

@@ -62,6 +62,29 @@ def _saved_expr(user: User):
     )
 
 
+def feed_columns(user: User, state) -> list:
+    """The shared feed-row projection (joined Channel fields + per-user watch state +
+    `saved`), consumed by both the feed query and the playlist-detail query so a new column
+    is added in exactly one place. `state` is the caller's aliased VideoState (outer-joined)."""
+    return [
+        Video.id,
+        Video.title,
+        Video.channel_id,
+        Channel.title.label("channel_title"),
+        Channel.thumbnail_url.label("channel_thumbnail"),
+        Channel.handle.label("channel_handle"),
+        Video.published_at,
+        Video.thumbnail_url,
+        Video.duration_seconds,
+        Video.view_count,
+        Video.is_short,
+        Video.live_status,
+        func.coalesce(state.status, "new").label("status"),
+        func.coalesce(state.position_seconds, 0).label("position_seconds"),
+        _saved_expr(user),
+    ]
+
+
 def _serialize(row) -> dict:
     return {
         "id": row.id,
@@ -115,23 +138,9 @@ def _filtered_query(
     status_expr = func.coalesce(state.status, "new")
     position_expr = func.coalesce(state.position_seconds, 0)
 
-    query = select(
+    query = select(*feed_columns(user, state)).join(
-        Video.id,
+        Channel, Channel.id == Video.channel_id
-        Video.title,
+    )
-        Video.channel_id,
-        Channel.title.label("channel_title"),
-        Channel.thumbnail_url.label("channel_thumbnail"),
-        Channel.handle.label("channel_handle"),
-        Video.published_at,
-        Video.thumbnail_url,
-        Video.duration_seconds,
-        Video.view_count,
-        Video.is_short,
-        Video.live_status,
-        status_expr.label("status"),
-        position_expr.label("position_seconds"),
-        _saved_expr(user),
-    ).join(Channel, Channel.id == Video.channel_id)
 
     if scope == "all":
         # Whole shared catalog; subscription is optional (only for priority sort).

backend/app/routes/me.py

@@ -3,6 +3,7 @@ from sqlalchemy import func, select
 from sqlalchemy.orm import Session
 
 from app.auth import (
+    count_admins,
     current_user,
     google_enabled,
     has_read_scope,
@@ -111,13 +112,11 @@ def delete_my_account(
     (that would lock everyone out of the admin surfaces)."""
     if user.is_demo:
         raise HTTPException(status_code=403, detail="The shared demo account can't be deleted.")
-    if user.role == "admin":
+    if user.role == "admin" and count_admins(db) <= 1:
-        admins = db.scalar(select(func.count()).select_from(User).where(User.role == "admin")) or 0
+        raise HTTPException(
-        if admins <= 1:
+            status_code=400,
-            raise HTTPException(
+            detail="You're the only admin — promote another admin before deleting your account.",
-                status_code=400,
+        )
-                detail="You're the only admin — promote another admin before deleting your account.",
-            )
     user_id = user.id
     # Full erasure (cascades) + access-request cleanup + Google-grant revocation; shared with the
     # admin delete path. Capture the id first — `user` is detached once purge_user deletes the row.

backend/app/routes/messages.py

@@ -103,6 +103,7 @@ def _serialize_msg(m: Message) -> dict:
 
 
 def _messageable() -> list:
+    """SQL clauses for "a real, active, non-suspended human" — for directory/recipient queries."""
     return [
         User.is_demo.is_(False),
         User.is_active.is_(True),
@@ -110,6 +111,11 @@ def _messageable() -> list:
     ]
 
 
+def is_messageable_user(u: User | None) -> bool:
+    """Python mirror of `_messageable()` for an already-loaded row (WS auth, send recipient)."""
+    return bool(u and not u.is_demo and u.is_active and not u.is_suspended)
+
+
 def ensure_welcome(db: Session, user: User) -> None:
     """Create the one-time system welcome message for a (human) user the first time it's needed.
     Idempotent: guarded on whether any system message already exists for them."""
@@ -315,7 +321,7 @@ def send_message(
     if payload.recipient_id == user.id:
         raise HTTPException(status_code=400, detail="You can't message yourself.")
     recipient = db.get(User, payload.recipient_id)
-    if recipient is None or recipient.is_demo or not recipient.is_active or recipient.is_suspended:
+    if not is_messageable_user(recipient):
         raise HTTPException(status_code=404, detail="Recipient not available.")
     if db.get(MessageKey, recipient.id) is None:
         raise HTTPException(status_code=404, detail="That member hasn't set up messaging yet.")
@@ -363,8 +369,7 @@ async def messages_ws(ws: WebSocket) -> None:
         return
     db = SessionLocal()
     try:
-        u = db.get(User, uid)
+        ok = is_messageable_user(db.get(User, uid))
-        ok = bool(u and not u.is_demo and not u.is_suspended and u.is_active)
     finally:
         db.close()
     if not ok:

backend/app/routes/playlists.py

@@ -9,7 +9,7 @@ from app import quota
 from app.auth import current_user, has_read_scope, has_write_scope
 from app.db import get_db
 from app.models import Channel, Playlist, PlaylistItem, User, Video, VideoState
-from app.routes.feed import _saved_expr, _serialize
+from app.routes.feed import _serialize, feed_columns
 from app.sync.playlists import (
     plan_push,
     push_playlist,
@@ -29,6 +29,51 @@ def _own_playlist(db: Session, user: User, playlist_id: int) -> Playlist:
     return pl
 
 
+def _next_playlist_position(db: Session, user_id: int) -> int:
+    """Append position for a new playlist in this user's rail."""
+    return (
+        db.scalar(
+            select(func.coalesce(func.max(Playlist.position), -1)).where(
+                Playlist.user_id == user_id
+            )
+        )
+        + 1
+    )
+
+
+def _next_item_position(db: Session, playlist_id: int) -> int:
+    """Append position for a new item at the end of a playlist."""
+    return (
+        db.scalar(
+            select(func.coalesce(func.max(PlaylistItem.position), -1)).where(
+                PlaylistItem.playlist_id == playlist_id
+            )
+        )
+        + 1
+    )
+
+
+def _add_item(db: Session, pl: Playlist, video_id: str, *, mark_dirty: bool) -> bool:
+    """Append a video to a playlist if not already present (idempotent). Returns whether it
+    was added. `mark_dirty` recomputes the YouTube-sync dirty flag (skip for Watch later)."""
+    exists = db.scalar(
+        select(PlaylistItem).where(
+            PlaylistItem.playlist_id == pl.id, PlaylistItem.video_id == video_id
+        )
+    )
+    if exists is not None:
+        return False
+    db.add(
+        PlaylistItem(
+            playlist_id=pl.id, video_id=video_id, position=_next_item_position(db, pl.id)
+        )
+    )
+    if mark_dirty:
+        recompute_dirty(db, pl)
+    db.commit()
+    return True
+
+
 def _summary(
     db: Session,
     pl: Playlist,
@@ -142,15 +187,7 @@ def create_playlist(
     name = (payload.get("name") or "").strip()
     if not name:
         raise HTTPException(status_code=400, detail="name is required")
-    next_pos = (
+    pl = Playlist(user_id=user.id, name=name, position=_next_playlist_position(db, user.id))
-        db.scalar(
-            select(func.coalesce(func.max(Playlist.position), -1)).where(
-                Playlist.user_id == user.id
-            )
-        )
-        + 1
-    )
-    pl = Playlist(user_id=user.id, name=name, position=next_pos)
     db.add(pl)
     db.commit()
     return _summary(db, pl, 0)
@@ -164,16 +201,11 @@ def _watch_later(db: Session, user: User) -> Playlist:
         )
     )
     if pl is None:
-        next_pos = (
-            db.scalar(
-                select(func.coalesce(func.max(Playlist.position), -1)).where(
-                    Playlist.user_id == user.id
-                )
-            )
-            + 1
-        )
         pl = Playlist(
-            user_id=user.id, name="Watch later", kind="watch_later", position=next_pos
+            user_id=user.id,
+            name="Watch later",
+            kind="watch_later",
+            position=_next_playlist_position(db, user.id),
         )
         db.add(pl)
         db.commit()
@@ -191,22 +223,7 @@ def add_watch_later(
     if not video_id or db.get(Video, video_id) is None:
         raise HTTPException(status_code=404, detail="Unknown video")
     pl = _watch_later(db, user)
-    exists = db.scalar(
+    _add_item(db, pl, video_id, mark_dirty=False)
-        select(PlaylistItem).where(
-            PlaylistItem.playlist_id == pl.id, PlaylistItem.video_id == video_id
-        )
-    )
-    if exists is None:
-        next_pos = (
-            db.scalar(
-                select(func.coalesce(func.max(PlaylistItem.position), -1)).where(
-                    PlaylistItem.playlist_id == pl.id
-                )
-            )
-            + 1
-        )
-        db.add(PlaylistItem(playlist_id=pl.id, video_id=video_id, position=next_pos))
-        db.commit()
     return {"saved": True, "playlist_id": pl.id}
 
 
@@ -344,23 +361,7 @@ def get_playlist(
     pl = _own_playlist(db, user, playlist_id)
     state = aliased(VideoState)
     rows = db.execute(
-        select(
+        select(*feed_columns(user, state))
-            Video.id,
-            Video.title,
-            Video.channel_id,
-            Channel.title.label("channel_title"),
-            Channel.thumbnail_url.label("channel_thumbnail"),
-            Channel.handle.label("channel_handle"),
-            Video.published_at,
-            Video.thumbnail_url,
-            Video.duration_seconds,
-            Video.view_count,
-            Video.is_short,
-            Video.live_status,
-            func.coalesce(state.status, "new").label("status"),
-            func.coalesce(state.position_seconds, 0).label("position_seconds"),
-            _saved_expr(user),
-        )
         .join(PlaylistItem, PlaylistItem.video_id == Video.id)
         .join(Channel, Channel.id == Video.channel_id)
         .outerjoin(state, and_(state.video_id == Video.id, state.user_id == user.id))
@@ -439,25 +440,7 @@ def add_item(
     video_id = payload.get("video_id")
     if not video_id or db.get(Video, video_id) is None:
         raise HTTPException(status_code=404, detail="Unknown video")
-    exists = db.scalar(
+    _add_item(db, pl, video_id, mark_dirty=True)
-        select(PlaylistItem).where(
-            PlaylistItem.playlist_id == pl.id, PlaylistItem.video_id == video_id
-        )
-    )
-    if exists is None:
-        next_pos = (
-            db.scalar(
-                select(func.coalesce(func.max(PlaylistItem.position), -1)).where(
-                    PlaylistItem.playlist_id == pl.id
-                )
-            )
-            + 1
-        )
-        db.add(
-            PlaylistItem(playlist_id=pl.id, video_id=video_id, position=next_pos)
-        )
-        recompute_dirty(db, pl)
-        db.commit()
     return {"playlist_id": pl.id, "video_id": video_id}
 
 

backend/app/routes/sync.py

@@ -3,7 +3,7 @@ from sqlalchemy import and_, case, func, select, update
 from sqlalchemy.orm import Session
 
 from app import quota, state
-from app.auth import current_user, has_read_scope, require_human
+from app.auth import admin_user, current_user, has_read_scope, require_human
 from app.db import get_db
 from app.models import Channel, Subscription, User, Video
 from app.sync.runner import (
@@ -40,11 +40,10 @@ def sync_subscriptions(
             status_code=403,
             detail="Connect your YouTube account (read access) to import subscriptions.",
         )
-    before = quota.units_used_today(db)
+    with quota.measured(db, user.id, quota.QuotaAction.SUBSCRIPTIONS_PULL) as m:
-    with quota.attribute(user.id, quota.QuotaAction.SUBSCRIPTIONS_PULL):
         result = import_subscriptions(db, user)
-    result["quota_used_estimate"] = quota.units_used_today(db) - before
+    result["quota_used_estimate"] = m.used
-    result["quota_remaining_today"] = quota.remaining_today(db)
+    result["quota_remaining_today"] = m.remaining
     return result
 
 
@@ -62,10 +61,9 @@ def sync_backfill(
     db: Session = Depends(get_db),
     max_channels: int = 25,
 ) -> dict:
-    before = quota.units_used_today(db)
+    with quota.measured(db, user.id, quota.QuotaAction.VIDEOS_BACKFILL_RECENT) as m:
-    with quota.attribute(user.id, quota.QuotaAction.VIDEOS_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
+    result["quota_used_estimate"] = m.used
     return result
 
 
@@ -73,13 +71,9 @@ def sync_backfill(
 def sync_enrich(
     user: User = Depends(require_human), db: Session = Depends(get_db)
 ) -> dict:
-    before = quota.units_used_today(db)
+    with quota.measured(db, user.id, quota.QuotaAction.VIDEOS_ENRICH) as m:
-    with quota.attribute(user.id, quota.QuotaAction.VIDEOS_ENRICH):
         enriched = run_enrich(db)
-    return {
+    return {"enriched": enriched, "quota_used_estimate": m.used}
-        "enriched": enriched,
-        "quota_used_estimate": quota.units_used_today(db) - before,
-    }
 
 
 @router.get("/status")
@@ -194,20 +188,12 @@ def deep_all(
 
 
 @router.post("/pause")
-def pause_sync(
+def pause_sync(_: User = Depends(admin_user), db: Session = Depends(get_db)) -> dict:
-    user: User = Depends(current_user), db: Session = Depends(get_db)
-) -> dict:
-    if user.role != "admin":
-        raise HTTPException(status_code=403, detail="Admin only")
     state.set_sync_paused(db, True)
     return {"paused": True}
 
 
 @router.post("/resume")
-def resume_sync(
+def resume_sync(_: User = Depends(admin_user), db: Session = Depends(get_db)) -> dict:
-    user: User = Depends(current_user), db: Session = Depends(get_db)
-) -> dict:
-    if user.role != "admin":
-        raise HTTPException(status_code=403, detail="Admin only")
     state.set_sync_paused(db, False)
     return {"paused": False}

backend/app/routes/tags.py

@@ -3,7 +3,7 @@ from sqlalchemy import func, or_, select
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
-from app.auth import current_user
+from app.auth import admin_user, current_user
 from app.db import get_db
 from app.models import ChannelTag, Tag, User
 from app.sync.autotag import run_autotag_all
@@ -103,10 +103,8 @@ def delete_tag(
 
 @router.post("/recompute")
 def recompute(
-    user: User = Depends(current_user),
+    _: User = Depends(admin_user),
     db: Session = Depends(get_db),
     only_missing: bool = False,
 ) -> dict:
-    if user.role != "admin":
-        raise HTTPException(status_code=403, detail="Admin only")
     return run_autotag_all(db, only_missing=only_missing)

backend/app/security.py

@@ -1,3 +1,5 @@
+import hashlib
+
 from argon2 import PasswordHasher
 from cryptography.fernet import Fernet
 
@@ -7,6 +9,12 @@ from app.config import settings
 _ph = PasswordHasher()
 
 
+def hash_token(raw: str) -> str:
+    """SHA-256 hex digest for storing single-use tokens (email-link tokens, the setup token)
+    by hash rather than in cleartext. Not a password hash — these are high-entropy randoms."""
+    return hashlib.sha256(raw.encode()).hexdigest()
+
+
 def hash_password(password: str) -> str:
     return _ph.hash(password)
 

backend/app/state.py

@@ -1,5 +1,4 @@
 """Global admin-controlled app state (e.g. pausing background sync, first-run setup)."""
-import hashlib
 import logging
 import secrets
 
@@ -7,6 +6,7 @@ from sqlalchemy.orm import Session
 
 from app.config import settings
 from app.models import AppState
+from app.security import hash_token
 
 log = logging.getLogger("siftlode.state")
 
@@ -73,16 +73,12 @@ def mark_configured(db: Session) -> None:
     log.info("Instance marked configured; setup wizard disabled.")
 
 
-def _hash_token(raw: str) -> str:
-    return hashlib.sha256(raw.encode()).hexdigest()
-
-
 def rotate_setup_token(db: Session) -> str:
     """Generate a fresh one-time setup token, persist only its hash, and return the raw token
     (logged at startup so the admin can open the wizard)."""
     raw = secrets.token_urlsafe(24)
     row = _row(db)
-    row.setup_token_hash = _hash_token(raw)
+    row.setup_token_hash = hash_token(raw)
     db.add(row)
     db.commit()
     return raw
@@ -93,7 +89,7 @@ def check_setup_token(db: Session, raw: str | None) -> bool:
     if not raw:
         return False
     stored = _row(db).setup_token_hash
-    return stored is not None and secrets.compare_digest(stored, _hash_token(raw))
+    return stored is not None and secrets.compare_digest(stored, hash_token(raw))
 
 
 def get_maintenance_batch(db: Session) -> int:

backend/app/sync/maintenance.py

@@ -20,18 +20,19 @@ Quota: phase 1 costs ~(flagged / 50) units, phase 2 ~(batch / 50). Both respect
 backfill reserve as the other jobs so interactive syncs never starve.
 """
 import logging
-from datetime import datetime, timedelta, timezone
+from datetime import timedelta
 
-from sqlalchemy import or_, select
+from sqlalchemy import select
 from sqlalchemy.orm import Session
 
 from app import progress, quota, sysconfig
 from app.config import settings
-from app.models import Playlist, PlaylistItem, Video, VideoState
+from app.models import Playlist, PlaylistItem, Video
 from app.notifications import create_notification
 from app.state import get_maintenance_batch
 from app.sync.runner import get_service_user
 from app.sync.videos import apply_video_details, parse_dt
+from app.utils import now_utc as _now
 from app.youtube.client import YouTubeClient
 
 log = logging.getLogger("siftlode.maintenance")
@@ -45,10 +46,6 @@ GRACE_DAYS_BY_REASON = {
 DEFAULT_GRACE_DAYS = settings.maintenance_grace_days
 
 
-def _now() -> datetime:
-    return datetime.now(timezone.utc)
-
-
 def _grace_days(reason: str | None) -> int:
     return GRACE_DAYS_BY_REASON.get(reason or "", DEFAULT_GRACE_DAYS)
 
@@ -91,14 +88,9 @@ def _collect_user_impact(db: Session, videos: list[Video]) -> dict[int, list[dic
     impact: dict[int, set[str]] = {}
     if not ids:
         return {}
-    # Saved (VideoState.status == "saved").
+    # Anyone who had it in a playlist — which now also covers "saved": Watch later became a
-    for user_id, video_id in db.execute(
+    # built-in playlist, so "saved" is no longer a VideoState status (the old status=="saved"
-        select(VideoState.user_id, VideoState.video_id).where(
+    # scan matched nothing and under-counted impacted users).
-            VideoState.video_id.in_(ids), VideoState.status == "saved"
-        )
-    ):
-        impact.setdefault(user_id, set()).add(video_id)
-    # In any of the user's playlists (covers watch-later, which is a playlist).
     for user_id, video_id in db.execute(
         select(Playlist.user_id, PlaylistItem.video_id)
         .join(PlaylistItem, PlaylistItem.playlist_id == Playlist.id)

backend/app/sync/playlists.py

@@ -94,6 +94,34 @@ def _ensure_videos(db: Session, yt: YouTubeClient, video_ids: list[str]) -> None
     db.commit()
 
 
+def _replace_items_from_youtube(
+    db: Session, yt: YouTubeClient, pl: Playlist, ids: list[str]
+) -> list[str]:
+    """Mirror a YouTube playlist's ordered video ids into `pl`'s local items (authoritative
+    replace): ensure the videos exist in the catalog, keep only those present in their YT
+    order (de-duplicated), swap the items out, and reset the synced fingerprint + dirty flag.
+    `pl.name` must already be set to the desired name. The caller commits. Returns the final
+    ordered ids."""
+    _ensure_videos(db, yt, ids)
+    present = (
+        set(db.execute(select(Video.id).where(Video.id.in_(ids))).scalars().all())
+        if ids
+        else set()
+    )
+    ordered: list[str] = []
+    seen: set[str] = set()
+    for v in ids:
+        if v in present and v not in seen:
+            seen.add(v)
+            ordered.append(v)
+    db.execute(delete(PlaylistItem).where(PlaylistItem.playlist_id == pl.id))
+    for pos, vid in enumerate(ordered):
+        db.add(PlaylistItem(playlist_id=pl.id, video_id=vid, position=pos))
+    pl.synced_fingerprint = fingerprint(pl.name, ordered)
+    pl.dirty = False
+    return ordered
+
+
 def sync_user_playlists(db: Session, user: User) -> dict:
     """Mirror the user's YouTube playlists into local source='youtube' playlists. Leaves
     the user's local playlists and the built-in Watch later untouched."""
@@ -152,25 +180,9 @@ def sync_user_playlists(db: Session, user: User) -> dict:
                 db.flush()
             else:
                 pl.name = p.get("title") or pl.name
-            ids = list(yt.iter_my_playlist_video_ids(ytid))
-            _ensure_videos(db, yt, ids)
-            present = (
-                set(db.execute(select(Video.id).where(Video.id.in_(ids))).scalars().all())
-                if ids
-                else set()
-            )
-            ordered: list[str] = []
-            seen: set[str] = set()
-            for v in ids:
-                if v in present and v not in seen:
-                    seen.add(v)
-                    ordered.append(v)
             # Mirror is authoritative: replace the items to match YouTube's order.
-            db.execute(delete(PlaylistItem).where(PlaylistItem.playlist_id == pl.id))
+            ids = list(yt.iter_my_playlist_video_ids(ytid))
-            for pos, vid in enumerate(ordered):
+            _replace_items_from_youtube(db, yt, pl, ids)
-                db.add(PlaylistItem(playlist_id=pl.id, video_id=vid, position=pos))
-            pl.synced_fingerprint = fingerprint(pl.name, ordered)
-            pl.dirty = False
             db.commit()
             synced += 1
     return {"synced": synced}
@@ -349,25 +361,9 @@ def repull_playlist(db: Session, user: User, pl: Playlist) -> dict:
         snippet = yt.get_playlist_snippet(pl.yt_playlist_id)
         if snippet is None:
             raise YouTubeError("Playlist no longer exists on YouTube")
-        ids = list(yt.iter_my_playlist_video_ids(pl.yt_playlist_id))
-        _ensure_videos(db, yt, ids)
-        present = (
-            set(db.execute(select(Video.id).where(Video.id.in_(ids))).scalars().all())
-            if ids
-            else set()
-        )
-        ordered: list[str] = []
-        seen: set[str] = set()
-        for v in ids:
-            if v in present and v not in seen:
-                seen.add(v)
-                ordered.append(v)
         pl.name = snippet.get("title") or pl.name
-        db.execute(delete(PlaylistItem).where(PlaylistItem.playlist_id == pl.id))
+        ids = list(yt.iter_my_playlist_video_ids(pl.yt_playlist_id))
-        for pos, vid in enumerate(ordered):
+        ordered = _replace_items_from_youtube(db, yt, pl, ids)
-            db.add(PlaylistItem(playlist_id=pl.id, video_id=vid, position=pos))
-        pl.synced_fingerprint = fingerprint(pl.name, ordered)
-        pl.dirty = False
         db.commit()
     return {"items": len(ordered)}
 

backend/app/utils.py

@@ -0,0 +1,17 @@
+"""Small cross-cutting helpers shared across modules (kept dependency-light on purpose)."""
+import re
+from datetime import datetime, timezone
+
+# Single source of truth for "looks like an email" — used by every endpoint that accepts an
+# address (auth register/reset/demo, admin invites + demo whitelist, the setup wizard). Keep
+# the strictness in one place so routes don't drift between this and a looser "@"-in check.
+_EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
+
+
+def valid_email(value: str | None) -> bool:
+    return bool(value and _EMAIL_RE.match(value))
+
+
+def now_utc() -> datetime:
+    """Current time as a timezone-aware UTC datetime."""
+    return datetime.now(timezone.utc)