diff --git a/app/api/__init__.py b/app/api/__init__.py
index dcee1703..311c572e 100644
--- a/app/api/__init__.py
+++ b/app/api/__init__.py
@@ -20,6 +20,7 @@ from app.api.google_drive import router as google_drive_router
from app.api.imap_accounts import router as imap_accounts_router
from app.api.integrations import router as integrations_router
from app.api.logs import router as logs_router
+from app.api.notifications import router as notifications_router
from app.api.onboarding import router as onboarding_router
from app.api.onedrive import router as onedrive_router
from app.api.openai import router as openai_router
@@ -74,3 +75,4 @@ router.include_router(billing_router)
router.include_router(pipelines_router)
router.include_router(imap_accounts_router)
router.include_router(integrations_router)
+router.include_router(notifications_router)
diff --git a/app/api/notifications.py b/app/api/notifications.py
new file mode 100644
index 00000000..7927b255
--- /dev/null
+++ b/app/api/notifications.py
@@ -0,0 +1,484 @@
+"""API endpoints for per-user notification targets, preferences, and in-app inbox.
+
+Users can define notification targets (email via SMTP, webhook via HTTP POST)
+and configure which document events trigger which targets. In-app notifications
+are always created and surfaced via the bell icon / inbox endpoints.
+"""
+
+import json
+import logging
+from typing import Annotated, Any
+
+from fastapi import APIRouter, Depends, HTTPException, Request, status
+from pydantic import BaseModel, Field
+from sqlalchemy.orm import Session
+
+from app.database import get_db
+from app.models import InAppNotification, UserNotificationPreference, UserNotificationTarget
+from app.utils.user_notification import USER_EVENT_LABELS
+from app.utils.user_scope import get_current_owner_id
+
+logger = logging.getLogger(__name__)
+router = APIRouter(prefix="/user-notifications", tags=["user-notifications"])
+
+DbSession = Annotated[Session, Depends(get_db)]
+
+# ---------------------------------------------------------------------------
+# Auth helper (mirrors api_tokens.py pattern)
+# ---------------------------------------------------------------------------
+
+
+def _get_owner_id(request: Request) -> str:
+ """Return the current user's owner ID, raising 401 if unauthenticated."""
+ owner_id = get_current_owner_id(request)
+ if not owner_id:
+ raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated")
+ return owner_id
+
+
+CurrentOwner = Annotated[str, Depends(_get_owner_id)]
+
+# ---------------------------------------------------------------------------
+# Pydantic schemas
+# ---------------------------------------------------------------------------
+
+VALID_CHANNEL_TYPES = {"email", "webhook"}
+VALID_EVENT_TYPES = set(USER_EVENT_LABELS.keys())
+
+
+class NotificationTargetCreate(BaseModel):
+ """Schema for creating a new notification target."""
+
+ channel_type: str = Field(..., pattern="^(email|webhook)$")
+ name: str = Field(..., min_length=1, max_length=255)
+ config: dict[str, Any] = Field(default_factory=dict)
+ is_active: bool = True
+
+
+class NotificationTargetUpdate(BaseModel):
+ """Schema for updating an existing notification target."""
+
+ name: str | None = Field(None, min_length=1, max_length=255)
+ config: dict[str, Any] | None = None
+ is_active: bool | None = None
+
+
+class PreferenceItem(BaseModel):
+ """A single preference toggle for one event+channel combination."""
+
+ is_enabled: bool
+ target_id: int | None = None
+
+
+class PreferenceItemFull(BaseModel):
+ """Full preference item including event and channel type (used in bulk update)."""
+
+ event_type: str
+ channel_type: str
+ is_enabled: bool
+ target_id: int | None = None
+
+
+class PreferencesUpdate(BaseModel):
+ """Bulk preferences update payload — a flat list of preference items."""
+
+ preferences: list[PreferenceItemFull]
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _mask_email_config(config: dict[str, Any]) -> dict[str, Any]:
+ """Return a copy of an email config dict with the password masked."""
+ masked = dict(config)
+ if masked.get("smtp_password"):
+ masked["smtp_password"] = "****"
+ return masked
+
+
+def _target_to_dict(target: UserNotificationTarget) -> dict[str, Any]:
+ """Serialize a UserNotificationTarget to a response dict, masking secrets."""
+ config: dict[str, Any] = {}
+ if target.config:
+ try:
+ config = json.loads(target.config)
+ except (json.JSONDecodeError, ValueError):
+ config = {}
+
+ if target.channel_type == "email":
+ config = _mask_email_config(config)
+
+ return {
+ "id": target.id,
+ "channel_type": target.channel_type,
+ "name": target.name,
+ "config": config,
+ "is_active": target.is_active,
+ "created_at": target.created_at,
+ "updated_at": target.updated_at,
+ }
+
+
+# ---------------------------------------------------------------------------
+# Inbox endpoints
+# ---------------------------------------------------------------------------
+
+
+@router.get("/inbox")
+async def list_inbox(
+ owner_id: CurrentOwner,
+ db: DbSession,
+ skip: int = 0,
+ limit: int = 50,
+) -> list[dict[str, Any]]:
+ """List in-app notifications for the authenticated user, newest first."""
+ notifications = (
+ db.query(InAppNotification)
+ .filter(InAppNotification.owner_id == owner_id)
+ .order_by(InAppNotification.created_at.desc())
+ .offset(skip)
+ .limit(limit)
+ .all()
+ )
+ return [
+ {
+ "id": n.id,
+ "event_type": n.event_type,
+ "title": n.title,
+ "message": n.message,
+ "is_read": n.is_read,
+ "file_id": n.file_id,
+ "created_at": n.created_at,
+ }
+ for n in notifications
+ ]
+
+
+@router.get("/inbox/unread-count")
+async def unread_count(
+ owner_id: CurrentOwner,
+ db: DbSession,
+) -> dict[str, int]:
+ """Return the number of unread in-app notifications."""
+ count = (
+ db.query(InAppNotification)
+ .filter(InAppNotification.owner_id == owner_id, InAppNotification.is_read == False) # noqa: E712
+ .count()
+ )
+ return {"count": count}
+
+
+@router.post("/inbox/{notification_id}/read", status_code=status.HTTP_200_OK)
+async def mark_read(
+ notification_id: int,
+ owner_id: CurrentOwner,
+ db: DbSession,
+) -> dict[str, str]:
+ """Mark a single in-app notification as read."""
+ notif = (
+ db.query(InAppNotification)
+ .filter(InAppNotification.id == notification_id, InAppNotification.owner_id == owner_id)
+ .first()
+ )
+ if not notif:
+ raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Notification not found")
+ try:
+ notif.is_read = True
+ db.commit()
+ except Exception:
+ db.rollback()
+ raise
+ return {"detail": "Marked as read"}
+
+
+@router.post("/inbox/read-all", status_code=status.HTTP_200_OK)
+async def mark_all_read(
+ owner_id: CurrentOwner,
+ db: DbSession,
+) -> dict[str, str]:
+ """Mark all in-app notifications as read for the authenticated user."""
+ try:
+ db.query(InAppNotification).filter(
+ InAppNotification.owner_id == owner_id,
+ InAppNotification.is_read == False, # noqa: E712
+ ).update({"is_read": True})
+ db.commit()
+ except Exception:
+ db.rollback()
+ raise
+ return {"detail": "All notifications marked as read"}
+
+
+# ---------------------------------------------------------------------------
+# Notification target endpoints
+# ---------------------------------------------------------------------------
+
+
+@router.get("/targets")
+async def list_targets(
+ owner_id: CurrentOwner,
+ db: DbSession,
+) -> list[dict[str, Any]]:
+ """List all notification targets for the authenticated user."""
+ targets = (
+ db.query(UserNotificationTarget)
+ .filter(UserNotificationTarget.owner_id == owner_id)
+ .order_by(UserNotificationTarget.created_at.desc())
+ .all()
+ )
+ return [_target_to_dict(t) for t in targets]
+
+
+@router.post("/targets", status_code=status.HTTP_201_CREATED)
+async def create_target(
+ body: NotificationTargetCreate,
+ owner_id: CurrentOwner,
+ db: DbSession,
+) -> dict[str, Any]:
+ """Create a new notification target (email or webhook)."""
+ target = UserNotificationTarget(
+ owner_id=owner_id,
+ channel_type=body.channel_type,
+ name=body.name,
+ config=json.dumps(body.config),
+ is_active=body.is_active,
+ )
+ try:
+ db.add(target)
+ db.commit()
+ db.refresh(target)
+ except Exception:
+ db.rollback()
+ raise
+
+ logger.info("Notification target created: id=%s owner=%s type=%s", target.id, owner_id, body.channel_type)
+ return _target_to_dict(target)
+
+
+@router.put("/targets/{target_id}", status_code=status.HTTP_200_OK)
+async def update_target(
+ target_id: int,
+ body: NotificationTargetUpdate,
+ owner_id: CurrentOwner,
+ db: DbSession,
+) -> dict[str, Any]:
+ """Update an existing notification target."""
+ target = (
+ db.query(UserNotificationTarget)
+ .filter(UserNotificationTarget.id == target_id, UserNotificationTarget.owner_id == owner_id)
+ .first()
+ )
+ if not target:
+ raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Target not found")
+
+ try:
+ if body.name is not None:
+ target.name = body.name
+ if body.config is not None:
+ # Merge new config over existing, preserving masked password field if unchanged
+ existing_config: dict[str, Any] = {}
+ if target.config:
+ try:
+ existing_config = json.loads(target.config)
+ except (json.JSONDecodeError, ValueError):
+ existing_config = {}
+ merged = dict(existing_config)
+ for k, v in body.config.items():
+ # Skip writing back a masked password placeholder
+ if k == "smtp_password" and v == "****":
+ continue
+ merged[k] = v
+ target.config = json.dumps(merged)
+ if body.is_active is not None:
+ target.is_active = body.is_active
+ db.commit()
+ db.refresh(target)
+ except Exception:
+ db.rollback()
+ raise
+
+ logger.info("Notification target updated: id=%s owner=%s", target_id, owner_id)
+ return _target_to_dict(target)
+
+
+@router.delete("/targets/{target_id}", status_code=status.HTTP_200_OK)
+async def delete_target(
+ target_id: int,
+ owner_id: CurrentOwner,
+ db: DbSession,
+) -> dict[str, str]:
+ """Delete a notification target and its associated preferences."""
+ target = (
+ db.query(UserNotificationTarget)
+ .filter(UserNotificationTarget.id == target_id, UserNotificationTarget.owner_id == owner_id)
+ .first()
+ )
+ if not target:
+ raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Target not found")
+
+ try:
+ # Remove any preferences that reference this target
+ db.query(UserNotificationPreference).filter(
+ UserNotificationPreference.owner_id == owner_id,
+ UserNotificationPreference.target_id == target_id,
+ ).delete()
+ db.delete(target)
+ db.commit()
+ except Exception:
+ db.rollback()
+ raise
+
+ logger.info("Notification target deleted: id=%s owner=%s", target_id, owner_id)
+ return {"detail": "Target deleted"}
+
+
+@router.post("/targets/{target_id}/test", status_code=status.HTTP_200_OK)
+async def test_target(
+ target_id: int,
+ owner_id: CurrentOwner,
+ db: DbSession,
+) -> dict[str, str]:
+ """Send a test notification to the specified target."""
+ target = (
+ db.query(UserNotificationTarget)
+ .filter(UserNotificationTarget.id == target_id, UserNotificationTarget.owner_id == owner_id)
+ .first()
+ )
+ if not target:
+ raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Target not found")
+
+ config: dict[str, Any] = {}
+ if target.config:
+ try:
+ config = json.loads(target.config)
+ except (json.JSONDecodeError, ValueError):
+ config = {}
+
+ title = "DocuElevate Test Notification"
+ message = f"This is a test notification from DocuElevate for target '{target.name}'."
+
+ if target.channel_type == "email":
+ from app.utils.user_notification import _send_email_notification
+
+ ok = _send_email_notification(config, title, message)
+ elif target.channel_type == "webhook":
+ from app.utils.user_notification import _send_webhook_notification
+
+ ok = _send_webhook_notification(config, "test", title, message)
+ else:
+ raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Unknown channel type")
+
+ if not ok:
+ raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="Failed to send test notification")
+
+ return {"detail": "Test notification sent"}
+
+
+# ---------------------------------------------------------------------------
+# Preferences endpoints
+# ---------------------------------------------------------------------------
+
+
+@router.get("/preferences")
+async def get_preferences(
+ owner_id: CurrentOwner,
+ db: DbSession,
+) -> dict[str, Any]:
+ """Return all notification preferences for the authenticated user.
+
+ Response structure:
+ {
+ "event_types": ["document.processed", "document.failed"],
+ "event_labels": {"document.processed": "Document Processed", ...},
+ "preferences": {
+ "document.processed": {
+ "in_app": {"is_enabled": true, "target_id": null},
+ "email": {"is_enabled": false, "target_id": 1},
+ ...
+ }
+ }
+ }
+ """
+ prefs = db.query(UserNotificationPreference).filter(UserNotificationPreference.owner_id == owner_id).all()
+
+ # Build nested dict: event_type -> channel_type -> {is_enabled, target_id}
+ result: dict[str, dict[str, dict[str, Any]]] = {}
+ for pref in prefs:
+ result.setdefault(pref.event_type, {})[pref.channel_type] = {
+ "is_enabled": pref.is_enabled,
+ "target_id": pref.target_id,
+ }
+
+ return {
+ "event_types": list(USER_EVENT_LABELS.keys()),
+ "event_labels": USER_EVENT_LABELS,
+ "preferences": result,
+ }
+
+
+@router.put("/preferences", status_code=status.HTTP_200_OK)
+async def update_preferences(
+ body: PreferencesUpdate,
+ owner_id: CurrentOwner,
+ db: DbSession,
+) -> dict[str, str]:
+ """Bulk upsert notification preferences for the authenticated user.
+
+ Validates that any referenced target_id belongs to the requesting user.
+ """
+ # Collect all target IDs referenced in the payload for ownership validation
+ referenced_target_ids: set[int] = set()
+ for item in body.preferences:
+ if item.target_id is not None:
+ referenced_target_ids.add(item.target_id)
+
+ if referenced_target_ids:
+ owned_ids = {
+ row.id
+ for row in db.query(UserNotificationTarget.id)
+ .filter(
+ UserNotificationTarget.owner_id == owner_id,
+ UserNotificationTarget.id.in_(referenced_target_ids),
+ )
+ .all()
+ }
+ invalid = referenced_target_ids - owned_ids
+ if invalid:
+ raise HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail=f"Invalid or inaccessible target_id(s): {sorted(invalid)}",
+ )
+
+ try:
+ for item in body.preferences:
+ existing = (
+ db.query(UserNotificationPreference)
+ .filter(
+ UserNotificationPreference.owner_id == owner_id,
+ UserNotificationPreference.event_type == item.event_type,
+ UserNotificationPreference.channel_type == item.channel_type,
+ UserNotificationPreference.target_id == item.target_id,
+ )
+ .first()
+ )
+ if existing:
+ existing.is_enabled = item.is_enabled
+ else:
+ db.add(
+ UserNotificationPreference(
+ owner_id=owner_id,
+ event_type=item.event_type,
+ channel_type=item.channel_type,
+ target_id=item.target_id,
+ is_enabled=item.is_enabled,
+ )
+ )
+ db.commit()
+ except Exception:
+ db.rollback()
+ raise
+
+ logger.info("Notification preferences updated for owner=%s", owner_id)
+ return {"detail": "Preferences updated"}
diff --git a/app/models.py b/app/models.py
index 687bf3da..be05e2b5 100644
--- a/app/models.py
+++ b/app/models.py
@@ -637,3 +637,50 @@ class ApiToken(Base):
is_active = Column(Boolean, nullable=False, default=True)
created_at = Column(DateTime(timezone=True), server_default=func.now())
revoked_at = Column(DateTime(timezone=True), nullable=True)
+
+
+class UserNotificationTarget(Base):
+ """Per-user notification target (email or webhook channel)."""
+
+ __tablename__ = "user_notification_targets"
+
+ id = Column(Integer, primary_key=True, index=True)
+ owner_id = Column(String, nullable=False, index=True)
+ channel_type = Column(String(20), nullable=False) # "email" or "webhook"
+ name = Column(String(255), nullable=False) # Human-readable label
+ config = Column(Text, nullable=True) # JSON: smtp config or webhook url
+ is_active = Column(Boolean, nullable=False, default=True)
+ created_at = Column(DateTime(timezone=True), server_default=func.now())
+ updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
+
+
+class UserNotificationPreference(Base):
+ """Mapping: which user events trigger which notification channel."""
+
+ __tablename__ = "user_notification_preferences"
+
+ id = Column(Integer, primary_key=True, index=True)
+ owner_id = Column(String, nullable=False, index=True)
+ event_type = Column(String(50), nullable=False) # "document.processed", "document.failed"
+ channel_type = Column(String(20), nullable=False) # "in_app", "email", "webhook"
+ target_id = Column(Integer, nullable=True) # NULL = in_app, else UserNotificationTarget.id
+ is_enabled = Column(Boolean, nullable=False, default=True)
+
+ __table_args__ = (UniqueConstraint("owner_id", "event_type", "channel_type", "target_id"),)
+ created_at = Column(DateTime(timezone=True), server_default=func.now())
+ updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
+
+
+class InAppNotification(Base):
+ """In-app notification record for the bell icon / inbox."""
+
+ __tablename__ = "in_app_notifications"
+
+ id = Column(Integer, primary_key=True, index=True)
+ owner_id = Column(String, nullable=False, index=True)
+ event_type = Column(String(50), nullable=False) # "document.processed", "document.failed"
+ title = Column(String(255), nullable=False)
+ message = Column(Text, nullable=True)
+ is_read = Column(Boolean, nullable=False, default=False, index=True)
+ file_id = Column(Integer, nullable=True) # Optional link to FileRecord
+ created_at = Column(DateTime(timezone=True), server_default=func.now(), index=True)
diff --git a/app/utils/user_notification.py b/app/utils/user_notification.py
new file mode 100644
index 00000000..f6a277cd
--- /dev/null
+++ b/app/utils/user_notification.py
@@ -0,0 +1,233 @@
+"""Per-user notification dispatch service.
+
+Handles user-centric events (document.processed, document.failed) by:
+1. Always creating an InAppNotification record
+2. Sending via configured email/webhook targets (UserNotificationTarget)
+ if the user has enabled that channel/event combination.
+"""
+
+import json
+import logging
+import smtplib
+from email.mime.multipart import MIMEMultipart
+from email.mime.text import MIMEText
+from typing import Any
+
+import httpx
+
+from app.database import SessionLocal
+from app.models import InAppNotification, UserNotificationPreference, UserNotificationTarget
+
+logger = logging.getLogger(__name__)
+
+# Supported user-centric event types
+EVENT_DOCUMENT_PROCESSED = "document.processed"
+EVENT_DOCUMENT_FAILED = "document.failed"
+
+USER_EVENT_LABELS: dict[str, str] = {
+ EVENT_DOCUMENT_PROCESSED: "Document Processed",
+ EVENT_DOCUMENT_FAILED: "Document Processing Failed",
+}
+
+
+def create_in_app_notification(
+ owner_id: str,
+ event_type: str,
+ title: str,
+ message: str,
+ file_id: int | None = None,
+) -> InAppNotification | None:
+ """Persist an InAppNotification record for the given user.
+
+ Returns:
+ The created InAppNotification, or None on error.
+ """
+ db = SessionLocal()
+ try:
+ notif = InAppNotification(
+ owner_id=owner_id,
+ event_type=event_type,
+ title=title,
+ message=message,
+ file_id=file_id,
+ )
+ db.add(notif)
+ db.commit()
+ db.refresh(notif)
+ return notif
+ except Exception:
+ db.rollback()
+ logger.exception("Failed to create in-app notification for owner_id=%s", owner_id)
+ return None
+ finally:
+ db.close()
+
+
+def _send_email_notification(target_config: dict[str, Any], title: str, message: str) -> bool:
+ """Send an email notification via the configured SMTP target.
+
+ Args:
+ target_config: dict with keys: smtp_host, smtp_port, smtp_username,
+ smtp_password, smtp_use_tls, recipient_email
+ title: Email subject
+ message: Email body text
+
+ Returns:
+ True if the email was sent successfully, False otherwise.
+ """
+ try:
+ smtp_host = target_config.get("smtp_host", "")
+ smtp_port = int(target_config.get("smtp_port", 587))
+ smtp_username = target_config.get("smtp_username", "")
+ smtp_password = target_config.get("smtp_password", "")
+ smtp_use_tls = bool(target_config.get("smtp_use_tls", True))
+ recipient_email = target_config.get("recipient_email", "")
+ sender_email = target_config.get("sender_email") or smtp_username or "noreply@docuelevate.local"
+
+ if not smtp_host or not recipient_email:
+ logger.warning("Email notification target missing smtp_host or recipient_email")
+ return False
+
+ msg = MIMEMultipart("alternative")
+ msg["Subject"] = title
+ msg["From"] = sender_email
+ msg["To"] = recipient_email
+ msg.attach(MIMEText(message, "plain"))
+
+ with smtplib.SMTP(smtp_host, smtp_port, timeout=30) as server:
+ if smtp_use_tls:
+ server.starttls()
+ if smtp_username and smtp_password:
+ server.login(smtp_username, smtp_password)
+ server.send_message(msg)
+
+ logger.info("Email notification sent to %s", recipient_email)
+ return True
+ except Exception:
+ logger.exception("Failed to send email notification")
+ return False
+
+
+def _send_webhook_notification(target_config: dict[str, Any], event_type: str, title: str, message: str) -> bool:
+ """Send a webhook POST notification to the configured URL.
+
+ Args:
+ target_config: dict with keys: url, secret (optional HMAC header value)
+ event_type: The event type string
+ title: Notification title
+ message: Notification body
+
+ Returns:
+ True if the webhook was delivered successfully, False otherwise.
+ """
+ try:
+ url = target_config.get("url", "")
+ secret = target_config.get("secret", "")
+
+ if not url:
+ logger.warning("Webhook notification target missing url")
+ return False
+
+ payload = {
+ "event": event_type,
+ "title": title,
+ "message": message,
+ }
+ headers = {"Content-Type": "application/json"}
+ if secret:
+ headers["X-DocuElevate-Secret"] = secret
+
+ response = httpx.post(url, json=payload, headers=headers, timeout=10)
+ response.raise_for_status()
+ logger.info("Webhook notification sent to %s (status %s)", url, response.status_code)
+ return True
+ except Exception:
+ logger.exception("Failed to send webhook notification to %s", target_config.get("url", ""))
+ return False
+
+
+def dispatch_user_notification(
+ owner_id: str,
+ event_type: str,
+ title: str,
+ message: str,
+ file_id: int | None = None,
+) -> None:
+ """Dispatch a user notification for the given event.
+
+ Always creates an in-app notification. Also sends via email/webhook
+ targets if the user has configured and enabled them for this event.
+
+ Args:
+ owner_id: The user's stable identifier.
+ event_type: e.g. "document.processed" or "document.failed"
+ title: Short notification title.
+ message: Longer notification body.
+ file_id: Optional FileRecord.id to link.
+ """
+ # 1. Always create an in-app notification
+ create_in_app_notification(
+ owner_id=owner_id,
+ event_type=event_type,
+ title=title,
+ message=message,
+ file_id=file_id,
+ )
+
+ # 2. Check for configured email/webhook preferences
+ db = SessionLocal()
+ try:
+ prefs = (
+ db.query(UserNotificationPreference)
+ .filter(
+ UserNotificationPreference.owner_id == owner_id,
+ UserNotificationPreference.event_type == event_type,
+ UserNotificationPreference.is_enabled == True, # noqa: E712
+ UserNotificationPreference.channel_type.in_(["email", "webhook"]),
+ )
+ .all()
+ )
+
+ for pref in prefs:
+ if not pref.target_id:
+ continue
+ target = db.get(UserNotificationTarget, pref.target_id)
+ if not target or not target.is_active:
+ continue
+ config: dict[str, Any] = {}
+ if target.config:
+ try:
+ config = json.loads(target.config)
+ except (json.JSONDecodeError, ValueError):
+ config = {}
+
+ if pref.channel_type == "email":
+ _send_email_notification(config, title, message)
+ elif pref.channel_type == "webhook":
+ _send_webhook_notification(config, event_type, title, message)
+ except Exception:
+ logger.exception("Error dispatching user notification for owner_id=%s event=%s", owner_id, event_type)
+ finally:
+ db.close()
+
+
+def notify_user_document_processed(owner_id: str, filename: str, file_id: int | None = None) -> None:
+ """Notify a user that their document was successfully processed."""
+ dispatch_user_notification(
+ owner_id=owner_id,
+ event_type=EVENT_DOCUMENT_PROCESSED,
+ title=f"Document processed: {filename}",
+ message=f"Your document '{filename}' has been successfully processed and uploaded.",
+ file_id=file_id,
+ )
+
+
+def notify_user_document_failed(owner_id: str, filename: str, error: str, file_id: int | None = None) -> None:
+ """Notify a user that their document processing failed."""
+ dispatch_user_notification(
+ owner_id=owner_id,
+ event_type=EVENT_DOCUMENT_FAILED,
+ title=f"Document processing failed: {filename}",
+ message=f"Processing of '{filename}' failed: {error}",
+ file_id=file_id,
+ )
diff --git a/app/views/__init__.py b/app/views/__init__.py
index b8fb9b0d..6e50a63e 100644
--- a/app/views/__init__.py
+++ b/app/views/__init__.py
@@ -18,6 +18,7 @@ from app.views.help import router as help_router # Built-in help / How-To docs
from app.views.imap_accounts import router as imap_accounts_router
from app.views.integrations import router as integrations_router # Unified integrations dashboard
from app.views.license_routes import router as license_router # Add the license router
+from app.views.notifications import router as notifications_router
from app.views.onboarding import router as onboarding_router
from app.views.onedrive import router as onedrive_router
from app.views.pipelines import router as pipelines_router # Processing pipelines
@@ -52,4 +53,5 @@ router.include_router(onboarding_router) # User onboarding wizard
router.include_router(pipelines_router) # Processing pipelines
router.include_router(imap_accounts_router) # Per-user IMAP ingestion accounts
router.include_router(integrations_router) # Unified integrations dashboard
+router.include_router(notifications_router) # User notification dashboard
router.include_router(help_router) # Built-in help / How-To docs
diff --git a/app/views/notifications.py b/app/views/notifications.py
new file mode 100644
index 00000000..7e4e26a3
--- /dev/null
+++ b/app/views/notifications.py
@@ -0,0 +1,20 @@
+"""View route for the notifications dashboard."""
+
+import logging
+
+from fastapi import Request
+
+from app.views.base import APIRouter, require_login, templates
+
+logger = logging.getLogger(__name__)
+router = APIRouter()
+
+
+@router.get("/notifications")
+@require_login
+async def notifications_dashboard(request: Request):
+ """Render the notifications dashboard."""
+ return templates.TemplateResponse(
+ "notifications_dashboard.html",
+ {"request": request, "page_title": "Notifications"},
+ )
diff --git a/docs/ConfigurationGuide.md b/docs/ConfigurationGuide.md
index 6df53ac3..424fa173 100644
--- a/docs/ConfigurationGuide.md
+++ b/docs/ConfigurationGuide.md
@@ -1070,6 +1070,55 @@ payment processors.
For detailed setup instructions, see the [Notifications Setup Guide](NotificationsSetup.md).
+#### Per-User Notification System
+
+In addition to the system-level Apprise notifications, DocuElevate includes a **per-user notification system** that gives each user full control over how they are notified about their own document events.
+
+**Notification Dashboard** — available at `/notifications` for every logged-in user. It has three tabs:
+
+| Tab | Description |
+|-----|-------------|
+| **Inbox** | In-app bell-icon notification feed. Persisted in the database; shows unread count badge in the navigation bar. Users can mark individual items or all items as read. |
+| **Targets** | User-defined notification channels: **Email (SMTP)** and **Webhook (HTTP POST)**. Each target can be tested independently from the UI. |
+| **Preferences** | Event/channel matrix. Users choose which channels are triggered for each event type. In-app notifications are always enabled. |
+
+**User-centric event types:**
+
+| Event | Description |
+|-------|-------------|
+| `document.processed` | A document uploaded by the user was successfully processed and uploaded to destinations |
+| `document.failed` | A document uploaded by the user failed during processing |
+
+**Email target configuration fields:**
+
+| Field | Description |
+|-------|-------------|
+| `smtp_host` | SMTP server hostname |
+| `smtp_port` | SMTP port (default `587`) |
+| `smtp_username` | SMTP login username |
+| `smtp_password` | SMTP login password (stored in database, masked in UI) |
+| `smtp_use_tls` | Enable STARTTLS (`true`/`false`, default `true`) |
+| `sender_email` | From address (defaults to `smtp_username` if omitted) |
+| `recipient_email` | Destination address for this target |
+
+**Webhook target configuration fields:**
+
+| Field | Description |
+|-------|-------------|
+| `url` | HTTP(S) URL to POST the notification payload to |
+| `secret` | Optional secret string sent as `X-DocuElevate-Secret` header |
+
+**Webhook payload format:**
+```json
+{
+ "event": "document.processed",
+ "title": "Document processed: invoice.pdf",
+ "message": "Your document 'invoice.pdf' has been successfully processed and uploaded."
+}
+```
+
+> **Note:** There are no additional environment variables for the per-user notification system — all settings are stored in the database and managed through the user-facing `/notifications` dashboard.
+
### Webhooks
Webhooks notify external systems via HTTP POST when document events occur.
diff --git a/frontend/templates/base.html b/frontend/templates/base.html
index 4ef16130..c1fd21dc 100644
--- a/frontend/templates/base.html
+++ b/frontend/templates/base.html
@@ -194,6 +194,19 @@
Help
+
+
+
+
+
+
diff --git a/frontend/templates/notifications_dashboard.html b/frontend/templates/notifications_dashboard.html
new file mode 100644
index 00000000..5bfcc947
--- /dev/null
+++ b/frontend/templates/notifications_dashboard.html
@@ -0,0 +1,979 @@
+{% extends "base.html" %}
+{% block title %}Notifications – DocuElevate{% endblock %}
+
+{% block content %}
+