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.desktop_notification import DesktopNotificationDelivery
from app.models.notification import Notification
from app.schemas.notification import GeneralNotificationFeed


async def _requeue_desktop_delivery(db: AsyncSession, notification_id: uuid.UUID) -> None:
    """Allow an escalated/reopened reminder to be delivered again by the desktop channel."""
    await db.execute(
        update(DesktopNotificationDelivery)
        .where(DesktopNotificationDelivery.notification_id == notification_id)
        .values(
            claim_token=None,
            claimed_at=None,
            delivered_at=None,
        )
    )


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,
    source_version: str | None = None,
    requires_action: bool = True,
    due_at: datetime | None = None,
) -> 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,
            source_version=source_version,
            dedupe_key=dedupe_key,
            requires_action=requires_action,
            due_at=due_at,
        ))


async def sync_state_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,
    source_version: str,
    active: bool,
    due_at: datetime | None = None,
    requires_action: bool = True,
    dedupe_key: str | None = None,
) -> None:
    """Synchronize one actionable business state into a recipient reminder."""
    dedupe_key = dedupe_key or f"state:{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 not active:
        if item and item.resolved_at is None:
            item.resolved_at = now
            item.read_at = item.read_at or now
        return
    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=source_version,
            dedupe_key=dedupe_key,
            requires_action=requires_action,
            due_at=due_at,
        ))
        return

    should_notify_again = item.resolved_at is not None or item.source_version != source_version
    item.study_id = study_id
    item.category = category
    item.priority = priority
    item.title = title
    item.message = message
    item.action_path = action_path
    item.source_version = source_version
    item.requires_action = requires_action
    item.due_at = due_at
    if should_notify_again:
        item.read_at = None
        item.created_at = now
        await _requeue_desktop_delivery(db, item.id)
    item.resolved_at = None


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,
    due_at: datetime | None = None,
) -> 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,
            requires_action=True,
            due_at=due_at,
        ))
        return
    previous_count = int(item.source_version or 0)
    item.category = category
    item.priority = priority
    item.title = title
    item.message = message
    item.action_path = action_path
    item.source_version = version
    item.due_at = due_at
    if item.resolved_at is not None or count > previous_count:
        item.read_at = None
        item.created_at = now
        await _requeue_desktop_delivery(db, item.id)
    item.resolved_at = None


async def resolve_missing_recipient_sources(
    db: AsyncSession,
    *,
    study_id: uuid.UUID,
    recipient_id: uuid.UUID,
    source_type: str,
    active_source_ids: set[str],
) -> None:
    filters = [
        Notification.study_id == study_id,
        Notification.recipient_id == recipient_id,
        Notification.source_type == source_type,
        Notification.resolved_at.is_(None),
    ]
    if active_source_ids:
        filters.append(Notification.source_id.not_in(active_source_ids))
    now = datetime.now(timezone.utc)
    await db.execute(
        update(Notification)
        .where(*filters)
        .values(resolved_at=now, read_at=func.coalesce(Notification.read_at, now))
    )


async def resolve_source_notifications(
    db: AsyncSession,
    *,
    source_type: str,
    source_id: str,
    recipient_id: uuid.UUID | None = None,
) -> None:
    filters = [
        Notification.source_type == source_type,
        Notification.source_id == source_id,
        Notification.resolved_at.is_(None),
    ]
    if recipient_id is not None:
        filters.append(Notification.recipient_id == recipient_id)
    now = datetime.now(timezone.utc)
    await db.execute(
        update(Notification)
        .where(*filters)
        .values(resolved_at=now, read_at=func.coalesce(Notification.read_at, now))
    )


async def resolve_recipient_study_notifications(
    db: AsyncSession,
    *,
    study_id: uuid.UUID,
    recipient_id: uuid.UUID,
) -> None:
    now = datetime.now(timezone.utc)
    await db.execute(
        update(Notification)
        .where(
            Notification.study_id == study_id,
            Notification.recipient_id == recipient_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,
    skip: int = 0,
    limit: int = 10,
    category: str | None = None,
    unread_only: bool = False,
    requires_action: bool | None = None,
) -> 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)

    filters = [
        Notification.study_id == study_id,
        Notification.recipient_id == recipient_id,
        Notification.resolved_at.is_(None),
    ]
    if category:
        filters.append(Notification.category == category)
    if unread_only:
        filters.append(Notification.read_at.is_(None))
    if requires_action is not None:
        filters.append(Notification.requires_action.is_(requires_action))
    total_count = int(await db.scalar(
        select(func.count(Notification.id)).where(*filters)
    ) or 0)
    items = (await db.scalars(
        select(Notification)
        .where(*filters)
        .order_by(
            Notification.read_at.is_not(None),
            Notification.created_at.desc(),
        )
        .offset(max(0, skip))
        .limit(max(1, min(limit, 100)))
    )).all()
    return GeneralNotificationFeed(
        unread_count=unread_count,
        total_count=total_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:
        now = datetime.now(timezone.utc)
        item.read_at = now
        if not item.requires_action:
            item.resolved_at = now
        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:
    now = datetime.now(timezone.utc)
    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),
            Notification.requires_action.is_(False),
        )
        .values(read_at=now, resolved_at=now)
    )
    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),
            Notification.requires_action.is_(True),
        )
        .values(read_at=now)
    )
    await db.commit()