"""Notification service — create and manage user notifications. As of B-NOTIF-EVT, the primary entry point is post_system_message() which posts to the Communication system channel. create_notification() is retained as a deprecated backward-compat wrapper that delegates to post_system_message(). """ from __future__ import annotations import logging import uuid from datetime import UTC from typing import Any from sqlalchemy import func, select, update from sqlalchemy.ext.asyncio import AsyncSession from app.models.notification import ( Notification, ) logger = logging.getLogger(__name__) # Re-export post_system_message from kommunikation services for convenience async def post_system_message( db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, message_type: str, title: str, body: str | None = None, entity_type: str | None = None, entity_id: uuid.UUID | None = None, severity: str = "info", ): """Post a typed system message to the tenant system channel. Delegates to kommunikation plugin via contract registry. Returns None if the kommunikation plugin is not active (graceful degradation). """ from app.plugins.builtins.contracts import get_contract komm_contract = get_contract("kommunikation") if komm_contract is None: logger.warning("kommunikation plugin not available — system message not posted") return None return await komm_contract.post_system_message( db, tenant_id, user_id, message_type, title, body, entity_type, entity_id, severity, ) async def create_notification( db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, type: str, title: str, body: str | None = None, entity_type: str | None = None, entity_id: uuid.UUID | None = None, ) -> Notification | None: """Create a new notification for a user if they have not disabled this type. .. deprecated:: B-NOTIF-EVT Use post_system_message() instead. This wrapper delegates to post_system_message() and also creates a legacy Notification record for backward compatibility with existing routes and frontend. Returns None if the user has opted out of this notification type. """ # Delegate to post_system_message for the comm channel comm_msg = await post_system_message( db, tenant_id, user_id, type, title, body, entity_type, entity_id, severity="info", ) if comm_msg is None: # User has muted this type — don't create legacy record either return None # Also create legacy Notification record for backward compat notif = Notification( tenant_id=tenant_id, user_id=user_id, type=type, title=title, body=body, entity_type=entity_type, entity_id=entity_id, ) db.add(notif) await db.flush() # Publish notification.created event (backward compat) from app.core.event_bus import get_event_bus event_bus = get_event_bus() await event_bus.publish("notification.created", { "notification_id": str(notif.id), "tenant_id": str(tenant_id), "user_id": str(user_id), "type": type, "title": title, "entity_type": entity_type, "entity_id": str(entity_id) if entity_id else None, "comm_message_id": str(comm_msg.id), }) return notif async def list_notifications( db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, page: int = 1, page_size: int = 25, ) -> dict[str, Any]: """List notifications for a user, unread first, then by created_at desc.""" offset = (page - 1) * page_size count_q = ( select(func.count()) .select_from(Notification) .where( Notification.tenant_id == tenant_id, Notification.user_id == user_id, ) ) total = (await db.execute(count_q)).scalar() or 0 q = ( select(Notification) .where( Notification.tenant_id == tenant_id, Notification.user_id == user_id, ) .order_by( Notification.read_at.isnot(None), Notification.created_at.desc(), ) .offset(offset) .limit(page_size) ) result = await db.execute(q) items = result.scalars().all() return { "items": [_notification_to_dict(n) for n in items], "total": total, "page": page, "page_size": page_size, } async def mark_notification_read( db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, notification_id: uuid.UUID, ) -> Notification | None: """Mark a notification as read.""" from datetime import datetime q = ( update(Notification) .where( Notification.id == notification_id, Notification.tenant_id == tenant_id, Notification.user_id == user_id, ) .values(read_at=datetime.now(UTC)) .returning(Notification) ) result = await db.execute(q) row = result.scalar_one_or_none() return row async def get_unread_count( db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, ) -> int: """Get unread notification count for a user.""" q = ( select(func.count()) .select_from(Notification) .where( Notification.tenant_id == tenant_id, Notification.user_id == user_id, Notification.read_at.is_(None), ) ) result = await db.execute(q) return result.scalar() or 0 def _notification_to_dict(n: Notification) -> dict[str, Any]: """Convert a notification to a dict.""" return { "id": str(n.id), "type": n.type, "title": n.title, "body": n.body, "entity_type": n.entity_type, "entity_id": str(n.entity_id) if n.entity_id else None, "read_at": n.read_at.isoformat() if n.read_at else None, "created_at": n.created_at.isoformat() if n.created_at else None, }