[owl] Notification Schema Updates (#916)

* Update schema

* DB migration

Author: haoshan98

clients/python/src/jamaibase/client.py

@@ -4108,12 +4108,14 @@ async def delete_notification(
 
     async def set_opened(
         self,
-        notification_group_id: str,
+        notification_group_ids: str | list[str],
         **kwargs,
     ) -> OkResponse:
+        if isinstance(notification_group_ids, str):
+            notification_group_ids = [notification_group_ids]
         return await self._patch(
             "/v2/notifications/opened",
-            params=dict(notification_group_id=notification_group_id),
+            params=dict(notification_group_ids=notification_group_ids),
             response_model=OkResponse,
             **kwargs,
         )
@@ -6817,8 +6819,8 @@ def get_notification(self, notification_group_id: str, **kwargs) -> Notification
     def delete_notification(self, notification_group_id: str, **kwargs) -> OkResponse:
         return LOOP.run(super().delete_notification(notification_group_id, **kwargs))
 
-    def set_opened(self, notification_group_id: str, **kwargs) -> OkResponse:
-        return LOOP.run(super().set_opened(notification_group_id, **kwargs))
+    def set_opened(self, notification_group_ids: str | list[str], **kwargs) -> OkResponse:
+        return LOOP.run(super().set_opened(notification_group_ids, **kwargs))
 
     def set_all_opened(self, **kwargs) -> OkResponse:
         return LOOP.run(super().set_all_opened(**kwargs))

clients/python/src/jamaibase/types/db.py

@@ -1374,10 +1374,18 @@ class NotificationGroupCreate(_BaseModel):
         None,
         description="ID of the user who triggered the event.",
     )
+    subject_id: str | None = Field(
+        None,
+        description="ID of the subject user (e.g. invitee, new owner).",
+    )
     recipient_ids: list[str] = Field(
         [],
         description="Explicit recipient user IDs. Used for fan-out.",
     )
+    message: str = Field(
+        "",
+        description="Notification text (Markdown).",
+    )
 
 
 class NotificationGroup_(NotificationGroupCreate, _TableBase):
@@ -1391,6 +1399,10 @@ class NotificationGroupRead(NotificationGroup_):
         None,
         description="User who triggered the event.",
     )
+    subject: "User_ | None" = Field(
+        None,
+        description="Subject user (e.g. invitee, new owner).",
+    )
 
 
 class NotificationCreate(_BaseModel):
@@ -1400,8 +1412,8 @@ class NotificationCreate(_BaseModel):
     notification_group_id: str = Field(
         description="Notification group ID.",
     )
-    body: str = Field(
-        description="Notification body text.",
+    message: str = Field(
+        description="Notification text (Markdown).",
     )
 
 

services/api/src/owl/db/__init__.py

@@ -649,6 +649,56 @@ async def _migrate_modeltype_enum(engine: AsyncEngine) -> bool:
         return True
 
 
+async def _migrate_notification_schema(engine: AsyncEngine) -> bool:
+    """
+    Migrate NotificationGroup and Notification tables:
+    - Add `subject_id` (FK→User SET NULL) and `message` columns to NotificationGroup
+    - Rename `body` → `message` in Notification
+    """
+    group_table = "NotificationGroup"
+    notif_table = "Notification"
+
+    async with engine.connect() as conn:
+        group_has_message = await _check_column_exists(conn, group_table, "message")
+        group_has_subject = await _check_column_exists(conn, group_table, "subject_id")
+        notif_has_message = await _check_column_exists(conn, notif_table, "message")
+
+    if group_has_message and group_has_subject and notif_has_message:
+        return False
+
+    async with engine.begin() as conn:
+        if not group_has_subject:
+            await conn.execute(
+                text(
+                    f"""
+                    ALTER TABLE {SCHEMA}."{group_table}"
+                    ADD COLUMN subject_id TEXT DEFAULT NULL
+                    REFERENCES {SCHEMA}."User"(id) ON DELETE SET NULL;
+                    """
+                )
+            )
+        if not group_has_message:
+            await conn.execute(
+                text(
+                    f"""
+                    ALTER TABLE {SCHEMA}."{group_table}"
+                    ADD COLUMN message TEXT NOT NULL DEFAULT '';
+                    """
+                )
+            )
+        if not notif_has_message:
+            await conn.execute(
+                text(
+                    f"""
+                    ALTER TABLE {SCHEMA}."{notif_table}"
+                    RENAME COLUMN body TO message;
+                    """
+                )
+            )
+    logger.info("Successfully migrated notification schema (subject_id, message, body→message).")
+    return True
+
+
 async def migrate_db():
     engine = create_db_engine_async()
     migrated = [
@@ -664,6 +714,7 @@ async def migrate_db():
         await _backfill_price_plan_image_products(engine),
         await _migrate_verification_codes(engine),
         await _migrate_reasoning_jsonb_keys(engine),
+        await _migrate_notification_schema(engine),
     ]
     if any(migrated):
         logger.success("DB migrations performed.")

services/api/src/owl/db/models/oss.py

@@ -1627,7 +1627,29 @@ class NotificationGroup(_TableBase, table=True):
         ondelete="SET NULL",
         description="ID of the user who triggered the event.",
     )
-    actor: "User" = _relationship(None, selectin=True, cascade=None)
+    subject_id: str | None = SqlField(
+        None,
+        foreign_key="User.id",
+        nullable=True,
+        ondelete="SET NULL",
+        description="ID of the subject user (e.g. invitee, new owner).",
+    )
+    message: str = SqlField(
+        "",
+        description="Notification message text (Markdown).",
+    )
+    actor: "User" = _relationship(
+        None,
+        selectin=True,
+        cascade=None,
+        sa_kwargs=dict(foreign_keys="[NotificationGroup.actor_id]"),
+    )
+    subject: "User" = _relationship(
+        None,
+        selectin=True,
+        cascade=None,
+        sa_kwargs=dict(foreign_keys="[NotificationGroup.subject_id]"),
+    )
     notifications: list["Notification"] = _relationship("notification_group")
 
 
@@ -1644,8 +1666,9 @@ class Notification(_TableBase, table=True):
         ondelete="CASCADE",
         description="Notification group ID.",
     )
-    body: str = SqlField(
-        description="Notification body text.",
+    message: str = SqlField(
+        "",
+        description="Notification message text (Markdown).",
     )
     opened_at: DatetimeUTC | None = SqlField(
         None,

services/api/src/owl/routers/notification.py

@@ -55,8 +55,9 @@ async def create_notification_group(
     intent = NotificationIntent(
         scope=body.scope,
         event_type=body.event_type,
-        meta=body.meta,
+        message=body.message,
         actor_id=body.actor_id,
+        subject_id=body.subject_id,
         organization_id=body.organization_id,
         project_id=body.project_id,
         recipient_ids=body.recipient_ids,
@@ -209,23 +210,28 @@ async def delete_notification(
 
 @router.patch(
     "/v2/notifications/opened",
-    summary="Mark a notification as opened.",
+    summary="Mark one or more notifications as opened.",
     description="Permissions: notification owner.",
 )
 @handle_exception
 async def set_opened(
     user: Annotated[UserAuth, Depends(auth_user_service_key)],
     session: Annotated[AsyncSession, Depends(yield_async_session)],
-    notification_group_id: Annotated[
-        str, Query(min_length=1, description="Notification group ID.")
+    notification_group_ids: Annotated[
+        list[str], Query(min_length=1, description="List of notification group IDs.")
     ],
 ) -> OkResponse:
-    notif = await session.get(Notification, (user.id, notification_group_id))
-    if notif is None or notif.deleted_at is not None:
-        raise ResourceNotFoundError("Notification not found.")
     timestamp = now()
-    notif.opened_at = timestamp
-    notif.updated_at = timestamp
+    await session.exec(
+        update(Notification)
+        .where(
+            Notification.user_id == user.id,
+            Notification.notification_group_id.in_(notification_group_ids),
+            Notification.opened_at.is_(None),
+            Notification.deleted_at.is_(None),
+        )
+        .values(opened_at=timestamp, updated_at=timestamp)
+    )
     await session.commit()
     return OkResponse()