ctms/backend/app/services/document_service.py

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 update as sa_update
from sqlalchemy.ext.asyncio import AsyncSession

from app.core import rbac
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 document_version as version_crud
from app.crud import member as member_crud
from app.crud import study as study_crud
from app.crud import user as user_crud
from app.crud import version_workflow as workflow_crud
from app.crud import workflow_template as template_crud
from app.models.acknowledgement import Acknowledgement, AcknowledgementType
from app.models.audit_log import AuditLog
from app.models.distribution import Distribution, DistributionTargetType
from app.models.document import Document, DocumentScopeType, DocumentStatus
from app.models.document_version import DocumentVersion, DocumentVersionStatus
from app.models.version_workflow import VersionWorkflow, WorkflowStatus
from app.models.workflow_action import WorkflowAction, WorkflowActionType
from app.models.workflow_template import WorkflowTemplate
from app.schemas.acknowledgement import AcknowledgementCreate
from app.schemas.distribution import DistributionCreate, DistributionRead, DistributionStats
from app.schemas.document import DocumentCreate, DocumentDetail, DocumentSummary
from app.schemas.document_version import DocumentVersionRead, DocumentVersionSummary
from app.schemas.user import UserDisplay

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


def _role_value(user) -> str:
    return user.role.value if hasattr(user.role, "value") else str(user.role)


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:
    return {
        "id": str(doc.id),
        "trial_id": str(doc.trial_id),
        "site_id": str(doc.site_id) if doc.site_id else None,
        "doc_no": doc.doc_no,
        "status": str(doc.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:
    return {
        "id": str(version.id),
        "document_id": str(version.document_id),
        "version_no": version.version_no,
        "status": str(version.status),
        "file_hash": version.file_hash,
    }


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 current_user.role == "ADMIN":
        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="不是项目成员")
    if not rbac.is_allowed(action, current_user, membership):
        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")
    if doc_in.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 doc_in.scope_type == DocumentScopeType.SITE else None
    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,
        doc_no=doc_in.doc_no,
        doc_type=doc_in.doc_type,
        title=doc_in.title,
        scope_type=doc_in.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=_role_value(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,
    skip: int,
    limit: int,
    current_user,
) -> list[DocumentSummary]:
    await _ensure_study_access(db, trial_id, current_user, action="view")
    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,
        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,
                doc_no=doc.doc_no,
                doc_type=doc.doc_type,
                title=doc.title,
                scope_type=doc.scope_type,
                owner_id=doc.owner_id,
                status=doc.status,
                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")
    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)
    return DocumentDetail(
        id=doc.id,
        trial_id=doc.trial_id,
        site_id=doc.site_id,
        doc_no=doc.doc_no,
        doc_type=doc.doc_type,
        title=doc.title,
        scope_type=doc.scope_type,
        owner_id=doc.owner_id,
        status=doc.status,
        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=[DocumentVersionRead.model_validate(v) for v in versions],
        distribution_stats=distribution_stats,
    )


async def create_version(
    db: AsyncSession,
    document_id: uuid.UUID,
    *,
    version_no: 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")

    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)

    version = DocumentVersion(
        document_id=document_id,
        version_no=version_no,
        parent_version_id=parent_version_id,
        status=DocumentVersionStatus.DRAFT,
        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)
    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=_role_value(current_user),
        )
    )
    await db.commit()
    await db.refresh(version)
    return version


async def submit_version(
    db: AsyncSession,
    version_id: uuid.UUID,
    template_id: uuid.UUID,
    comment: str | None,
    current_user,
) -> VersionWorkflow:
    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="submit")

    if version.status != DocumentVersionStatus.DRAFT:
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="版本状态不允许提交")
    pending = await version_crud.list_pending_versions(db, version.document_id)
    if any(v.id != version.id for v in pending):
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="存在待审批版本")

    template = await template_crud.get(db, template_id)
    if not template:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="审批模板不存在")

    before = _version_snapshot(version)
    version.status = DocumentVersionStatus.SUBMITTED
    version.submitted_at = datetime.now(timezone.utc)

    workflow = VersionWorkflow(
        document_id=version.document_id,
        version_id=version.id,
        template_id=template_id,
        status=WorkflowStatus.PENDING,
        current_node=1,
        submitted_by=current_user.id,
        submitted_at=version.submitted_at,
    )
    action = WorkflowAction(
        workflow=workflow,
        node_order=workflow.current_node,
        action=WorkflowActionType.SUBMIT,
        actor_id=current_user.id,
        comment=comment,
    )
    db.add(version)
    db.add(workflow)
    db.add(action)
    db.add(
        AuditLog(
            study_id=doc.trial_id,
            entity_type="DOCUMENT_VERSION",
            entity_id=version.id,
            action="VERSION_SUBMITTED",
            detail=_audit_detail(before, _version_snapshot(version)),
            operator_id=current_user.id,
            operator_role=_role_value(current_user),
        )
    )
    await db.commit()
    await db.refresh(workflow)
    return workflow


async def approve_version(
    db: AsyncSession,
    version_id: uuid.UUID,
    comment: str | None,
    current_user,
) -> VersionWorkflow:
    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="approve")

    workflow = await workflow_crud.get_by_version(db, version_id)
    if not workflow or workflow.status != WorkflowStatus.PENDING:
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="审批流程不可用")

    before = _version_snapshot(version)
    version.status = DocumentVersionStatus.APPROVED
    version.approved_at = datetime.now(timezone.utc)
    workflow.status = WorkflowStatus.APPROVED
    workflow.completed_at = version.approved_at

    db.add(version)
    db.add(workflow)
    db.add(
        WorkflowAction(
            workflow_id=workflow.id,
            node_order=workflow.current_node,
            action=WorkflowActionType.APPROVE,
            actor_id=current_user.id,
            comment=comment,
        )
    )
    db.add(
        AuditLog(
            study_id=doc.trial_id,
            entity_type="DOCUMENT_VERSION",
            entity_id=version.id,
            action="VERSION_APPROVED",
            detail=_audit_detail(before, _version_snapshot(version)),
            operator_id=current_user.id,
            operator_role=_role_value(current_user),
        )
    )
    await db.commit()
    await db.refresh(workflow)
    return workflow


async def reject_version(
    db: AsyncSession,
    version_id: uuid.UUID,
    comment: str | None,
    current_user,
) -> VersionWorkflow:
    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="approve")

    workflow = await workflow_crud.get_by_version(db, version_id)
    if not workflow or workflow.status != WorkflowStatus.PENDING:
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="审批流程不可用")

    before = _version_snapshot(version)
    version.status = DocumentVersionStatus.WITHDRAWN
    version.withdrawn_at = datetime.now(timezone.utc)
    workflow.status = WorkflowStatus.REJECTED
    workflow.completed_at = version.withdrawn_at

    db.add(version)
    db.add(workflow)
    db.add(
        WorkflowAction(
            workflow_id=workflow.id,
            node_order=workflow.current_node,
            action=WorkflowActionType.REJECT,
            actor_id=current_user.id,
            comment=comment,
        )
    )
    db.add(
        AuditLog(
            study_id=doc.trial_id,
            entity_type="DOCUMENT_VERSION",
            entity_id=version.id,
            action="VERSION_REJECTED",
            detail=_audit_detail(before, _version_snapshot(version)),
            operator_id=current_user.id,
            operator_role=_role_value(current_user),
        )
    )
    await db.commit()
    await db.refresh(workflow)
    return workflow


async def make_version_effective(
    db: AsyncSession,
    version_id: uuid.UUID,
    current_user,
) -> DocumentVersion:
    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="approve")

    if version.status != DocumentVersionStatus.APPROVED:
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="版本未审批通过")

    now = datetime.now(timezone.utc)
    before = _version_snapshot(version)
    version.status = DocumentVersionStatus.EFFECTIVE
    version.effective_at = now

    await db.execute(
        sa_update(DocumentVersion)
        .where(
            DocumentVersion.document_id == version.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 == version.document_id)
        .values(current_effective_version_id=version.id)
    )
    db.add(version)
    db.add(
        AuditLog(
            study_id=doc.trial_id,
            entity_type="DOCUMENT_VERSION",
            entity_id=version.id,
            action="VERSION_EFFECTIVE",
            detail=_audit_detail(before, _version_snapshot(version)),
            operator_id=current_user.id,
            operator_role=_role_value(current_user),
        )
    )
    await db.commit()
    await db.refresh(version)
    return version


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=_role_value(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 list_workflow_templates(
    db: AsyncSession,
    trial_id: uuid.UUID | None,
    current_user,
) -> list[WorkflowTemplate]:
    if trial_id:
        await _ensure_study_access(db, trial_id, current_user, action="view")
    elif current_user.role != "ADMIN":
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="权限不足")
    templates = await template_crud.list_active(db, trial_id)
    return list(templates)


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=_role_value(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_access(db, doc.trial_id, current_user, action="view")

    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:
        role_value = _role_value(current_user)
        if distribution.target_id not in (role_value, 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=_role_value(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