feat: persistent in-app notification system

README.md

@@ -42,6 +42,7 @@ This template integrates the best-in-class Python ecosystem tools to provide a s
 - **Standardized API Responses:** Global exception handlers standardizing success/error schemas, utilizing a centralized `messages` module (`app/core/messages/`) to prevent hardcoded responses.
 - **Role-Based Access Control (RBAC):** Three roles — `user` / `admin` / `superadmin`. Superadmins bypass every check and **create/delete admins** and assign their permissions; plain admins are gated per **`resource:action`** permission (`users:read`, `users:write`, `support:update`, …) stored in `admin_permission` and enforced by a single `require_permission(...)` dependency. The **superadmin tier** (promote an admin to superadmin, demote a superadmin, transfer root via email OTP) is reserved for the **root superadmin** (`is_root_superadmin`); the last active superadmin is always protected. See [Admin & RBAC](#-admin--rbac-roles--permissions).
 - **Support / Ticketing System:** Full ticket lifecycle — open, reply with image attachments, status/priority, admin assignment — fanned out over a **Redis-backed WebSocket bus** so the ticket thread, the admin queue, and per-user feeds update live across workers. See [Support System & Realtime](#-support-ticketing-system--realtime).
+- **Notification System:** Persistent in-app notifications (`notification` table) layered on the same realtime bus — domain events (a support reply, a ticket status change, an RBAC grant change) call a single `notify()` use case that stores the row **and** pushes it to the recipient's `notifications:{id}` feed, so offline users still find it in their inbox on the next fetch. Notification copy is stored as a stable machine `type` + `data` payload (no human text), so the frontend renders it in any locale. REST covers paginated listing (`unread_only` filter), the unread badge count, and mark-(all-)read. See [Notification System](#-notification-system).
 - **First Superadmin Seed:** On startup a **root superadmin** (`is_root_superadmin`) is created from `FIRST_SUPERUSER` / `FIRST_SUPERUSER_PASSWORD` if none exists (a matching existing account is promoted instead; older deployments get their oldest superadmin flagged as root) — guaranteeing a root account.
 - **Tooling:** [uv](https://docs.astral.sh/uv/) for blazing-fast package management, and [Ruff](https://docs.astral.sh/ruff/) for linting and formatting.
 - **Testing:** Comprehensive async testing setup with `pytest` and `pytest-asyncio`, in-memory SQLite via `aiosqlite`, `fakeredis`, and autouse SMTP/MX patches — tests never hit Postgres or the network.
@@ -379,7 +380,25 @@ A complete support desk built on a generic realtime bus that any feature can reu
 | `WS /admin/support/ws` · `/support/ws` | admin / owner | Multiplexed ticket + queue feed. |
 | `WS /users/me/events` | any user | Per-user account feed (e.g. live `permissions_updated`). |
 
-**Realtime bus (`app/core/realtime.py`).** A thin in-process `ConnectionManager` sits behind a Redis pub/sub bridge: services `publish()` to a topic — `ticket:{id}`, `admin`, `user:{id}`, or `account:{id}` — and a per-worker listener fans the event to the matching local sockets, so an event raised in one worker reaches clients connected to any worker. Publishing is best-effort (`publish_safe`), so a realtime hiccup never breaks the originating request.
+**Realtime bus (`app/core/realtime.py`).** A thin in-process `ConnectionManager` sits behind a Redis pub/sub bridge: services `publish()` to a topic — `ticket:{id}`, `admin`, `user:{id}`, `account:{id}`, or `notifications:{id}` — and a per-worker listener fans the event to the matching local sockets, so an event raised in one worker reaches clients connected to any worker. Publishing is best-effort (`publish_safe`), so a realtime hiccup never breaks the originating request.
+
+---
+
+## 🔔 Notification System
+
+Persistent, per-user in-app notifications built on the same realtime bus as support — so an event raised in any worker reaches the recipient live, and is still waiting in their inbox if they were offline.
+
+| Method & path | Auth | Purpose |
+| --- | --- | --- |
+| `GET /notifications` | self | List own notifications, newest first (`skip` / `limit`, `unread_only` filter). |
+| `GET /notifications/unread-count` | self | Unread count for the bell badge. |
+| `POST /notifications/{id}/read` | owner | Mark one notification read (idempotent; IDOR-guarded). |
+| `POST /notifications/read-all` | self | Mark every unread notification read in one set-based `UPDATE`. |
+| `WS /notifications/ws` | self | Live feed — new notifications arrive as `notification_created` frames. |
+
+**Emitting (`app/use_cases/notify.py`).** Any domain service calls one cross-cutting `notify(session, user_id, type, data)` — it persists the row and best-effort `publish_safe`s it to the recipient's `notifications:{id}` topic (mirroring `log_activity`, so a service never has to call another service). It is wired into admin support replies (`support_ticket_replied`), ticket status changes (`support_ticket_status_changed`), and RBAC grant changes (`admin_permissions_changed`); self-actions are skipped so nobody is notified about their own action.
+
+**Locale-free storage.** A notification stores a stable machine `type` plus a `data` JSON payload (ids, subject, status) — never human text — so the frontend produces the message in the active language. Adding a type means adding an enum value plus a locale entry; the table schema never changes.
 
 ---
 
@@ -394,11 +413,11 @@ A complete support desk built on a generic realtime bus that any feature can reu
 │   │       └── admin/    # Admin surface — RBAC-gated via require_permission;
 │   │                     #   admins/ (admin management) is superadmin-only
 │   ├── core/             # config, db, security, redis, rate_limit, email, storage, realtime, bootstrap, messages/
-│   ├── models/           # Domain Layer: SQLAlchemy ORM (User, UserActivity, File, SupportTicket, AdminPermission, …)
+│   ├── models/           # Domain Layer: SQLAlchemy ORM (User, UserActivity, File, SupportTicket, Notification, AdminPermission, …)
 │   ├── repositories/     # Data Layer: async DB queries (no business rules)
 │   ├── schemas/          # Pydantic v2 DTOs (Create / Update / Response per domain)
 │   ├── services/         # Business Logic Layer: pure async functions, take AsyncSession
-│   ├── use_cases/        # Cross-domain orchestration (e.g. activity logging)
+│   ├── use_cases/        # Cross-domain orchestration (e.g. activity logging, notifications)
 │   ├── worker/           # arq worker — settings + cron jobs (e.g. account deletion)
 │   ├── utils/            # Helper functions (datetime, email templates)
 │   ├── tests/            # pytest suite (in-memory SQLite, fakeredis, mocked SMTP)
@@ -424,6 +443,7 @@ A complete support desk built on a generic realtime bus that any feature can reu
 - [ ] Create an admin account and assign `resource:action` permissions from the admin endpoints (superadmin-only)
 - [ ] Log in as that permission-limited admin and verify each route is gated by `require_permission(...)`
 - [ ] Try the support flow: open a ticket, then reply/assign it as an admin and watch the WebSocket bus push updates live
+- [ ] Reply to that ticket as an admin and confirm the owner receives a notification at `GET /notifications` (and live over the `notifications:{id}` feed)
 - [ ] Set Cloudinary keys (`CLOUDINARY_*`) if you need `POST /upload` for files/avatars
 - [ ] Run the test suite with `uv run pytest` to confirm everything is green
 - [ ] Add a new message constant to `app/core/messages/` before introducing any user-facing string

app/alembic/versions/5cd8e80e6bce_add_notification_model.py

@@ -0,0 +1,50 @@
+"""add notification model
+
+Revision ID: 5cd8e80e6bce
+Revises: 12501ac72e9d
+Create Date: 2026-06-11 17:08:36.094804
+
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision: str = "5cd8e80e6bce"
+down_revision: str | Sequence[str] | None = "12501ac72e9d"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        "notification",
+        sa.Column("id", sa.UUID(), nullable=False),
+        sa.Column("user_id", sa.Uuid(), nullable=False),
+        sa.Column("type", sa.String(length=50), nullable=False),
+        sa.Column("data", postgresql.JSON(astext_type=sa.Text()), nullable=False),
+        sa.Column("read_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.ForeignKeyConstraint(["user_id"], ["user.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index(
+        "ix_notification_user_created",
+        "notification",
+        ["user_id", "created_at"],
+        unique=False,
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_index("ix_notification_user_created", table_name="notification")
+    op.drop_table("notification")
+    # ### end Alembic commands ###

app/api/main.py

@@ -1,12 +1,15 @@
 from fastapi import APIRouter
 
-from app.api.routes import admin, auth, files, health, support, users
+from app.api.routes import admin, auth, files, health, notifications, support, users
 
 api_router = APIRouter()
 
 api_router.include_router(health.router, prefix="/health", tags=["health"])
 api_router.include_router(auth.router, prefix="/auth", tags=["auth"])
 api_router.include_router(users.router, prefix="/users", tags=["users"])
+api_router.include_router(
+    notifications.router, prefix="/notifications", tags=["notifications"]
+)
 api_router.include_router(support.router, prefix="/support", tags=["support"])
 api_router.include_router(admin.router, prefix="/admin", tags=["admin"])
 api_router.include_router(files.router, tags=["files"])

app/api/routes/notifications.py

@@ -0,0 +1,96 @@
+import uuid
+from typing import Annotated
+
+from fastapi import APIRouter, Query, Request, WebSocket
+
+from app.api.deps import CurrentActiveUser, SessionDep, get_ws_user
+from app.core.rate_limit import rate_limit_authenticated
+from app.core.realtime import notifications_topic, serve_account_socket
+from app.schemas.notification import (
+    NotificationListResponse,
+    NotificationReadResponse,
+    NotificationsMarkAllReadResponse,
+    UnreadCountResponse,
+)
+from app.services.notification_service import (
+    list_my_notifications_service,
+    mark_all_read_service,
+    mark_read_service,
+    unread_count_service,
+)
+
+router = APIRouter()
+
+# Same semantics as the support socket: auth fails before the handshake
+# completes, so we refuse with 1008 (Policy Violation) without accepting.
+_WS_POLICY_VIOLATION = 1008
+
+
+@router.get("", response_model=NotificationListResponse)
+async def list_my_notifications(
+    _request: Request,
+    session: SessionDep,
+    current_user: CurrentActiveUser,
+    skip: Annotated[int, Query(ge=0)] = 0,
+    limit: Annotated[int, Query(ge=1, le=100)] = 50,
+    unread_only: Annotated[bool, Query()] = False,
+) -> NotificationListResponse:
+    """List the caller's notifications, newest first."""
+    return await list_my_notifications_service(
+        session=session,
+        user=current_user,
+        skip=skip,
+        limit=limit,
+        unread_only=unread_only,
+    )
+
+
+@router.get("/unread-count", response_model=UnreadCountResponse)
+async def get_unread_count(
+    _request: Request,
+    session: SessionDep,
+    current_user: CurrentActiveUser,
+) -> UnreadCountResponse:
+    """Return the caller's unread notification count (bell badge)."""
+    return await unread_count_service(session=session, user=current_user)
+
+
+@router.post("/read-all", response_model=NotificationsMarkAllReadResponse)
+@rate_limit_authenticated("20/minute")
+async def mark_all_notifications_read(
+    request: Request,  # noqa: ARG001 - slowapi resolves the limit key by this name
+    session: SessionDep,
+    current_user: CurrentActiveUser,
+) -> NotificationsMarkAllReadResponse:
+    """Mark every unread notification of the caller as read."""
+    return await mark_all_read_service(session=session, user=current_user)
+
+
+@router.post("/{notification_id}/read", response_model=NotificationReadResponse)
+@rate_limit_authenticated("60/minute")
+async def mark_notification_read(
+    request: Request,  # noqa: ARG001 - slowapi resolves the limit key by this name
+    session: SessionDep,
+    current_user: CurrentActiveUser,
+    notification_id: uuid.UUID,
+) -> NotificationReadResponse:
+    """Mark one of the caller's notifications as read."""
+    return await mark_read_service(
+        session=session, user=current_user, notification_id=notification_id
+    )
+
+
+@router.websocket("/ws")
+async def notifications_ws(websocket: WebSocket, session: SessionDep) -> None:
+    """Live notification feed for the authenticated user.
+
+    Authenticates from the ``access_token`` cookie and subscribes the socket to
+    the caller's ``notifications:{id}`` topic. The client sends nothing; new
+    notifications arrive as ``NotificationRealtimeEvent`` frames.
+    """
+    user = await get_ws_user(websocket, session)
+    if user is None:
+        await websocket.close(code=_WS_POLICY_VIOLATION)
+        return
+
+    await serve_account_socket(websocket, topic=notifications_topic(user.id))

app/core/messages/error_message.py

@@ -61,6 +61,10 @@ class ErrorMessages:
     FILE_FORBIDDEN = "error.file.forbidden"
     FILE_UPLOAD_FAILED = "error.file.upload_failed"
 
+    # Notifications
+    NOTIFICATION_NOT_FOUND = "error.notification.not_found"
+    NOTIFICATION_ACCESS_DENIED = "error.notification.access_denied"
+
     # Support / tickets
     TICKET_NOT_FOUND = "error.support.ticket_not_found"
     TICKET_ACCESS_DENIED = "error.support.ticket_access_denied"

app/core/messages/success_message.py

@@ -37,6 +37,10 @@ class SuccessMessages:
     FILE_UPLOADED = "success.file.uploaded"
     ADMIN_FILE_DELETED = "success.admin.file_deleted"
 
+    # Notifications
+    NOTIFICATION_READ = "success.notification.read"
+    NOTIFICATIONS_ALL_READ = "success.notification.all_read"
+
     # Support / tickets
     TICKET_CREATED = "success.support.ticket_created"
     TICKET_CLOSED = "success.support.ticket_closed"

app/core/realtime.py

@@ -11,6 +11,7 @@
   * ``admin``          — the global admin feed (new tickets, status changes)
   * ``user:{uuid}``    — a single user's support feed (across all their tickets)
   * ``account:{uuid}`` — a single user's account feed (e.g. RBAC permission changes)
+  * ``notifications:{uuid}`` — a single user's persistent in-app notification feed
 
 The Redis channel is the topic prefixed with ``rt:``.
 """
@@ -124,6 +125,11 @@ def account_topic(user_id: uuid.UUID) -> str:
     return f"account:{user_id}"
 
 
+def notifications_topic(user_id: uuid.UUID) -> str:
+    """Realtime topic carrying a single user's in-app notifications."""
+    return f"notifications:{user_id}"
+
+
 async def publish_feeds(owner_id: uuid.UUID, event: RealtimeEvent) -> None:
     """Fan a ticket-summary event out to both the admin queue and its owner's feed.
 

app/models/__init__.py

@@ -1,5 +1,6 @@
 from app.models.admin_permission import AdminPermission
 from app.models.file import File
+from app.models.notification import Notification
 from app.models.support import (
     SupportMessage,
     SupportMessageAttachment,
@@ -11,6 +12,7 @@
 __all__ = [
     "AdminPermission",
     "File",
+    "Notification",
     "SupportMessage",
     "SupportMessageAttachment",
     "SupportTicket",

app/models/notification.py

@@ -0,0 +1,56 @@
+import uuid
+from datetime import datetime
+from typing import TYPE_CHECKING
+
+from sqlalchemy import DateTime, ForeignKey, Index, String
+from sqlalchemy.dialects.postgresql import JSON
+from sqlalchemy.dialects.postgresql import UUID as PG_UUID
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from app.core.db import Base
+from app.schemas.common import JsonValue
+from app.utils import utc_now
+
+if TYPE_CHECKING:
+    from app.models.user import User
+
+
+class Notification(Base):
+    """A persistent in-app notification addressed to a single user.
+
+    ``type`` is a stable machine code (e.g. ``support_ticket_replied``) that the
+    frontend maps to a translated message; no human-readable text is stored so
+    notifications render correctly in every locale. ``data`` carries the
+    type-specific payload (ids, names) needed to build the message and link.
+    """
+
+    __tablename__ = "notification"
+    __table_args__ = (
+        # Inbox listing: "my notifications, newest first". The user_id prefix
+        # also serves the unread-count query (WHERE user_id = ? AND read_at IS NULL).
+        Index("ix_notification_user_created", "user_id", "created_at"),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        PG_UUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+    # Recipient.
+    user_id: Mapped[uuid.UUID] = mapped_column(
+        ForeignKey("user.id", ondelete="CASCADE"),
+        nullable=False,
+    )
+    # Stored as a plain string (mirroring ``User.role``) and validated at the
+    # schema layer, so adding a new notification type never needs a migration.
+    type: Mapped[str] = mapped_column(String(50), nullable=False)
+    data: Mapped[dict[str, JsonValue]] = mapped_column(
+        JSON, default=dict, nullable=False
+    )
+    # Set when the user opens/acknowledges the notification; powers unread counts.
+    read_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), default=None
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), default=utc_now, nullable=False
+    )
+
+    user: Mapped["User"] = relationship("User")