leocrm/app/core/notifications.py

"""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,
    }