ctms/backend/app/services/notification_service.py

from __future__ import annotations

import uuid
from datetime import datetime, timezone
from typing import Iterable

from fastapi import HTTPException, status
from sqlalchemy import func, select, update
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.notification import Notification
from app.schemas.notification import GeneralNotificationFeed


async def create_recipient_notifications(
    db: AsyncSession,
    *,
    study_id: uuid.UUID,
    recipient_ids: Iterable[uuid.UUID],
    category: str,
    priority: str,
    title: str,
    message: str,
    action_path: str | None,
    source_type: str,
    source_id: str,
    dedupe_key: str,
) -> None:
    recipients = set(recipient_ids)
    if not recipients:
        return
    existing = set((await db.scalars(
        select(Notification.recipient_id).where(
            Notification.recipient_id.in_(recipients),
            Notification.dedupe_key == dedupe_key,
        )
    )).all())
    for recipient_id in recipients - existing:
        db.add(Notification(
            study_id=study_id,
            recipient_id=recipient_id,
            category=category,
            priority=priority,
            title=title,
            message=message,
            action_path=action_path,
            source_type=source_type,
            source_id=source_id,
            dedupe_key=dedupe_key,
        ))


async def sync_aggregate_notification(
    db: AsyncSession,
    *,
    study_id: uuid.UUID,
    recipient_id: uuid.UUID,
    category: str,
    priority: str,
    title: str,
    message: str,
    action_path: str,
    source_type: str,
    source_id: str,
    count: int,
) -> None:
    dedupe_key = f"aggregate:{source_type}:{source_id}"
    item = await db.scalar(select(Notification).where(
        Notification.recipient_id == recipient_id,
        Notification.dedupe_key == dedupe_key,
    ))
    now = datetime.now(timezone.utc)
    if count <= 0:
        if item and item.resolved_at is None:
            item.resolved_at = now
            item.read_at = item.read_at or now
        return
    version = str(count)
    if item is None:
        db.add(Notification(
            study_id=study_id,
            recipient_id=recipient_id,
            category=category,
            priority=priority,
            title=title,
            message=message,
            action_path=action_path,
            source_type=source_type,
            source_id=source_id,
            source_version=version,
            dedupe_key=dedupe_key,
        ))
        return
    previous_count = int(item.source_version or 0)
    item.priority = priority
    item.title = title
    item.message = message
    item.action_path = action_path
    item.source_version = version
    if item.resolved_at is not None or count > previous_count:
        item.read_at = None
        item.created_at = now
    item.resolved_at = None


async def resolve_source_notifications(
    db: AsyncSession,
    *,
    source_type: str,
    source_id: str,
) -> None:
    now = datetime.now(timezone.utc)
    await db.execute(
        update(Notification)
        .where(
            Notification.source_type == source_type,
            Notification.source_id == source_id,
            Notification.resolved_at.is_(None),
        )
        .values(resolved_at=now, read_at=func.coalesce(Notification.read_at, now))
    )


async def list_feed(
    db: AsyncSession,
    *,
    study_id: uuid.UUID,
    recipient_id: uuid.UUID,
    limit: int = 10,
) -> GeneralNotificationFeed:
    active_filter = (
        Notification.study_id == study_id,
        Notification.recipient_id == recipient_id,
        Notification.resolved_at.is_(None),
    )
    unread_count = int(await db.scalar(
        select(func.count(Notification.id)).where(*active_filter, Notification.read_at.is_(None))
    ) or 0)
    items = (await db.scalars(
        select(Notification)
        .where(*active_filter)
        .order_by(Notification.read_at.is_not(None), Notification.created_at.desc())
        .limit(max(1, min(limit, 50)))
    )).all()
    return GeneralNotificationFeed(unread_count=unread_count, items=list(items))


async def mark_read(
    db: AsyncSession,
    *,
    study_id: uuid.UUID,
    recipient_id: uuid.UUID,
    notification_id: uuid.UUID,
) -> Notification:
    item = await db.scalar(select(Notification).where(
        Notification.id == notification_id,
        Notification.study_id == study_id,
        Notification.recipient_id == recipient_id,
        Notification.resolved_at.is_(None),
    ))
    if item is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="通知不存在")
    if item.read_at is None:
        item.read_at = datetime.now(timezone.utc)
        await db.commit()
        await db.refresh(item)
    return item


async def mark_all_read(
    db: AsyncSession,
    *,
    study_id: uuid.UUID,
    recipient_id: uuid.UUID,
) -> None:
    await db.execute(
        update(Notification)
        .where(
            Notification.study_id == study_id,
            Notification.recipient_id == recipient_id,
            Notification.resolved_at.is_(None),
            Notification.read_at.is_(None),
        )
        .values(read_at=datetime.now(timezone.utc))
    )
    await db.commit()