diff --git a/README.md b/README.md
index 7421249..73537e6 100644
--- a/README.md
+++ b/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
diff --git a/app/alembic/versions/5cd8e80e6bce_add_notification_model.py b/app/alembic/versions/5cd8e80e6bce_add_notification_model.py
new file mode 100644
index 0000000..0d682b1
--- /dev/null
+++ b/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 ###
diff --git a/app/api/main.py b/app/api/main.py
index 750698f..8769eb3 100644
--- a/app/api/main.py
+++ b/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"])
diff --git a/app/api/routes/notifications.py b/app/api/routes/notifications.py
new file mode 100644
index 0000000..2fe23b2
--- /dev/null
+++ b/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))
diff --git a/app/core/messages/error_message.py b/app/core/messages/error_message.py
index 82f42cc..ee7e5c7 100644
--- a/app/core/messages/error_message.py
+++ b/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"
diff --git a/app/core/messages/success_message.py b/app/core/messages/success_message.py
index 627a2bb..57ed778 100644
--- a/app/core/messages/success_message.py
+++ b/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"
diff --git a/app/core/realtime.py b/app/core/realtime.py
index d99ec88..05aaaff 100644
--- a/app/core/realtime.py
+++ b/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.
 
diff --git a/app/models/__init__.py b/app/models/__init__.py
index c71968a..b0d17c3 100644
--- a/app/models/__init__.py
+++ b/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",
diff --git a/app/models/notification.py b/app/models/notification.py
new file mode 100644
index 0000000..172955f
--- /dev/null
+++ b/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")
diff --git a/app/repositories/notification.py b/app/repositories/notification.py
new file mode 100644
index 0000000..118ae35
--- /dev/null
+++ b/app/repositories/notification.py
@@ -0,0 +1,104 @@
+import uuid
+from collections.abc import Sequence
+
+from sqlalchemy import func, select, update
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.notification import Notification
+from app.utils import utc_now
+
+
+async def create_notification(
+    session: AsyncSession, notification: Notification
+) -> Notification:
+    """Persist a new notification."""
+    session.add(notification)
+    await session.commit()
+    await session.refresh(notification)
+    return notification
+
+
+async def get_notification(
+    session: AsyncSession, notification_id: uuid.UUID
+) -> Notification | None:
+    """Get a single notification by id."""
+    return await session.get(Notification, notification_id)
+
+
+async def list_notifications(
+    session: AsyncSession,
+    *,
+    user_id: uuid.UUID,
+    skip: int = 0,
+    limit: int = 50,
+    unread_only: bool = False,
+) -> tuple[Sequence[Notification], int]:
+    """Return a user's notifications (newest first) plus total count."""
+    base_stmt = select(Notification).where(Notification.user_id == user_id)
+    if unread_only:
+        base_stmt = base_stmt.where(Notification.read_at.is_(None))
+
+    count_stmt = base_stmt.with_only_columns(
+        func.count(), maintain_column_froms=True
+    ).order_by(None)
+    total = (await session.execute(count_stmt)).scalar_one()
+
+    list_stmt = (
+        base_stmt.order_by(Notification.created_at.desc()).offset(skip).limit(limit)
+    )
+    notifications = (await session.execute(list_stmt)).scalars().all()
+    return notifications, total
+
+
+async def count_unread(session: AsyncSession, *, user_id: uuid.UUID) -> int:
+    """Count a user's unread notifications (badge counter)."""
+    statement = (
+        select(func.count())
+        .select_from(Notification)
+        .where(
+            Notification.user_id == user_id,
+            Notification.read_at.is_(None),
+        )
+    )
+    return (await session.execute(statement)).scalar_one()
+
+
+async def mark_read(
+    session: AsyncSession, *, notification_id: uuid.UUID
+) -> Notification | None:
+    """Stamp a single notification as read and return the updated row.
+
+    Set-based so it never touches possibly-expired ORM attributes; the
+    ``read_at IS NULL`` guard makes it idempotent (the original read time is
+    never overwritten). Returns ``None`` if the notification no longer exists.
+    """
+    statement = (
+        update(Notification)
+        .where(
+            Notification.id == notification_id,
+            Notification.read_at.is_(None),
+        )
+        .values(read_at=utc_now())
+    )
+    await session.execute(statement)
+    await session.commit()
+    return await session.get(Notification, notification_id)
+
+
+async def mark_all_read(session: AsyncSession, *, user_id: uuid.UUID) -> int:
+    """Mark every unread notification of a user as read in one bulk UPDATE.
+
+    A set-based UPDATE (rather than load-and-loop) because the inbox is
+    unbounded; returns the number of rows updated.
+    """
+    statement = (
+        update(Notification)
+        .where(
+            Notification.user_id == user_id,
+            Notification.read_at.is_(None),
+        )
+        .values(read_at=utc_now())
+    )
+    result = await session.execute(statement)
+    await session.commit()
+    return result.rowcount
diff --git a/app/schemas/notification.py b/app/schemas/notification.py
new file mode 100644
index 0000000..8b5af15
--- /dev/null
+++ b/app/schemas/notification.py
@@ -0,0 +1,76 @@
+import uuid
+from datetime import datetime
+from enum import StrEnum
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from app.schemas.common import JsonValue
+
+
+class NotificationType(StrEnum):
+    """Machine codes for notification kinds.
+
+    Each value doubles as the frontend translation key, so adding a type means
+    adding it here plus a locale entry — the database schema never changes.
+    """
+
+    SUPPORT_TICKET_REPLIED = "support_ticket_replied"
+    SUPPORT_TICKET_STATUS_CHANGED = "support_ticket_status_changed"
+    ADMIN_PERMISSIONS_CHANGED = "admin_permissions_changed"
+
+
+class NotificationRead(BaseModel):
+    """A single notification as returned to its recipient."""
+
+    model_config = ConfigDict(from_attributes=True)
+
+    id: uuid.UUID
+    type: NotificationType
+    data: dict[str, JsonValue] = Field(default_factory=dict)
+    read_at: datetime | None = None
+    created_at: datetime
+
+
+class NotificationListResponse(BaseModel):
+    """Paginated listing of the caller's notifications."""
+
+    data: list[NotificationRead]
+    total: int
+    skip: int
+    limit: int
+
+
+class UnreadCountResponse(BaseModel):
+    """Number of unread notifications for the caller (badge counter)."""
+
+    unread_count: int
+
+
+class NotificationReadResponse(BaseModel):
+    """Standard response after marking a single notification as read."""
+
+    data: NotificationRead
+    message: str
+
+
+class NotificationsMarkAllReadResponse(BaseModel):
+    """Standard response after marking every unread notification as read."""
+
+    updated: int
+    message: str
+
+
+# --- Realtime (WebSocket / Redis pub-sub) ----------------------------------
+
+
+class NotificationEventType(StrEnum):
+    """Kinds of events pushed over the notification WebSocket."""
+
+    NOTIFICATION_CREATED = "notification_created"
+
+
+class NotificationRealtimeEvent(BaseModel):
+    """Envelope broadcast to a user's notification feed on new notifications."""
+
+    type: NotificationEventType
+    notification: NotificationRead
diff --git a/app/services/admin/admin_service.py b/app/services/admin/admin_service.py
index 2ea2970..0af4ff0 100644
--- a/app/services/admin/admin_service.py
+++ b/app/services/admin/admin_service.py
@@ -44,9 +44,11 @@
 )
 from app.schemas.admin_permission import Permission
 from app.schemas.msg import Message
+from app.schemas.notification import NotificationType
 from app.schemas.user import Language, SystemRole
 from app.schemas.user_activity import ActivityType, ResourceType
 from app.use_cases.log_activity import log_activity
+from app.use_cases.notify import notify
 from app.utils.email_templates import generate_root_transfer_otp_email
 
 _ROOT_TRANSFER_OTP_TTL_SECONDS = 180
@@ -74,16 +76,27 @@ def _effective_permissions(user: User, granted: list[Permission]) -> list[Permis
     return granted
 
 
-async def _notify_permissions_changed(user_id: uuid.UUID) -> None:
-    """Push a best-effort ``permissions_updated`` event to the user's socket.
+async def _notify_permissions_changed(
+    session: AsyncSession, user_id: uuid.UUID, *, action: str
+) -> None:
+    """Signal a permission change live and record it in the user's inbox.
 
-    The browser uses this as a signal to refetch ``/users/me`` so a permission
-    grant or revoke takes effect immediately without a re-login.
+    The account-socket event makes the browser refetch ``/users/me`` so the
+    grant or revoke takes effect immediately without a re-login; the persistent
+    notification tells the user *that* it happened even if they were offline.
+    ``action`` mirrors the audit-log action string so the frontend can phrase
+    the message per change kind.
     """
     await publish_safe(
         account_topic(user_id),
         AccountEvent(type=AccountEventType.PERMISSIONS_UPDATED),
     )
+    await notify(
+        session,
+        user_id=user_id,
+        notification_type=NotificationType.ADMIN_PERMISSIONS_CHANGED,
+        data={"action": action},
+    )
 
 
 def get_permission_catalog_service() -> PermissionCatalogResponse:
@@ -202,7 +215,9 @@ async def promote_admin_to_superadmin_service(
         details={"action": "promoted_to_superadmin"},
         request=request,
     )
-    await _notify_permissions_changed(target.id)
+    await _notify_permissions_changed(
+        session, target.id, action="promoted_to_superadmin"
+    )
 
     return AdminMutationResponse(
         admin=_to_list_item(target, _effective_permissions(target, [])),
@@ -257,7 +272,9 @@ async def demote_superadmin_service(
         details={"action": "demoted_superadmin_to_admin"},
         request=request,
     )
-    await _notify_permissions_changed(target.id)
+    await _notify_permissions_changed(
+        session, target.id, action="demoted_superadmin_to_admin"
+    )
 
     return AdminMutationResponse(
         admin=_to_list_item(target, []),
@@ -306,7 +323,9 @@ async def update_admin_permissions_service(
         },
         request=request,
     )
-    await _notify_permissions_changed(target.id)
+    await _notify_permissions_changed(
+        session, target.id, action="set_admin_permissions"
+    )
 
     return AdminMutationResponse(
         admin=_to_list_item(target, permissions),
@@ -480,8 +499,10 @@ async def confirm_root_transfer_service(
         details={"action": "root_transferred", "new_root": str(target.id)},
         request=request,
     )
-    await _notify_permissions_changed(target.id)
-    await _notify_permissions_changed(current_user.id)
+    await _notify_permissions_changed(session, target.id, action="root_transferred")
+    await _notify_permissions_changed(
+        session, current_user.id, action="root_transferred"
+    )
 
     return AdminMutationResponse(
         admin=_to_list_item(target, _effective_permissions(target, [])),
diff --git a/app/services/admin/support_service.py b/app/services/admin/support_service.py
index a35536a..038bb69 100644
--- a/app/services/admin/support_service.py
+++ b/app/services/admin/support_service.py
@@ -23,6 +23,7 @@
 )
 from app.repositories.user import get_user_by_id
 from app.schemas.file import FileCategory
+from app.schemas.notification import NotificationType
 from app.schemas.support import (
     AdminTicketDetail,
     AdminTicketListItem,
@@ -41,6 +42,7 @@
 from app.schemas.user import SystemRole
 from app.schemas.user_activity import ActivityType, ResourceType
 from app.use_cases.log_activity import log_activity
+from app.use_cases.notify import notify
 from app.use_cases.support_attachments import resolve_attachment_files
 from app.utils import utc_now
 
@@ -249,6 +251,16 @@ async def reply_ticket_admin_service(
             ),
         )
 
+    # Inbox notification for the ticket owner — skipped when an admin replies
+    # to their own ticket, so nobody is notified about their own action.
+    if ticket.user_id != admin.id:
+        await notify(
+            session,
+            user_id=ticket.user_id,
+            notification_type=NotificationType.SUPPORT_TICKET_REPLIED,
+            data={"ticket_id": str(ticket.id), "subject": ticket.subject},
+        )
+
     return SupportMessageResponse(
         data=message_read,
         message=SuccessMessages.ADMIN_TICKET_REPLIED,
@@ -265,6 +277,7 @@ async def update_ticket_admin_service(
 ) -> AdminTicketResponse:
     """Change a ticket's status, priority, or assignment."""
     ticket = await _load_ticket_or_404(session, ticket_id)
+    old_status = ticket.status
 
     update_data = payload.model_dump(exclude_unset=True)
 
@@ -299,6 +312,24 @@ async def update_ticket_admin_service(
 
     await _publish_ticket_update(session, ticket)
 
+    # Inbox notification for the owner when the status actually changed —
+    # priority/assignment shuffles are admin-internal and stay silent.
+    if (
+        payload.status is not None
+        and payload.status.value != old_status
+        and ticket.user_id != admin.id
+    ):
+        await notify(
+            session,
+            user_id=ticket.user_id,
+            notification_type=NotificationType.SUPPORT_TICKET_STATUS_CHANGED,
+            data={
+                "ticket_id": str(ticket.id),
+                "subject": ticket.subject,
+                "status": payload.status.value,
+            },
+        )
+
     detail = await _serialize_admin_detail(session, ticket_id)
     return AdminTicketResponse(
         ticket=detail, message=SuccessMessages.ADMIN_TICKET_UPDATED
diff --git a/app/services/notification_service.py b/app/services/notification_service.py
new file mode 100644
index 0000000..eb748ea
--- /dev/null
+++ b/app/services/notification_service.py
@@ -0,0 +1,94 @@
+import uuid
+
+from fastapi import HTTPException, status
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.messages.error_message import ErrorMessages
+from app.core.messages.success_message import SuccessMessages
+from app.models.notification import Notification
+from app.models.user import User
+from app.repositories.notification import (
+    count_unread,
+    get_notification,
+    list_notifications,
+    mark_all_read,
+    mark_read,
+)
+from app.schemas.notification import (
+    NotificationListResponse,
+    NotificationRead,
+    NotificationReadResponse,
+    NotificationsMarkAllReadResponse,
+    UnreadCountResponse,
+)
+
+
+async def _load_owned_notification(
+    session: AsyncSession, *, notification_id: uuid.UUID, user: User
+) -> Notification:
+    """Fetch a notification and assert the caller owns it, else raise 404/403."""
+    notification = await get_notification(session, notification_id)
+    if notification is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=ErrorMessages.NOTIFICATION_NOT_FOUND,
+        )
+    if notification.user_id != user.id:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail=ErrorMessages.NOTIFICATION_ACCESS_DENIED,
+        )
+    return notification
+
+
+async def list_my_notifications_service(
+    session: AsyncSession,
+    *,
+    user: User,
+    skip: int,
+    limit: int,
+    unread_only: bool = False,
+) -> NotificationListResponse:
+    """Return the caller's notifications, newest first."""
+    notifications, total = await list_notifications(
+        session, user_id=user.id, skip=skip, limit=limit, unread_only=unread_only
+    )
+    items = [NotificationRead.model_validate(n) for n in notifications]
+    return NotificationListResponse(data=items, total=total, skip=skip, limit=limit)
+
+
+async def unread_count_service(
+    session: AsyncSession, *, user: User
+) -> UnreadCountResponse:
+    """Return the caller's unread notification count (badge counter)."""
+    unread = await count_unread(session, user_id=user.id)
+    return UnreadCountResponse(unread_count=unread)
+
+
+async def mark_read_service(
+    session: AsyncSession, *, user: User, notification_id: uuid.UUID
+) -> NotificationReadResponse:
+    """Mark one of the caller's notifications as read."""
+    await _load_owned_notification(session, notification_id=notification_id, user=user)
+    updated = await mark_read(session, notification_id=notification_id)
+    if updated is None:
+        # Deleted between the ownership check and the update — treat as gone.
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=ErrorMessages.NOTIFICATION_NOT_FOUND,
+        )
+    return NotificationReadResponse(
+        data=NotificationRead.model_validate(updated),
+        message=SuccessMessages.NOTIFICATION_READ,
+    )
+
+
+async def mark_all_read_service(
+    session: AsyncSession, *, user: User
+) -> NotificationsMarkAllReadResponse:
+    """Mark every unread notification of the caller as read."""
+    updated = await mark_all_read(session, user_id=user.id)
+    return NotificationsMarkAllReadResponse(
+        updated=updated,
+        message=SuccessMessages.NOTIFICATIONS_ALL_READ,
+    )
diff --git a/app/tests/notifications/__init__.py b/app/tests/notifications/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/app/tests/notifications/conftest.py b/app/tests/notifications/conftest.py
new file mode 100644
index 0000000..7edb11d
--- /dev/null
+++ b/app/tests/notifications/conftest.py
@@ -0,0 +1,43 @@
+"""Shared fixtures and helpers for the /notifications endpoint tests.
+
+Reuses the support test client factory (multiple logged-in identities per
+test) and adds seeding straight through the real ``notify`` use case, so every
+test row went through the same code path production uses.
+"""
+
+import uuid
+
+from httpx import AsyncClient
+
+from app.models.notification import Notification
+from app.schemas.common import JsonValue
+from app.schemas.notification import NotificationType
+from app.tests.conftest import TestingSessionLocal
+from app.tests.support.conftest import (  # noqa: F401 - re-exported fixture
+    ClientFactory,
+    client_factory,
+    make_user_client,
+)
+from app.use_cases.notify import notify
+
+__all__ = ["ClientFactory", "make_user_client", "get_user_id", "seed_notification"]
+
+
+async def get_user_id(client: AsyncClient) -> uuid.UUID:
+    """Return the authenticated client's own user id via /users/me."""
+    response = await client.get("/users/me")
+    assert response.status_code == 200, response.text
+    return uuid.UUID(response.json()["id"])
+
+
+async def seed_notification(
+    user_id: uuid.UUID,
+    *,
+    notification_type: NotificationType = NotificationType.SUPPORT_TICKET_REPLIED,
+    data: dict[str, JsonValue] | None = None,
+) -> Notification:
+    """Persist one notification for ``user_id`` through the real use case."""
+    async with TestingSessionLocal() as session:
+        return await notify(
+            session, user_id=user_id, notification_type=notification_type, data=data
+        )
diff --git a/app/tests/notifications/test_notifications.py b/app/tests/notifications/test_notifications.py
new file mode 100644
index 0000000..bd04a3c
--- /dev/null
+++ b/app/tests/notifications/test_notifications.py
@@ -0,0 +1,167 @@
+"""End-to-end tests for the user-facing /notifications endpoints.
+
+Covers listing/ordering, the unread counter, pagination and the unread-only
+filter, per-user isolation, and the read-marking flows with their ownership
+guards.
+"""
+
+import uuid
+
+import pytest
+
+from app.core.messages.error_message import ErrorMessages
+from app.core.messages.success_message import SuccessMessages
+from app.schemas.notification import NotificationType
+from app.tests.notifications.conftest import (
+    ClientFactory,
+    get_user_id,
+    make_user_client,
+    seed_notification,
+)
+
+
+@pytest.mark.asyncio
+async def test_list_requires_auth(client_factory: ClientFactory):
+    """The notification endpoints reject unauthenticated callers."""
+    anonymous = await client_factory()
+
+    assert (await anonymous.get("/notifications")).status_code == 401
+    assert (await anonymous.get("/notifications/unread-count")).status_code == 401
+    assert (await anonymous.post("/notifications/read-all")).status_code == 401
+
+
+@pytest.mark.asyncio
+async def test_list_empty(client_factory: ClientFactory):
+    """A fresh user has an empty inbox and a zero badge."""
+    user = await make_user_client(client_factory, "u1@test.com")
+
+    response = await user.get("/notifications")
+    assert response.status_code == 200
+    assert response.json() == {"data": [], "total": 0, "skip": 0, "limit": 50}
+
+    response = await user.get("/notifications/unread-count")
+    assert response.status_code == 200
+    assert response.json() == {"unread_count": 0}
+
+
+@pytest.mark.asyncio
+async def test_list_newest_first_and_unread_count(client_factory: ClientFactory):
+    """Listing returns seeded notifications newest first; all count as unread."""
+    user = await make_user_client(client_factory, "u1@test.com")
+    user_id = await get_user_id(user)
+    await seed_notification(user_id, data={"subject": "first"})
+    await seed_notification(user_id, data={"subject": "second"})
+
+    response = await user.get("/notifications")
+
+    assert response.status_code == 200
+    body = response.json()
+    assert body["total"] == 2
+    assert [row["data"]["subject"] for row in body["data"]] == ["second", "first"]
+    assert all(row["read_at"] is None for row in body["data"])
+
+    response = await user.get("/notifications/unread-count")
+    assert response.json() == {"unread_count": 2}
+
+
+@pytest.mark.asyncio
+async def test_list_pagination_and_unread_only(client_factory: ClientFactory):
+    """skip/limit page through the inbox; unread_only hides read entries."""
+    user = await make_user_client(client_factory, "u1@test.com")
+    user_id = await get_user_id(user)
+    seeded = [await seed_notification(user_id) for _ in range(3)]
+
+    response = await user.get("/notifications?skip=0&limit=2")
+    body = response.json()
+    assert body["total"] == 3
+    assert len(body["data"]) == 2
+    assert body["limit"] == 2
+
+    await user.post(f"/notifications/{seeded[0].id}/read")
+
+    response = await user.get("/notifications?unread_only=true")
+    body = response.json()
+    assert body["total"] == 2
+    assert all(row["read_at"] is None for row in body["data"])
+
+
+@pytest.mark.asyncio
+async def test_list_only_my_notifications(client_factory: ClientFactory):
+    """Users only ever see their own inbox."""
+    user = await make_user_client(client_factory, "u1@test.com")
+    other = await make_user_client(client_factory, "u2@test.com")
+    await seed_notification(await get_user_id(user), data={"subject": "mine"})
+    await seed_notification(await get_user_id(other), data={"subject": "theirs"})
+
+    body = (await user.get("/notifications")).json()
+
+    assert body["total"] == 1
+    assert body["data"][0]["data"]["subject"] == "mine"
+
+
+@pytest.mark.asyncio
+async def test_mark_read_is_idempotent(client_factory: ClientFactory):
+    """Marking read stamps read_at once; a repeat keeps the original stamp."""
+    user = await make_user_client(client_factory, "u1@test.com")
+    seeded = await seed_notification(await get_user_id(user))
+
+    response = await user.post(f"/notifications/{seeded.id}/read")
+    assert response.status_code == 200
+    body = response.json()
+    assert body["message"] == SuccessMessages.NOTIFICATION_READ
+    first_read_at = body["data"]["read_at"]
+    assert first_read_at is not None
+
+    response = await user.post(f"/notifications/{seeded.id}/read")
+    assert response.status_code == 200
+    assert response.json()["data"]["read_at"] == first_read_at
+
+    response = await user.get("/notifications/unread-count")
+    assert response.json() == {"unread_count": 0}
+
+
+@pytest.mark.asyncio
+async def test_mark_read_not_owned_rejected(client_factory: ClientFactory):
+    """A user cannot mark someone else's notification as read (IDOR guard)."""
+    owner = await make_user_client(client_factory, "owner@test.com")
+    other = await make_user_client(client_factory, "other@test.com")
+    seeded = await seed_notification(await get_user_id(owner))
+
+    response = await other.post(f"/notifications/{seeded.id}/read")
+
+    assert response.status_code == 403
+    assert response.json()["error"] == ErrorMessages.NOTIFICATION_ACCESS_DENIED
+
+
+@pytest.mark.asyncio
+async def test_mark_read_missing_rejected(client_factory: ClientFactory):
+    """Marking a non-existent notification returns 404."""
+    user = await make_user_client(client_factory, "u1@test.com")
+
+    response = await user.post(f"/notifications/{uuid.uuid4()}/read")
+
+    assert response.status_code == 404
+    assert response.json()["error"] == ErrorMessages.NOTIFICATION_NOT_FOUND
+
+
+@pytest.mark.asyncio
+async def test_mark_all_read(client_factory: ClientFactory):
+    """read-all stamps every unread entry and reports how many it touched."""
+    user = await make_user_client(client_factory, "u1@test.com")
+    user_id = await get_user_id(user)
+    seeded = [
+        await seed_notification(
+            user_id, notification_type=NotificationType.ADMIN_PERMISSIONS_CHANGED
+        )
+        for _ in range(3)
+    ]
+    await user.post(f"/notifications/{seeded[0].id}/read")
+
+    response = await user.post("/notifications/read-all")
+
+    assert response.status_code == 200
+    body = response.json()
+    assert body["message"] == SuccessMessages.NOTIFICATIONS_ALL_READ
+    assert body["updated"] == 2
+
+    assert (await user.get("/notifications/unread-count")).json() == {"unread_count": 0}
diff --git a/app/tests/notifications/test_notify_hooks.py b/app/tests/notifications/test_notify_hooks.py
new file mode 100644
index 0000000..2bdda29
--- /dev/null
+++ b/app/tests/notifications/test_notify_hooks.py
@@ -0,0 +1,175 @@
+"""Tests for the notify use case and the notification emit hooks.
+
+The use case is exercised directly against the realtime bridge (StubWebSocket
+on the user's feed, mirroring the support realtime tests); the emit hooks are
+driven through the real admin HTTP flows so a notification only ever appears
+when the production code path created it.
+"""
+
+import asyncio
+import uuid
+
+import pytest
+
+from app.core import realtime
+from app.schemas.notification import NotificationEventType, NotificationType
+from app.tests.admin.conftest import (
+    login,
+    promote_to_superadmin,
+    register_and_verify,
+)
+from app.tests.notifications.conftest import (
+    ClientFactory,
+    get_user_id,
+    seed_notification,
+)
+from app.tests.support.conftest import (
+    make_admin_client,
+    make_user_client,
+    open_ticket,
+)
+
+
+class StubWebSocket:
+    """Minimal WebSocket stand-in capturing the frames sent to it."""
+
+    def __init__(self) -> None:
+        """Initialise the stub with an empty captured-frame list."""
+        self.frames: list[str] = []
+
+    async def send_text(self, data: str) -> None:
+        """Record an outbound text frame."""
+        self.frames.append(data)
+
+
+def test_notifications_topic_format():
+    """The notification feed topic is namespaced by the user id."""
+    uid = uuid.uuid4()
+    assert realtime.notifications_topic(uid) == f"notifications:{uid}"
+
+
+@pytest.mark.asyncio
+async def test_notify_persists_and_publishes(client_factory: ClientFactory):
+    """notify() stores the row and pushes it over the user's feed."""
+    user = await make_user_client(client_factory, "u1@test.com")
+    user_id = await get_user_id(user)
+
+    ws = StubWebSocket()
+    topic = realtime.notifications_topic(user_id)
+    await realtime.manager.connect(topic, ws)
+    await realtime.start_realtime()
+    try:
+        # Let the listener finish its psubscribe before publishing.
+        await asyncio.sleep(0.1)
+        await seed_notification(user_id, data={"subject": "Hello"})
+        for _ in range(50):
+            if ws.frames:
+                break
+            await asyncio.sleep(0.02)
+    finally:
+        await realtime.stop_realtime()
+        await realtime.manager.disconnect(topic, ws)
+
+    assert len(ws.frames) == 1
+    assert NotificationEventType.NOTIFICATION_CREATED.value in ws.frames[0]
+
+    body = (await user.get("/notifications")).json()
+    assert body["total"] == 1
+    assert body["data"][0]["data"]["subject"] == "Hello"
+
+
+@pytest.mark.asyncio
+async def test_admin_reply_notifies_owner(client_factory: ClientFactory):
+    """An admin reply lands in the ticket owner's inbox with the ticket ref."""
+    owner = await make_user_client(client_factory, "owner@test.com")
+    admin = await make_admin_client(client_factory, "admin@test.com")
+    ticket = await open_ticket(owner, subject="Billing question")
+
+    response = await admin.post(
+        f"/admin/support/tickets/{ticket['id']}/messages",
+        json={"body": "We are on it"},
+    )
+    assert response.status_code == 201, response.text
+
+    body = (await owner.get("/notifications")).json()
+    assert body["total"] == 1
+    notification = body["data"][0]
+    assert notification["type"] == NotificationType.SUPPORT_TICKET_REPLIED.value
+    assert notification["data"]["ticket_id"] == ticket["id"]
+    assert notification["data"]["subject"] == "Billing question"
+
+
+@pytest.mark.asyncio
+async def test_admin_self_reply_stays_silent(client_factory: ClientFactory):
+    """An admin replying to their own ticket gets no self-notification."""
+    admin = await make_admin_client(client_factory, "admin@test.com")
+    ticket = await open_ticket(admin, subject="My own issue")
+
+    response = await admin.post(
+        f"/admin/support/tickets/{ticket['id']}/messages",
+        json={"body": "Answering myself"},
+    )
+    assert response.status_code == 201, response.text
+
+    assert (await admin.get("/notifications")).json()["total"] == 0
+
+
+@pytest.mark.asyncio
+async def test_status_change_notifies_owner(client_factory: ClientFactory):
+    """A real status change lands in the owner's inbox with the new status."""
+    owner = await make_user_client(client_factory, "owner@test.com")
+    admin = await make_admin_client(client_factory, "admin@test.com")
+    ticket = await open_ticket(owner, subject="Login fails")
+
+    response = await admin.patch(
+        f"/admin/support/tickets/{ticket['id']}",
+        json={"status": "closed"},
+    )
+    assert response.status_code == 200, response.text
+
+    body = (await owner.get("/notifications")).json()
+    assert body["total"] == 1
+    notification = body["data"][0]
+    assert notification["type"] == NotificationType.SUPPORT_TICKET_STATUS_CHANGED.value
+    assert notification["data"]["status"] == "closed"
+    assert notification["data"]["subject"] == "Login fails"
+
+
+@pytest.mark.asyncio
+async def test_priority_only_change_stays_silent(client_factory: ClientFactory):
+    """Priority/assignment shuffles are admin-internal — no owner notification."""
+    owner = await make_user_client(client_factory, "owner@test.com")
+    admin = await make_admin_client(client_factory, "admin@test.com")
+    ticket = await open_ticket(owner)
+
+    response = await admin.patch(
+        f"/admin/support/tickets/{ticket['id']}",
+        json={"priority": "high"},
+    )
+    assert response.status_code == 200, response.text
+
+    assert (await owner.get("/notifications")).json()["total"] == 0
+
+
+@pytest.mark.asyncio
+async def test_permissions_update_notifies_admin(client_factory: ClientFactory):
+    """Changing an admin's grants drops an RBAC notification in their inbox."""
+    target = await make_admin_client(client_factory, "target@test.com")
+    target_id = await get_user_id(target)
+
+    superadmin = await client_factory()
+    await register_and_verify(superadmin, "boss@test.com")
+    await promote_to_superadmin("boss@test.com")
+    await login(superadmin, "boss@test.com")
+
+    response = await superadmin.patch(
+        f"/admin/admins/{target_id}/permissions",
+        json={"permissions": ["users:read"]},
+    )
+    assert response.status_code == 200, response.text
+
+    body = (await target.get("/notifications")).json()
+    assert body["total"] == 1
+    notification = body["data"][0]
+    assert notification["type"] == NotificationType.ADMIN_PERMISSIONS_CHANGED.value
+    assert notification["data"]["action"] == "set_admin_permissions"
diff --git a/app/use_cases/notify.py b/app/use_cases/notify.py
new file mode 100644
index 0000000..d532237
--- /dev/null
+++ b/app/use_cases/notify.py
@@ -0,0 +1,51 @@
+"""Cross-cutting helper to emit a persistent in-app notification.
+
+Lives in ``use_cases/`` (not the notification service) so any domain service —
+support, admin, files — can emit notifications without a service-to-service
+call, mirroring ``log_activity``.
+"""
+
+import uuid
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.realtime import notifications_topic, publish_safe
+from app.models.notification import Notification
+from app.repositories.notification import create_notification
+from app.schemas.common import JsonValue
+from app.schemas.notification import (
+    NotificationEventType,
+    NotificationRead,
+    NotificationRealtimeEvent,
+    NotificationType,
+)
+
+
+async def notify(
+    session: AsyncSession,
+    *,
+    user_id: uuid.UUID,
+    notification_type: NotificationType,
+    data: dict[str, JsonValue] | None = None,
+) -> Notification:
+    """Persist a notification for ``user_id`` and push it over their feed.
+
+    The database row is the source of truth; the realtime publish is
+    best-effort (``publish_safe``), so an offline user simply finds the
+    notification in their inbox on the next fetch.
+    """
+    notification = Notification(
+        user_id=user_id,
+        type=notification_type.value,
+        data=data or {},
+    )
+    notification = await create_notification(session, notification)
+
+    await publish_safe(
+        notifications_topic(user_id),
+        NotificationRealtimeEvent(
+            type=NotificationEventType.NOTIFICATION_CREATED,
+            notification=NotificationRead.model_validate(notification),
+        ),
+    )
+    return notification