feat(notifications): durable per-user inbox (P1) + maintenance schema

Add a server-backed notification center that coexists with the client-side
transient bell: a per-user `notifications` table (type/title/body/data JSON/
read/dismissed), a `/api/me/notifications` CRUD API (list, unread_count, read,
read_all, dismiss, clear), and a left-nav inbox module with a live unread badge
polled via useLiveQuery. Known types render trilingual text from type+data
(English stored text is the fallback); read rows are trimmed past a soft cap.

Also adds the schema the maintenance job builds on: videos.list?part=status
columns (embeddable/privacy_status/upload_status) and the validation lifecycle
columns (last_checked_at, unavailable_since, unavailable_reason).

Page-id validation is centralized in one PAGES source of truth (isPage) so the
new page survives reload without a second allowlist to keep in sync.

backend/app/main.py

@@ -32,6 +32,7 @@ from app.routes import (
     feed,
     health,
     me,
+    notifications,
     playlists,
     quota,
     scheduler as scheduler_routes,
@@ -77,6 +78,7 @@ app.include_router(sync.router)
 app.include_router(tags.router)
 app.include_router(feed.router)
 app.include_router(me.router)
+app.include_router(notifications.router)
 app.include_router(channels.router)
 app.include_router(playlists.router)
 app.include_router(admin.router)

backend/app/models.py

@@ -203,6 +203,22 @@ class Video(Base):
         String(16), default="none", server_default="none", index=True
     )
     enriched_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    # videos.list?part=status fields (populated during enrichment); used by the
+    # maintenance/validation job to decide whether a video is playable anywhere.
+    embeddable: Mapped[bool | None] = mapped_column(Boolean)
+    privacy_status: Mapped[str | None] = mapped_column(String(16))  # public|unlisted|private
+    upload_status: Mapped[str | None] = mapped_column(String(16))  # processed|failed|rejected|deleted
+    # Maintenance/validation lifecycle. last_checked_at drives the rolling re-validation
+    # (least-recently-checked first); unavailable_since marks a video as currently
+    # unplayable (hidden from the feed immediately, hard-deleted after a grace period);
+    # unavailable_reason records why (deleted|private|paywalled|abandoned|stuck_live).
+    last_checked_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), index=True
+    )
+    unavailable_since: Mapped[datetime | None] = mapped_column(
+        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()
     )
@@ -399,3 +415,36 @@ class PlaylistItem(Base):
     )
 
     playlist: Mapped["Playlist"] = relationship(back_populates="items")
+
+
+class Notification(Base):
+    """A durable, server-backed per-user notification (the inbox center, phase 1).
+
+    Distinct from the client-side transient "bell" (toasts kept only in localStorage):
+    these survive reloads and devices, and are produced server-side (first producer = the
+    maintenance/validation job's batched "N saved/playlisted videos removed" notice). The
+    column is named `data` (not `metadata`, which SQLAlchemy's declarative Base reserves):
+    a free-form JSON payload (e.g. the list of removed videos) the UI can render. `read`
+    and `dismissed` are separate: read clears the unread badge but keeps the row in the
+    inbox; dismissed removes it from the default view. Unread rows never auto-expire; read
+    ones are trimmed past a soft per-user cap (see app.notifications.trim_read)."""
+
+    __tablename__ = "notifications"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    user_id: Mapped[int] = mapped_column(
+        ForeignKey("users.id", ondelete="CASCADE"), index=True
+    )
+    type: Mapped[str] = mapped_column(String(32), index=True)  # e.g. "maintenance"
+    title: Mapped[str] = mapped_column(String(255))
+    body: Mapped[str | None] = mapped_column(Text)
+    data: Mapped[dict | None] = mapped_column(JSON)
+    read: Mapped[bool] = mapped_column(
+        Boolean, default=False, server_default="false", index=True
+    )
+    dismissed: Mapped[bool] = mapped_column(
+        Boolean, default=False, server_default="false"
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), index=True
+    )

backend/app/notifications.py

@@ -0,0 +1,58 @@
+"""Helpers for the durable, server-backed per-user notification inbox (phase 1).
+
+The inbox COEXISTS with the client-side transient "bell" (localStorage toasts): these rows
+survive reloads/devices and are produced server-side. First producer is the maintenance
+job. Unread rows never auto-expire; read rows are trimmed past a soft per-user cap so the
+table can't grow without bound for a heavy user.
+"""
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.models import Notification
+
+# Keep at most this many READ notifications per user; older read ones are pruned. Unread
+# rows are never pruned (the user hasn't seen them yet).
+READ_SOFT_CAP = 200
+
+
+def create_notification(
+    db: Session,
+    user_id: int,
+    type: str,
+    title: str,
+    body: str | None = None,
+    data: dict | None = None,
+    *,
+    commit: bool = True,
+) -> Notification:
+    """Insert one notification for a user. Callers that batch many inserts in a single
+    transaction can pass commit=False and commit once themselves."""
+    notif = Notification(
+        user_id=user_id, type=type, title=title, body=body, data=data
+    )
+    db.add(notif)
+    if commit:
+        db.commit()
+        db.refresh(notif)
+    return notif
+
+
+def trim_read(db: Session, user_id: int, cap: int = READ_SOFT_CAP) -> int:
+    """Delete a user's oldest READ notifications beyond `cap`. Returns the number removed.
+    Idempotent and cheap; safe to call after marking things read."""
+    ids = (
+        db.execute(
+            select(Notification.id)
+            .where(Notification.user_id == user_id, Notification.read.is_(True))
+            .order_by(Notification.created_at.desc())
+            .offset(cap)
+        )
+        .scalars()
+        .all()
+    )
+    if not ids:
+        return 0
+    for nid in ids:
+        db.delete(db.get(Notification, nid))
+    db.commit()
+    return len(ids)

backend/app/routes/notifications.py

@@ -0,0 +1,136 @@
+"""Per-user notification inbox (phase 1): the durable, server-backed center that coexists
+with the client-side transient bell. Read-only listing plus read/dismiss/clear mutations.
+
+All endpoints use `current_user` (not `require_human`): notifications and their read/dismiss
+state are personal UI state with no quota or YouTube involvement, so the shared demo account
+manages its own inbox normally."""
+from fastapi import APIRouter, Depends, HTTPException, Query
+from sqlalchemy import func, select, update
+from sqlalchemy.orm import Session
+
+from app.auth import current_user
+from app.db import get_db
+from app.models import Notification, User
+from app.notifications import trim_read
+
+router = APIRouter(prefix="/api/me/notifications", tags=["notifications"])
+
+
+def _serialize(n: Notification) -> dict:
+    return {
+        "id": n.id,
+        "type": n.type,
+        "title": n.title,
+        "body": n.body,
+        "data": n.data,
+        "read": n.read,
+        "dismissed": n.dismissed,
+        "created_at": n.created_at.isoformat() if n.created_at else None,
+    }
+
+
+@router.get("")
+def list_notifications(
+    include_dismissed: bool = False,
+    limit: int = Query(default=100, le=500),
+    offset: int = 0,
+    user: User = Depends(current_user),
+    db: Session = Depends(get_db),
+) -> dict:
+    where = [Notification.user_id == user.id]
+    if not include_dismissed:
+        where.append(Notification.dismissed.is_(False))
+    total = db.scalar(
+        select(func.count()).select_from(Notification).where(*where)
+    ) or 0
+    rows = (
+        db.execute(
+            select(Notification)
+            .where(*where)
+            .order_by(Notification.created_at.desc())
+            .offset(offset)
+            .limit(limit)
+        )
+        .scalars()
+        .all()
+    )
+    return {"items": [_serialize(n) for n in rows], "total": total}
+
+
+@router.get("/unread_count")
+def unread_count(
+    user: User = Depends(current_user), db: Session = Depends(get_db)
+) -> dict:
+    """Cheap poll target for the nav badge (uses the (user_id, read) index)."""
+    count = db.scalar(
+        select(func.count())
+        .select_from(Notification)
+        .where(
+            Notification.user_id == user.id,
+            Notification.read.is_(False),
+            Notification.dismissed.is_(False),
+        )
+    )
+    return {"count": count or 0}
+
+
+def _owned(db: Session, notification_id: int, user: User) -> Notification:
+    n = db.get(Notification, notification_id)
+    if n is None or n.user_id != user.id:
+        raise HTTPException(status_code=404, detail="Notification not found")
+    return n
+
+
+@router.post("/{notification_id}/read")
+def mark_read(
+    notification_id: int,
+    user: User = Depends(current_user),
+    db: Session = Depends(get_db),
+) -> dict:
+    n = _owned(db, notification_id, user)
+    n.read = True
+    db.commit()
+    trim_read(db, user.id)
+    return {"ok": True}
+
+
+@router.post("/read_all")
+def mark_all_read(
+    user: User = Depends(current_user), db: Session = Depends(get_db)
+) -> dict:
+    db.execute(
+        update(Notification)
+        .where(Notification.user_id == user.id, Notification.read.is_(False))
+        .values(read=True)
+    )
+    db.commit()
+    trim_read(db, user.id)
+    return {"ok": True}
+
+
+@router.post("/{notification_id}/dismiss")
+def dismiss(
+    notification_id: int,
+    user: User = Depends(current_user),
+    db: Session = Depends(get_db),
+) -> dict:
+    n = _owned(db, notification_id, user)
+    # Dismiss implies read (it no longer counts toward the unread badge).
+    n.dismissed = True
+    n.read = True
+    db.commit()
+    return {"ok": True}
+
+
+@router.post("/clear")
+def clear_all(
+    user: User = Depends(current_user), db: Session = Depends(get_db)
+) -> dict:
+    """Dismiss every (still-visible) notification for this user."""
+    db.execute(
+        update(Notification)
+        .where(Notification.user_id == user.id, Notification.dismissed.is_(False))
+        .values(dismissed=True, read=True)
+    )
+    db.commit()
+    return {"ok": True}