Notifications backend

pydatalab/src/pydatalab/config.py

@@ -237,6 +237,11 @@ class ServerConfig(BaseSettings):
         description="Whether to disable magic-link email authentication while retaining SMTP-backed notification emails.",
     )
 
+    ENABLE_NOTIFICATIONS: bool = Field(
+        False,
+        description="Whether to enable in-app notifications and their API endpoints.",
+    )
+
     MAX_CONTENT_LENGTH: int = Field(
         10 * 1000**3,
         description=r"""Direct mapping to the equivalent Flask setting. In practice, limits the file size that can be uploaded.

pydatalab/src/pydatalab/feature_flags.py

@@ -7,7 +7,7 @@
 from pydatalab.config import CONFIG
 from pydatalab.logger import LOGGER
 
-__all__ = ("FEATURE_FLAGS", "check_feature_flags", "FeatureFlags")
+__all__ = ("FEATURE_FLAGS", "check_feature_flags", "FeatureFlags", "NotificationFeatures")
 
 
 class AuthMechanisms(BaseModel):
@@ -23,10 +23,15 @@ class AIIntegrations(BaseModel):
     anthropic: bool = False
 
 
+class NotificationFeatures(BaseModel):
+    enabled: bool = False
+
+
 class FeatureFlags(BaseModel):
     auth_mechanisms: AuthMechanisms = AuthMechanisms()
     ai_integrations: AIIntegrations = AIIntegrations()
     email_notifications: bool = False
+    notifications: NotificationFeatures = NotificationFeatures()
 
 
 FEATURE_FLAGS: FeatureFlags = FeatureFlags()
@@ -59,6 +64,8 @@ def check_feature_flags(app):
 
     """
 
+    FEATURE_FLAGS.notifications = NotificationFeatures(enabled=CONFIG.ENABLE_NOTIFICATIONS)
+
     if CONFIG.EMAIL_AUTH_SMTP_SETTINGS is None:
         LOGGER.warning(
             "No email auth SMTP settings provided, email registration will not be enabled."

pydatalab/src/pydatalab/main.py

@@ -226,6 +226,10 @@ def register_endpoints(app: Flask):
     versions = ["", f"v{major}", f"v{major}.{minor}", f"v{major}.{minor}.{patch}"]
 
     for bp in BLUEPRINTS:
+        if bp.name == "notifications" and not CONFIG.ENABLE_NOTIFICATIONS:
+            LOGGER.info("Skipping notification routes because notifications are disabled.")
+            continue
+
         for ver in versions:
             app.register_blueprint(
                 bp, url_prefix=f"{CONFIG.ROOT_PATH}{ver}", name=f"{ver}/{bp.name}"

pydatalab/src/pydatalab/models/__init__.py

@@ -4,6 +4,11 @@
 from pydatalab.models.collections import Collection
 from pydatalab.models.equipment import Equipment
 from pydatalab.models.files import File
+from pydatalab.models.notifications import (
+    Notification,
+    NotificationGrouping,
+    NotificationOccurrence,
+)
 from pydatalab.models.people import Person
 from pydatalab.models.samples import Sample
 from pydatalab.models.starting_materials import StartingMaterial
@@ -25,5 +30,8 @@
     "Collection",
     "Equipment",
     "ItemVersion",
+    "Notification",
+    "NotificationGrouping",
+    "NotificationOccurrence",
     "ITEM_MODELS",
 )

pydatalab/src/pydatalab/models/notifications.py

@@ -0,0 +1,116 @@
+from datetime import datetime, timezone
+from enum import Enum
+
+from pydantic import BaseModel, Field, root_validator
+
+from pydatalab.models.entries import Entry
+from pydatalab.models.utils import JSON_ENCODERS, PyObjectId
+
+
+class NotificationLevel(str, Enum):
+    LOW = "low"
+    NORMAL = "normal"
+    IMPORTANT = "important"
+    URGENT = "urgent"
+    CRITICAL = "critical"
+
+    @property
+    def priority(self) -> int:
+        return {
+            NotificationLevel.LOW: 10,
+            NotificationLevel.NORMAL: 20,
+            NotificationLevel.IMPORTANT: 30,
+            NotificationLevel.URGENT: 40,
+            NotificationLevel.CRITICAL: 50,
+        }[self]
+
+
+class NotificationGroupPolicy(str, Enum):
+    ONCE = "once"
+    WINDOW = "window"
+
+
+class NotificationGrouping(BaseModel):
+    """Rules for grouping repeated notification occurrences."""
+
+    key: str = Field(
+        ...,
+        min_length=1,
+        max_length=500,
+        description="Stable key used to group repeated notifications from the same thing.",
+    )
+    policy: NotificationGroupPolicy = Field(
+        NotificationGroupPolicy.WINDOW,
+        description="How repeated notification occurrences should be grouped.",
+    )
+    window_seconds: int | None = Field(
+        86400,
+        ge=1,
+        description=(
+            "Minimum interval before a grouped notification can create a new notification document."
+        ),
+    )
+    max_occurrences: int = Field(
+        100,
+        ge=1,
+        description="Maximum number of occurrences before a new notification document is created.",
+    )
+
+    @root_validator
+    def validate_grouping_policy(cls, values):
+        policy = NotificationGroupPolicy(values.get("policy"))
+        if policy == NotificationGroupPolicy.WINDOW and values.get("window_seconds") is None:
+            raise ValueError("window_seconds must be provided for window grouping.")
+        if policy == NotificationGroupPolicy.ONCE:
+            values["window_seconds"] = None
+        return values
+
+    class Config:
+        json_encoders = JSON_ENCODERS
+        use_enum_values = True
+
+
+class NotificationOccurrence(BaseModel):
+    """A single occurrence represented by a grouped notification."""
+
+    occurred_at: datetime = Field(default_factory=lambda: datetime.now(tz=timezone.utc))
+    summary: str | None = Field(None, max_length=1000)
+    message: str | None = Field(None, max_length=5000)
+    level: NotificationLevel = Field(NotificationLevel.NORMAL)
+    is_new: bool = Field(
+        False,
+        description="Whether this occurrence arrived since the notification was last read.",
+    )
+
+    class Config:
+        json_encoders = JSON_ENCODERS
+        use_enum_values = True
+
+
+class Notification(Entry):
+    """A notification addressed to one user."""
+
+    type: str = Field("notifications", const=True)
+    recipient_id: PyObjectId = Field(..., description="ID of the user receiving the notification")
+    title: str = Field(..., min_length=1, max_length=200)
+    summary: str | None = Field(
+        None,
+        max_length=1000,
+        description="Short text shown in compact notification lists.",
+    )
+    message: str | None = Field(None, max_length=5000)
+    level: NotificationLevel = Field(NotificationLevel.NORMAL)
+    created_at: datetime = Field(default_factory=lambda: datetime.now(tz=timezone.utc))
+    created_by: PyObjectId | None = Field(None, description="User ID that created the notification")
+    read_at: datetime | None = None
+    archived_at: datetime | None = None
+    grouping: NotificationGrouping | None = None
+    occurrence_count: int = Field(1, ge=1)
+    last_occurred_at: datetime = Field(default_factory=lambda: datetime.now(tz=timezone.utc))
+    occurrences: list[NotificationOccurrence] | None = Field(
+        None,
+        description="Individual occurrence details for grouped notifications.",
+    )
+
+    class Config(Entry.Config):
+        use_enum_values = True

pydatalab/src/pydatalab/mongo.py

@@ -374,6 +374,38 @@ def create_group_fts():
         "refcode", unique=True, name="unique refcode counter", background=background
     )
 
+    from pydatalab.config import CONFIG
+
+    if CONFIG.ENABLE_NOTIFICATIONS:
+        ret += db.notifications.create_index(
+            [("recipient_id", pymongo.ASCENDING), ("created_at", pymongo.DESCENDING)],
+            name="notification recipient and created",
+            background=background,
+        )
+        ret += db.notifications.create_index(
+            [("recipient_id", pymongo.ASCENDING), ("read_at", pymongo.ASCENDING)],
+            name="notification recipient and read",
+            background=background,
+        )
+        ret += db.notifications.create_index(
+            [("recipient_id", pymongo.ASCENDING), ("archived_at", pymongo.ASCENDING)],
+            name="notification recipient and archived",
+            background=background,
+        )
+        ret += db.notifications.create_index(
+            [
+                ("recipient_id", pymongo.ASCENDING),
+                ("grouping.key", pymongo.ASCENDING),
+                ("grouping.policy", pymongo.ASCENDING),
+                ("grouping.window_seconds", pymongo.ASCENDING),
+                ("grouping.max_occurrences", pymongo.ASCENDING),
+                ("occurrence_count", pymongo.ASCENDING),
+                ("last_occurred_at", pymongo.DESCENDING),
+            ],
+            name="notification grouping lookup",
+            background=background,
+        )
+
     return ret
 
 

pydatalab/src/pydatalab/notifications.py

@@ -0,0 +1,163 @@
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+from bson import ObjectId
+from pymongo import ReturnDocument
+
+from pydatalab.config import CONFIG
+from pydatalab.logger import LOGGER
+from pydatalab.models.notifications import (
+    Notification,
+    NotificationGrouping,
+    NotificationGroupPolicy,
+    NotificationLevel,
+    NotificationOccurrence,
+)
+from pydatalab.models.utils import PyObjectId
+from pydatalab.mongo import flask_mongo
+
+
+def _max_level(
+    current_level: NotificationLevel | str | None, new_level: NotificationLevel | str
+) -> str:
+    new_notification_level = NotificationLevel(new_level)
+    if current_level is None:
+        return new_notification_level.value
+
+    current_notification_level = NotificationLevel(current_level)
+    if new_notification_level.priority > current_notification_level.priority:
+        return new_notification_level.value
+
+    return current_notification_level.value
+
+
+def _find_grouped_notification(
+    *,
+    recipient_id: ObjectId,
+    title: str,
+    grouping: NotificationGrouping,
+    now: datetime,
+    session: Any | None = None,
+) -> dict | None:
+    query: dict[str, object] = {
+        "recipient_id": recipient_id,
+        "title": title,
+        "grouping.key": grouping.key,
+        "grouping.policy": grouping.policy,
+        "$expr": {"$lt": ["$occurrence_count", "$grouping.max_occurrences"]},
+        "archived_at": None,
+    }
+    if NotificationGroupPolicy(grouping.policy) == NotificationGroupPolicy.WINDOW:
+        query["grouping.window_seconds"] = grouping.window_seconds
+        query["last_occurred_at"] = {
+            "$gte": now - timedelta(seconds=int(grouping.window_seconds or 0))
+        }
+
+    return flask_mongo.db.notifications.find_one(
+        query,
+        sort=[("last_occurred_at", -1), ("created_at", -1)],
+        session=session,
+    )
+
+
+def _insert_notification(notification: Notification, *, session: Any | None = None) -> Notification:
+    result = flask_mongo.db.notifications.insert_one(
+        notification.dict(by_alias=True, exclude_none=True),
+        session=session,
+    )
+    notification.immutable_id = result.inserted_id
+    return notification
+
+
+def create_notification_with_result(
+    *,
+    recipient_id: str | ObjectId | PyObjectId,
+    title: str,
+    message: str | None = None,
+    summary: str | None = None,
+    level: NotificationLevel | str = NotificationLevel.NORMAL,
+    created_by: str | ObjectId | PyObjectId | None = None,
+    grouping: NotificationGrouping | dict[str, object] | None = None,
+    session: Any | None = None,
+) -> tuple[Notification, bool] | None:
+    """Create or group an in-app notification if the feature is enabled.
+
+    Returns:
+        A tuple of ``(notification, created)`` when a notification is created or
+        grouped. ``created`` is ``True`` when a new notification document was
+        inserted and ``False`` when the notification was folded into an existing
+        grouped notification. Returns ``None`` when notifications are disabled.
+    """
+
+    if not CONFIG.ENABLE_NOTIFICATIONS:
+        LOGGER.debug("Notifications are disabled; not creating notification %r", title)
+        return None
+
+    now = datetime.now(tz=timezone.utc)
+    recipient_object_id = ObjectId(recipient_id)
+
+    if grouping is None:
+        notification = Notification(
+            recipient_id=recipient_object_id,
+            title=title,
+            summary=summary,
+            message=message,
+            level=level,
+            created_by=ObjectId(created_by) if created_by else None,
+            occurrence_count=1,
+            last_occurred_at=now,
+        )
+        return _insert_notification(notification, session=session), True
+
+    if isinstance(grouping, dict):
+        grouping = NotificationGrouping(**grouping)
+
+    occurrence = NotificationOccurrence(
+        occurred_at=now,
+        message=message,
+        summary=summary,
+        level=level,
+        is_new=True,
+    )
+    grouped_notification = _find_grouped_notification(
+        recipient_id=recipient_object_id,
+        title=title,
+        grouping=grouping,
+        now=now,
+        session=session,
+    )
+
+    if grouped_notification is not None:
+        mongo_update = {
+            "$inc": {"occurrence_count": 1},
+            "$set": {
+                "message": message,
+                "summary": summary,
+                "last_occurred_at": now,
+                "level": _max_level(grouped_notification.get("level"), level),
+            },
+            "$push": {"occurrences": occurrence.dict(exclude_none=True)},
+            "$unset": {"read_at": ""},
+        }
+        updated_notification = flask_mongo.db.notifications.find_one_and_update(
+            {"_id": grouped_notification["_id"]},
+            mongo_update,
+            return_document=ReturnDocument.AFTER,
+            session=session,
+        )
+        return Notification(**updated_notification), False
+
+    notification = Notification(
+        recipient_id=recipient_object_id,
+        title=title,
+        summary=summary,
+        message=message,
+        level=level,
+        created_by=ObjectId(created_by) if created_by else None,
+        grouping=grouping,
+        occurrence_count=1,
+        last_occurred_at=now,
+        occurrences=[occurrence],
+    )
+
+    return _insert_notification(notification, session=session), True