document-management/backend/dms/services/attachment_binding_service.py

"""B6共享附件挂载和解除挂载事务。"""

from __future__ import annotations

from datetime import datetime, timezone
from typing import Any

from sqlalchemy import func, select
from sqlalchemy.exc import IntegrityError

from dms.common.enums import AuditAction, AuditTarget, DocumentType
from dms.common.errors import (
    ConflictError,
    InvalidArgumentError,
    ResourceNotFoundError,
)
from dms.common.response import serialize_id
from dms.database.transaction import execute_with_deadlock_retry, transaction
from dms.extensions import db
from dms.models import AttachmentBinding, Document
from dms.security.auth_context import get_auth_context
from dms.services.audit_service import business_audit
from dms.services.authorization_service import require_security_clearance

BIND_FIELDS = {"attachmentIds", "mainPlanRowVersion"}


def _now() -> datetime:
    return datetime.now(timezone.utc).replace(tzinfo=None)


def _version(value: Any, name: str) -> int:
    if type(value) is not int or value < 0:
        raise InvalidArgumentError(f"{name}必须是非负整数")
    return value


def _string_id(value: Any, name: str) -> int:
    if not isinstance(value, str) or not value.isdecimal() or int(value) <= 0:
        raise InvalidArgumentError(f"{name}必须是正整数形式的字符串ID")
    return int(value)


def _bind_payload(payload: Any) -> tuple[list[int], int]:
    if not isinstance(payload, dict) or set(payload) != BIND_FIELDS:
        raise InvalidArgumentError(
            "请求字段必须且只能包含attachmentIds和mainPlanRowVersion"
        )
    raw_ids = payload["attachmentIds"]
    if not isinstance(raw_ids, list) or not raw_ids:
        raise InvalidArgumentError("attachmentIds必须是非空数组")
    attachment_ids: list[int] = []
    seen: set[int] = set()
    for index, raw_id in enumerate(raw_ids):
        attachment_id = _string_id(raw_id, f"attachmentIds[{index}]")
        if attachment_id not in seen:
            attachment_ids.append(attachment_id)
            seen.add(attachment_id)
    return attachment_ids, _version(
        payload["mainPlanRowVersion"], "mainPlanRowVersion"
    )


def _locked_main(session, main_document_id: int) -> Document:
    main = session.scalar(
        select(Document)
        .where(
            Document.id == main_document_id,
            Document.is_deleted.is_(False),
        )
        .with_for_update()
    )
    if main is None or main.document_type != DocumentType.MAIN.value:
        raise ResourceNotFoundError("主案不存在或类型不匹配")
    require_security_clearance(main, get_auth_context())
    return main


def _check_version(main: Document, expected: int) -> None:
    if main.row_version != expected:
        raise ConflictError(
            "数据已被其他用户修改，请刷新后重试",
            details={"currentRowVersion": main.row_version},
        )


def _locked_attachments(session, attachment_ids: list[int]) -> dict[int, Document]:
    documents = session.scalars(
        select(Document)
        .where(Document.id.in_(sorted(attachment_ids)))
        .order_by(Document.id.asc())
        .with_for_update()
    ).all()
    by_id = {document.id: document for document in documents}
    if any(
        attachment_id not in by_id
        or by_id[attachment_id].is_deleted
        or by_id[attachment_id].document_type != DocumentType.ATTACHMENT.value
        for attachment_id in attachment_ids
    ):
        raise ResourceNotFoundError("共享附件不存在或类型不匹配")
    return by_id


def _active_count(session, main_document_id: int) -> int:
    return (
        session.scalar(
            select(func.count(AttachmentBinding.id)).where(
                AttachmentBinding.main_document_id == main_document_id,
                AttachmentBinding.is_deleted.is_(False),
            )
        )
        or 0
    )


def _current_version(main_document_id: int) -> int | None:
    return db.session.scalar(
        select(Document.row_version).where(
            Document.id == main_document_id,
            Document.is_deleted.is_(False),
        )
    )


def bind_attachments(main_document_id: int, payload: Any) -> dict[str, int]:
    """去重后批量挂载；同一主案上的写入由主案行锁串行化。"""
    attachment_ids, expected_version = _bind_payload(payload)
    context = get_auth_context()

    def operation() -> dict[str, int]:
        with transaction() as session:
            main = _locked_main(session, main_document_id)
            _check_version(main, expected_version)
            _locked_attachments(session, attachment_ids)

            bindings = session.scalars(
                select(AttachmentBinding)
                .where(AttachmentBinding.main_document_id == main.id)
                .order_by(AttachmentBinding.id.asc())
                .with_for_update()
            ).all()
            active_by_attachment = {
                binding.attachment_document_id: binding
                for binding in bindings
                if not binding.is_deleted
            }
            existing_ids = [
                attachment_id
                for attachment_id in attachment_ids
                if attachment_id in active_by_attachment
            ]
            created_ids = [
                attachment_id
                for attachment_id in attachment_ids
                if attachment_id not in active_by_attachment
            ]
            if not created_ids:
                return {
                    "createdCount": 0,
                    "existingCount": len(existing_ids),
                    "attachmentCount": _active_count(session, main.id),
                    "mainPlanRowVersion": main.row_version,
                }

            max_sort_no = max(
                (
                    binding.sort_no
                    for binding in bindings
                    if not binding.is_deleted
                ),
                default=0,
            )
            for offset, attachment_id in enumerate(created_ids, start=1):
                session.add(
                    AttachmentBinding(
                        main_document_id=main.id,
                        attachment_document_id=attachment_id,
                        sort_no=max_sort_no + offset * 10,
                        created_by=context.user_id,
                        updated_by=context.user_id,
                    )
                )
            session.flush()
            main.attachment_count = _active_count(session, main.id)
            main.row_version += 1
            main.updated_by = context.user_id
            main.updated_by_name = context.real_name
            main.updated_at = _now()
            session.add(
                business_audit(
                    action=AuditAction.BIND_ATTACHMENT,
                    target=AuditTarget.ATTACHMENT_BINDING,
                    target_id=main.id,
                    target_name=main.document_name,
                    detail={
                        "mainPlanId": serialize_id(main.id),
                        "requestedAttachmentIds": [
                            serialize_id(value) for value in attachment_ids
                        ],
                        "createdAttachmentIds": [
                            serialize_id(value) for value in created_ids
                        ],
                        "existingAttachmentIds": [
                            serialize_id(value) for value in existing_ids
                        ],
                        "createdCount": len(created_ids),
                        "existingCount": len(existing_ids),
                        "attachmentCount": main.attachment_count,
                        "mainPlanRowVersion": main.row_version,
                    },
                )
            )
            return {
                "createdCount": len(created_ids),
                "existingCount": len(existing_ids),
                "attachmentCount": main.attachment_count,
                "mainPlanRowVersion": main.row_version,
            }

    try:
        return execute_with_deadlock_retry(operation)
    except IntegrityError as exc:
        db.session.rollback()
        raise ConflictError(
            "挂载关系已被并发修改，请刷新后重试",
            details={"currentRowVersion": _current_version(main_document_id)},
        ) from exc


def unbind_attachment(
    main_document_id: int,
    attachment_document_id: int,
    expected_version: int,
) -> dict[str, int]:
    """逻辑删除一条有效挂载关系并重新计算主案计数。"""
    expected_version = _version(expected_version, "mainPlanRowVersion")
    context = get_auth_context()

    def operation() -> dict[str, int]:
        with transaction() as session:
            main = _locked_main(session, main_document_id)
            _check_version(main, expected_version)
            _locked_attachments(session, [attachment_document_id])
            binding = session.scalar(
                select(AttachmentBinding)
                .where(
                    AttachmentBinding.main_document_id == main.id,
                    AttachmentBinding.attachment_document_id
                    == attachment_document_id,
                    AttachmentBinding.is_deleted.is_(False),
                )
                .with_for_update()
            )
            if binding is None:
                raise ResourceNotFoundError("有效挂载关系不存在")

            now = _now()
            binding.is_deleted = True
            binding.deleted_at = now
            binding.updated_at = now
            binding.updated_by = context.user_id
            binding.row_version += 1
            session.flush()
            main.attachment_count = _active_count(session, main.id)
            main.row_version += 1
            main.updated_by = context.user_id
            main.updated_by_name = context.real_name
            main.updated_at = now
            session.add(
                business_audit(
                    action=AuditAction.UNBIND_ATTACHMENT,
                    target=AuditTarget.ATTACHMENT_BINDING,
                    target_id=binding.id,
                    target_name=main.document_name,
                    detail={
                        "bindingId": serialize_id(binding.id),
                        "mainPlanId": serialize_id(main.id),
                        "attachmentId": serialize_id(attachment_document_id),
                        "attachmentCount": main.attachment_count,
                        "mainPlanRowVersion": main.row_version,
                    },
                )
            )
            return {
                "attachmentCount": main.attachment_count,
                "mainPlanRowVersion": main.row_version,
            }

    return execute_with_deadlock_retry(operation)


__all__ = ["bind_attachments", "unbind_attachment"]