ctms/backend/app/services/document_service.py

from __future__ import annotations

import hashlib
import json
import uuid
from datetime import datetime, timezone
from pathlib import Path
from typing import Iterable

import aiofiles
from fastapi import HTTPException, UploadFile, status
from fastapi.responses import FileResponse
from sqlalchemy import and_, delete as sa_delete, or_, select, update as sa_update
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.deps import get_cra_site_scope
from app.core.deps import get_operator_role_label, is_system_admin
from app.core.project_permissions import role_has_api_permission
from app.crud import acknowledgement as acknowledgement_crud
from app.crud import distribution as distribution_crud
from app.crud import document as document_crud
from app.crud import etmf as etmf_crud
from app.crud import document_version as version_crud
from app.crud import member as member_crud
from app.crud import site as site_crud
from app.crud import study as study_crud
from app.crud import user as user_crud
from app.models.acknowledgement import Acknowledgement, AcknowledgementType
from app.models.audit_log import AuditLog
from app.models.distribution import Distribution, DistributionStatus, DistributionTargetType
from app.models.document import Document, DocumentScopeType, DocumentStatus
from app.models.document_version import DocumentVersion, DocumentVersionStatus
from app.models.desktop_notification import DesktopNotificationDelivery
from app.models.study import Study
from app.schemas.acknowledgement import AcknowledgementCreate
from app.schemas.distribution import DistributionCreate, DistributionRead, DistributionStats
from app.schemas.document import DocumentCreate, DocumentDetail, DocumentSummary, DocumentUpdate
from app.schemas.document_version import DocumentVersionRead, DocumentVersionSummary
from app.schemas.notification import NotificationItem
from app.schemas.user import UserDisplay

UPLOAD_ROOT = Path(__file__).resolve().parent.parent / "uploads" / "documents"

DOCUMENT_ACTION_PERMISSIONS = {
    "view": "documents:read",
    "ack": "documents:read",
    "create_document": "documents:create",
    "update_document": "documents:update",
    "create_version": "documents:update",
    "distribute": "documents:update",
    "delete_document": "documents:delete",
}


def _audit_detail(before: dict | None, after: dict | None) -> str:
    payload = {"before": before, "after": after}
    return json.dumps(payload, ensure_ascii=True)


def _doc_snapshot(doc: Document) -> dict:
    status = doc.status.value if hasattr(doc.status, "value") else str(doc.status)
    return {
        "id": str(doc.id),
        "trial_id": str(doc.trial_id),
        "site_id": str(doc.site_id) if doc.site_id else None,
        "etmf_node_id": str(doc.etmf_node_id) if doc.etmf_node_id else None,
        "doc_no": doc.doc_no,
        "status": status,
        "current_effective_version_id": str(doc.current_effective_version_id) if doc.current_effective_version_id else None,
    }


def _version_snapshot(version: DocumentVersion) -> dict:
    status = version.status.value if hasattr(version.status, "value") else str(version.status)
    return {
        "id": str(version.id),
        "document_id": str(version.document_id),
        "version_no": version.version_no,
        "status": status,
        "file_hash": version.file_hash,
    }


def _version_read(version: DocumentVersion, users_by_id: dict[uuid.UUID, object] | None = None) -> DocumentVersionRead:
    creator = users_by_id.get(version.created_by) if users_by_id and version.created_by else None
    return DocumentVersionRead.model_validate(version).model_copy(
        update={"created_by_user": UserDisplay.model_validate(creator) if creator else None}
    )


async def _ensure_study_access(db: AsyncSession, trial_id: uuid.UUID, current_user, action: str):
    study = await study_crud.get(db, trial_id)
    if not study:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="项目不存在")
    if is_system_admin(current_user):
        return None
    membership = await member_crud.get_member(db, trial_id, current_user.id)
    if not membership or not membership.is_active:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="不是项目成员")
    endpoint_key = DOCUMENT_ACTION_PERMISSIONS.get(action, "documents:update")
    allowed = await role_has_api_permission(db, trial_id, membership.role_in_study, endpoint_key, check_prerequisites=False)
    if not allowed:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="权限不足")
    return membership


async def _ensure_study_member(db: AsyncSession, trial_id: uuid.UUID, current_user):
    study = await study_crud.get(db, trial_id)
    if not study:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="项目不存在")
    if is_system_admin(current_user):
        return None
    membership = await member_crud.get_member(db, trial_id, current_user.id)
    if not membership or not membership.is_active:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="不是项目成员")
    return membership


async def create_document(
    db: AsyncSession,
    doc_in: DocumentCreate,
    current_user,
) -> Document:
    await _ensure_study_access(db, doc_in.trial_id, current_user, action="create_document")
    etmf_node = None
    if doc_in.etmf_node_id:
        etmf_node = await etmf_crud.get_node(db, doc_in.etmf_node_id)
        if not etmf_node:
            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="eTMF目录不存在")
        if etmf_node.study_id != doc_in.trial_id:
            raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="eTMF目录不属于当前项目")
        if etmf_node.scope_type == DocumentScopeType.SITE and not doc_in.site_id:
            raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="中心级eTMF目录需指定中心")

    scope_type = etmf_node.scope_type if etmf_node else doc_in.scope_type
    if scope_type == DocumentScopeType.SITE and not doc_in.site_id:
        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="分中心文件需指定分中心")
    site_id = doc_in.site_id if scope_type == DocumentScopeType.SITE else None
    if site_id:
        site = await site_crud.get_site(db, site_id)
        if not site or not site.is_active:
            raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="中心已停用")
    existing = await document_crud.get_by_trial_doc_no(db, doc_in.trial_id, doc_in.doc_no)
    if existing:
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="文档编号已存在")

    doc = Document(
        trial_id=doc_in.trial_id,
        site_id=site_id,
        etmf_node_id=doc_in.etmf_node_id,
        doc_no=doc_in.doc_no,
        doc_type=doc_in.doc_type,
        title=doc_in.title,
        scope_type=scope_type,
        owner_id=doc_in.owner_id,
        description=doc_in.description,
    )
    db.add(doc)
    db.add(
        AuditLog(
            study_id=doc_in.trial_id,
            entity_type="DOCUMENT",
            entity_id=doc.id,
            action="DOCUMENT_CREATED",
            detail=_audit_detail(None, _doc_snapshot(doc)),
            operator_id=current_user.id,
            operator_role=await get_operator_role_label(db, doc.trial_id, current_user),
        )
    )
    await db.commit()
    await db.refresh(doc)
    return doc


async def list_documents(
    db: AsyncSession,
    *,
    trial_id: uuid.UUID,
    site_id: uuid.UUID | None,
    doc_type: str | None,
    status: str | None,
    scope_type: str | None,
    etmf_node_id: uuid.UUID | None = None,
    skip: int,
    limit: int,
    current_user,
) -> list[DocumentSummary]:
    await _ensure_study_access(db, trial_id, current_user, action="view")
    cra_scope = await get_cra_site_scope(db, trial_id, current_user)
    cra_site_ids = cra_scope[0] if cra_scope else None
    docs = await document_crud.list_documents(
        db,
        trial_id=trial_id,
        site_id=site_id,
        doc_type=doc_type,
        status=status,
        scope_type=scope_type,
        etmf_node_id=etmf_node_id,
        cra_site_ids=cra_site_ids,
        skip=skip,
        limit=limit,
    )

    summaries: list[DocumentSummary] = []
    for doc in docs:
        current_version = None
        if doc.current_effective_version_id:
            version = await version_crud.get(db, doc.current_effective_version_id)
            if version:
                current_version = DocumentVersionSummary.model_validate(version)
        summaries.append(
            DocumentSummary(
                id=doc.id,
                trial_id=doc.trial_id,
                site_id=doc.site_id,
                etmf_node_id=doc.etmf_node_id,
                doc_type=doc.doc_type,
                title=doc.title,
                scope_type=doc.scope_type,
                owner_id=doc.owner_id,
                current_effective_version_id=doc.current_effective_version_id,
                current_effective_version=current_version,
                created_at=doc.created_at,
                updated_at=doc.updated_at,
            )
        )
    return summaries


async def get_document_detail(
    db: AsyncSession,
    document_id: uuid.UUID,
    current_user,
) -> DocumentDetail:
    doc = await document_crud.get(db, document_id)
    if not doc:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="文档不存在")
    await _ensure_study_access(db, doc.trial_id, current_user, action="view")
    cra_scope = await get_cra_site_scope(db, doc.trial_id, current_user)
    if cra_scope and doc.scope_type == DocumentScopeType.SITE and doc.site_id not in cra_scope[0]:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="权限不足")
    versions = await version_crud.list_by_document(db, document_id)

    current_version = None
    if doc.current_effective_version_id:
        version = await version_crud.get(db, doc.current_effective_version_id)
        if version:
            current_version = DocumentVersionSummary.model_validate(version)

    distribution_stats = await _aggregate_distribution_stats(db, doc.current_effective_version_id)
    owner = None
    if doc.owner_id:
        owner_obj = await user_crud.get_by_id(db, doc.owner_id)
        if owner_obj:
            owner = UserDisplay.model_validate(owner_obj)
    creator_ids = {v.created_by for v in versions if v.created_by}
    version_creators = await user_crud.get_users_by_ids(db, creator_ids)
    return DocumentDetail(
        id=doc.id,
        trial_id=doc.trial_id,
        site_id=doc.site_id,
        etmf_node_id=doc.etmf_node_id,
        doc_type=doc.doc_type,
        title=doc.title,
        scope_type=doc.scope_type,
        owner_id=doc.owner_id,
        current_effective_version_id=doc.current_effective_version_id,
        current_effective_version=current_version,
        description=doc.description,
        created_at=doc.created_at,
        updated_at=doc.updated_at,
        owner=owner,
        version_timeline=[_version_read(v, version_creators) for v in versions],
        distribution_stats=distribution_stats,
    )


async def update_document(
    db: AsyncSession,
    document_id: uuid.UUID,
    doc_in: DocumentUpdate,
    current_user,
) -> Document:
    doc = await document_crud.get(db, document_id)
    if not doc:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="文档不存在")
    await _ensure_study_access(db, doc.trial_id, current_user, action="update_document")

    values = doc_in.model_dump(exclude_unset=True)
    if not values:
        return doc

    next_scope = values.get("scope_type", doc.scope_type)
    next_scope_value = getattr(next_scope, "value", next_scope)
    next_site_id = values.get("site_id", doc.site_id)
    if next_scope_value == DocumentScopeType.SITE.value:
        if not next_site_id:
            raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="分中心文件需指定分中心")
        site = await site_crud.get_site(db, next_site_id)
        if not site or site.study_id != doc.trial_id:
            raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="分中心不属于当前项目")
        if not site.is_active:
            raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="中心已停用")
    else:
        values["site_id"] = None

    before = _doc_snapshot(doc)
    await document_crud.update(db, document_id, values, commit=False)
    updated = await document_crud.get(db, document_id)
    if not updated:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="文档不存在")
    db.add(
        AuditLog(
            study_id=updated.trial_id,
            entity_type="DOCUMENT",
            entity_id=updated.id,
            action="DOCUMENT_UPDATED",
            detail=_audit_detail(before, _doc_snapshot(updated)),
            operator_id=current_user.id,
            operator_role=await get_operator_role_label(db, updated.trial_id, current_user),
        )
    )
    await db.commit()
    await db.refresh(updated)
    return updated


async def create_version(
    db: AsyncSession,
    document_id: uuid.UUID,
    *,
    version_no: str,
    version_date: str,
    change_summary: str | None,
    parent_version_id: uuid.UUID | None,
    file: UploadFile,
    current_user,
) -> DocumentVersion:
    doc = await document_crud.get(db, document_id)
    if not doc:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="文档不存在")
    await _ensure_study_access(db, doc.trial_id, current_user, action="create_version")

    previous_effective_id = doc.current_effective_version_id

    existing = await version_crud.get_by_document_version_no(db, document_id, version_no)
    if existing:
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="版本号已存在")

    if parent_version_id:
        parent = await version_crud.get(db, parent_version_id)
        if not parent or parent.document_id != document_id:
            raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="父版本不存在或不属于该文档")

    content = await file.read()
    file_hash = hashlib.sha256(content).hexdigest()
    dest_dir = UPLOAD_ROOT / str(document_id)
    dest_dir.mkdir(parents=True, exist_ok=True)
    unique_name = f"{uuid.uuid4()}{Path(file.filename).suffix}"
    dest_path = dest_dir / unique_name
    async with aiofiles.open(dest_path, "wb") as out_file:
        await out_file.write(content)

    now = datetime.now(timezone.utc)
    effective_at = now
    try:
        parsed = datetime.fromisoformat(version_date)
        if parsed.tzinfo is None:
            parsed = parsed.replace(tzinfo=timezone.utc)
        effective_at = parsed
    except ValueError:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="版本日期格式不正确")
    version = DocumentVersion(
        document_id=document_id,
        version_no=version_no,
        parent_version_id=parent_version_id,
        status=DocumentVersionStatus.EFFECTIVE,
        effective_at=effective_at,
        file_uri=str(dest_path),
        file_hash=file_hash,
        file_size=len(content),
        mime_type=file.content_type,
        change_summary=change_summary,
        created_by=current_user.id,
    )
    db.add(version)
    await db.flush()
    await db.execute(
        sa_update(DocumentVersion)
        .where(
            DocumentVersion.document_id == document_id,
            DocumentVersion.status == DocumentVersionStatus.EFFECTIVE,
            DocumentVersion.id != version.id,
        )
        .values(status=DocumentVersionStatus.SUPERSEDED, superseded_at=now)
    )
    await db.execute(
        sa_update(Document)
        .where(Document.id == document_id)
        .values(current_effective_version_id=version.id)
    )
    db.add(
        AuditLog(
            study_id=doc.trial_id,
            entity_type="DOCUMENT_VERSION",
            entity_id=version.id,
            action="VERSION_CREATED",
            detail=_audit_detail(None, _version_snapshot(version)),
            operator_id=current_user.id,
            operator_role=await get_operator_role_label(db, doc.trial_id, current_user),
        )
    )

    if previous_effective_id:
        previous_distributions = await distribution_crud.list_by_version(db, previous_effective_id)
        if previous_distributions:
            seen: set[tuple[str, str]] = set()
            for dist in previous_distributions:
                key = (dist.target_type, dist.target_id)
                if key in seen:
                    continue
                seen.add(key)
                db.add(
                    Distribution(
                        document_id=document_id,
                        version_id=version.id,
                        target_type=dist.target_type,
                        target_id=dist.target_id,
                        created_by=current_user.id,
                    )
                )
    await db.commit()
    await db.refresh(version)
    return version


async def delete_version(
    db: AsyncSession,
    version_id: uuid.UUID,
    current_user,
) -> None:
    version = await version_crud.get(db, version_id)
    if not version:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="版本不存在")
    doc = await document_crud.get(db, version.document_id)
    if not doc:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="文档不存在")
    await _ensure_study_access(db, doc.trial_id, current_user, action="delete_document")

    distribution_ids = (
        await db.execute(
            select(Distribution.id).where(Distribution.version_id == version.id)
        )
    ).scalars().all()
    if distribution_ids:
        await db.execute(sa_delete(Acknowledgement).where(Acknowledgement.distribution_id.in_(distribution_ids)))
        await db.execute(sa_delete(Distribution).where(Distribution.id.in_(distribution_ids)))

    before = _version_snapshot(version)

    if doc.current_effective_version_id == version.id:
        candidates = await version_crud.list_by_document(db, version.document_id)
        candidate = next((item for item in candidates if item.id != version.id), None)
        if candidate:
            update_values: dict[str, object] = {"status": DocumentVersionStatus.EFFECTIVE, "superseded_at": None}
            if not candidate.effective_at:
                update_values["effective_at"] = datetime.now(timezone.utc)
            await db.execute(
                sa_update(DocumentVersion)
                .where(DocumentVersion.id == candidate.id)
                .values(**update_values)
            )
            await db.execute(
                sa_update(Document)
                .where(Document.id == doc.id)
                .values(current_effective_version_id=candidate.id)
            )
        else:
            await db.execute(
                sa_update(Document)
                .where(Document.id == doc.id)
                .values(current_effective_version_id=None)
            )

    await db.execute(sa_delete(DocumentVersion).where(DocumentVersion.id == version.id))
    db.add(
        AuditLog(
            study_id=doc.trial_id,
            entity_type="DOCUMENT_VERSION",
            entity_id=version.id,
            action="VERSION_DELETED",
            detail=_audit_detail(before, None),
            operator_id=current_user.id,
            operator_role=await get_operator_role_label(db, doc.trial_id, current_user),
        )
    )
    await db.commit()

    file_path = Path(version.file_uri)
    if file_path.exists():
        try:
            file_path.unlink()
        except OSError:
            pass


async def delete_document(
    db: AsyncSession,
    document_id: uuid.UUID,
    current_user,
) -> Document:
    doc = await document_crud.get(db, document_id)
    if not doc:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="文档不存在")
    await _ensure_study_access(db, doc.trial_id, current_user, action="delete_document")

    if doc.status == DocumentStatus.ARCHIVED:
        return doc

    before = _doc_snapshot(doc)
    doc.status = DocumentStatus.ARCHIVED
    db.add(doc)
    db.add(
        AuditLog(
            study_id=doc.trial_id,
            entity_type="DOCUMENT",
            entity_id=doc.id,
            action="DOCUMENT_ARCHIVED",
            detail=_audit_detail(before, _doc_snapshot(doc)),
            operator_id=current_user.id,
            operator_role=await get_operator_role_label(db, doc.trial_id, current_user),
        )
    )
    await db.commit()
    await db.refresh(doc)
    return doc


async def get_version_download_response(
    db: AsyncSession,
    version_id: uuid.UUID,
    current_user,
) -> FileResponse:
    version = await version_crud.get(db, version_id)
    if not version:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="版本不存在")
    doc = await document_crud.get(db, version.document_id)
    if not doc:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="文档不存在")
    await _ensure_study_access(db, doc.trial_id, current_user, action="view")

    file_path = Path(version.file_uri)
    if not file_path.exists():
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="文件不存在")
    filename = file_path.name
    return FileResponse(
        path=str(file_path),
        filename=filename,
        media_type=version.mime_type or "application/octet-stream",
        headers={"Content-Disposition": f'inline; filename="{filename}"'},
    )


async def create_distributions(
    db: AsyncSession,
    version_id: uuid.UUID,
    payload: DistributionCreate,
    current_user,
) -> list[Distribution]:
    version = await version_crud.get(db, version_id)
    if not version:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="版本不存在")
    doc = await document_crud.get(db, version.document_id)
    if not doc:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="文档不存在")
    await _ensure_study_access(db, doc.trial_id, current_user, action="distribute")

    if version.status != DocumentVersionStatus.EFFECTIVE:
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="只有生效版本可分发")

    created: list[Distribution] = []
    for target in payload.targets:
        distribution = Distribution(
            document_id=version.document_id,
            version_id=version.id,
            target_type=target.target_type,
            target_id=target.target_id,
            due_at=payload.due_at,
            created_by=current_user.id,
        )
        db.add(distribution)
        created.append(distribution)
        db.add(
            AuditLog(
                study_id=doc.trial_id,
                entity_type="DISTRIBUTION",
                entity_id=distribution.id,
                action="DISTRIBUTION_CREATED",
                detail=_audit_detail(None, {"version_id": str(version.id), "target_id": target.target_id}),
                operator_id=current_user.id,
                operator_role=await get_operator_role_label(db, doc.trial_id, current_user),
            )
        )
    await db.commit()
    for distribution in created:
        await db.refresh(distribution)
    return created


async def list_distributions(
    db: AsyncSession,
    version_id: uuid.UUID,
    current_user,
) -> list[DistributionRead]:
    version = await version_crud.get(db, version_id)
    if not version:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="版本不存在")
    doc = await document_crud.get(db, version.document_id)
    if not doc:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="文档不存在")
    await _ensure_study_member(db, doc.trial_id, current_user)

    distributions = await distribution_crud.list_by_version(db, version_id)
    result: list[DistributionRead] = []
    for distribution in distributions:
        stats = await _stats_for_distribution(db, distribution.id, distribution.due_at)
        result.append(
            DistributionRead(
                id=distribution.id,
                document_id=distribution.document_id,
                version_id=distribution.version_id,
                target_type=distribution.target_type,
                target_id=distribution.target_id,
                status=distribution.status,
                due_at=distribution.due_at,
                created_by=distribution.created_by,
                created_at=distribution.created_at,
                stats=stats,
            )
        )
    return result


async def create_acknowledgement(
    db: AsyncSession,
    distribution_id: uuid.UUID,
    payload: AcknowledgementCreate,
    current_user,
) -> Acknowledgement:
    distribution = await distribution_crud.get(db, distribution_id)
    if not distribution:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="分发不存在")
    version = await version_crud.get(db, distribution.version_id)
    if not version:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="版本不存在")
    doc = await document_crud.get(db, version.document_id)
    if not doc:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="文档不存在")
    membership = await _ensure_study_access(db, doc.trial_id, current_user, action="ack")

    if distribution.target_type == DistributionTargetType.USER and distribution.target_id != str(current_user.id):
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="不在分发范围内")
    if distribution.target_type == DistributionTargetType.ROLE:
        if distribution.target_id not in ("ADMIN" if is_system_admin(current_user) else "", getattr(membership, "role_in_study", "")):
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="不在分发范围内")

    if payload.ack_type != AcknowledgementType.RECEIVED:
        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="仅支持已接收回执")

    existing = await acknowledgement_crud.get_existing(
        db,
        distribution_id=distribution_id,
        user_id=current_user.id,
        ack_type=payload.ack_type,
    )
    if existing:
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="回执已存在")

    ack = Acknowledgement(
        distribution_id=distribution_id,
        user_id=current_user.id,
        site_id=None,
        ack_type=payload.ack_type,
        due_at=distribution.due_at,
        acked_at=datetime.now(timezone.utc),
        note=payload.evidence,
    )
    db.add(ack)
    db.add(
        AuditLog(
            study_id=doc.trial_id,
            entity_type="ACKNOWLEDGEMENT",
            entity_id=ack.id,
            action="ACK_CREATED",
            detail=_audit_detail(None, {"distribution_id": str(distribution.id), "ack_type": payload.ack_type}),
            operator_id=current_user.id,
            operator_role=await get_operator_role_label(db, doc.trial_id, current_user),
        )
    )
    await db.commit()
    await db.refresh(ack)
    return ack


async def _stats_for_distribution(
    db: AsyncSession,
    distribution_id: uuid.UUID,
    due_at: datetime | None,
) -> DistributionStats:
    acks = await acknowledgement_crud.list_by_distribution(db, distribution_id)
    return _aggregate_ack_stats(acks, due_at, target_count=1)


async def _aggregate_distribution_stats(
    db: AsyncSession,
    version_id: uuid.UUID | None,
) -> DistributionStats:
    if not version_id:
        return DistributionStats()
    distributions = await distribution_crud.list_by_version(db, version_id)
    stats = DistributionStats()
    for distribution in distributions:
        acks = await acknowledgement_crud.list_by_distribution(db, distribution.id)
        dist_stats = _aggregate_ack_stats(acks, distribution.due_at, target_count=1)
        stats.total += dist_stats.total
        stats.received += dist_stats.received
        # only RECEIVED is supported
        stats.overdue += dist_stats.overdue
    return stats


def _aggregate_ack_stats(
    acks: Iterable[Acknowledgement],
    due_at: datetime | None,
    target_count: int,
) -> DistributionStats:
    stats = DistributionStats()
    stats.total = target_count
    for ack in acks:
        if ack.ack_type == AcknowledgementType.RECEIVED:
            stats.received += 1
    if due_at:
        now = datetime.now(timezone.utc)
        if due_at < now and not acks:
            stats.overdue = target_count
    return stats


async def list_distribution_notifications(
    db: AsyncSession,
    study_id: uuid.UUID,
    current_user,
    *,
    skip: int = 0,
    limit: int = 20,
) -> list[NotificationItem]:
    membership = None
    if not is_system_admin(current_user):
        membership = await member_crud.get_member(db, study_id, current_user.id)
    role_in_study = membership.role_in_study if membership else None

    target_filters = [
        (Distribution.target_type == DistributionTargetType.USER)
        & (Distribution.target_id == str(current_user.id))
    ]
    if role_in_study:
        target_filters.append(
            (Distribution.target_type == DistributionTargetType.ROLE)
            & (Distribution.target_id == role_in_study)
        )

    stmt = (
        select(
            Distribution.id,
            Distribution.document_id,
            Distribution.version_id,
            Distribution.created_at,
            Document.title,
            Document.doc_no,
            DocumentVersion.version_no,
            DocumentVersion.change_summary,
            DocumentVersion.effective_at,
            Study.name.label("study_name"),
            DesktopNotificationDelivery.delivered_at,
            DesktopNotificationDelivery.read_at,
        )
        .join(DocumentVersion, Distribution.version_id == DocumentVersion.id)
        .join(Document, Distribution.document_id == Document.id)
        .join(Study, Document.trial_id == Study.id)
        .outerjoin(
            DesktopNotificationDelivery,
            and_(
                DesktopNotificationDelivery.distribution_id == Distribution.id,
                DesktopNotificationDelivery.user_id == current_user.id,
            ),
        )
        .where(
            Document.trial_id == study_id,
            Distribution.status == DistributionStatus.ACTIVE,
            or_(*target_filters),
        )
        .order_by(Distribution.created_at.desc())
        .offset(skip)
        .limit(limit)
    )
    result = await db.execute(stmt)
    items: list[NotificationItem] = []
    for row in result.fetchall():
        items.append(
            NotificationItem(
                id=row.id,
                document_id=row.document_id,
                version_id=row.version_id,
                document_title=row.title,
                document_no=row.doc_no,
                version_no=row.version_no,
                change_summary=row.change_summary,
                effective_at=row.effective_at,
                created_at=row.created_at,
                study_id=study_id,
                study_name=row.study_name,
                delivered_at=row.delivered_at,
                read_at=row.read_at,
            )
        )
    return items