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

"""子案跨主案挂载与解除挂载事务。

与附件挂载（attachment_binding_service）同范式：主案行锁 + 密码动作 +
乐观锁 + 步长10排序 + 审计（复用 BIND_ATTACHMENT / UNBIND_ATTACHMENT
动作，detail 标注 scope=subPlanMount，不扩展 sys_audit_log 的 CHECK）。
子案权限仍继承其原主案（root_document_id 不变），挂载仅建立展示关系。
"""

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 Document, SubPlanBinding
from dms.security.auth_context import get_auth_context
from dms.services.audit_service import business_audit
from dms.common.enums import AllowedAction
from dms.services.authorization_service import (
    require_plan_action,
    require_security_clearance,
)
from dms.services.document_password_service import (
    DocumentPasswordAction,
    get_request_password,
    require_document_password,
)

BIND_SUB_FIELDS = {"subPlanIds", "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_SUB_FIELDS:
        raise InvalidArgumentError(
            "请求字段必须且只能包含subPlanIds和mainPlanRowVersion"
        )
    raw_ids = payload["subPlanIds"]
    if not isinstance(raw_ids, list) or not raw_ids:
        raise InvalidArgumentError("subPlanIds必须是非空数组")
    sub_plan_ids: list[int] = []
    seen: set[int] = set()
    for index, raw_id in enumerate(raw_ids):
        sub_plan_id = _string_id(raw_id, f"subPlanIds[{index}]")
        if sub_plan_id not in seen:
            sub_plan_ids.append(sub_plan_id)
            seen.add(sub_plan_id)
    return sub_plan_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 _locked_sub_plans(
    session,
    main_id: int,
    sub_plan_ids: list[int],
) -> dict[int, Document]:
    """校验目标均为有效子案；属于本主案的自有子案直接剔除（无需挂载）。"""
    documents = session.scalars(
        select(Document)
        .where(Document.id.in_(sorted(sub_plan_ids)))
        .order_by(Document.id.asc())
        .with_for_update()
    ).all()
    by_id = {document.id: document for document in documents}
    if any(
        sub_plan_id not in by_id
        or by_id[sub_plan_id].is_deleted
        or by_id[sub_plan_id].document_type != DocumentType.SUB_PLAN.value
        for sub_plan_id in sub_plan_ids
    ):
        raise ResourceNotFoundError("子案不存在或类型不匹配")
    return {
        sub_plan_id: by_id[sub_plan_id]
        for sub_plan_id in sub_plan_ids
        if by_id[sub_plan_id].parent_document_id != main_id
    }


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_sub_plans(main_document_id: int, payload: Any) -> dict[str, int]:
    """去重后批量挂载其他主案的子案；同一主案上的写入由主案行锁串行化。"""
    sub_plan_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)
            require_plan_action(main, context, AllowedAction.BIND_ATTACHMENT)
            require_document_password(
                main,
                DocumentPasswordAction.BIND_ATTACHMENT.value,
                get_request_password(),
                context,
            )
            if main.row_version != expected_version:
                raise ConflictError(
                    "数据已被其他用户修改，请刷新后重试",
                    details={"currentRowVersion": main.row_version},
                )
            mountable_by_id = _locked_sub_plans(session, main.id, sub_plan_ids)

            bindings = session.scalars(
                select(SubPlanBinding)
                .where(SubPlanBinding.main_document_id == main.id)
                .order_by(SubPlanBinding.id.asc())
                .with_for_update()
            ).all()
            active_by_sub = {
                binding.sub_document_id: binding
                for binding in bindings
                if not binding.is_deleted
            }
            existing_ids = [
                sub_plan_id
                for sub_plan_id in mountable_by_id
                if sub_plan_id in active_by_sub
            ]
            created_ids = [
                sub_plan_id
                for sub_plan_id in mountable_by_id
                if sub_plan_id not in active_by_sub
            ]
            if not created_ids:
                return {
                    "createdCount": 0,
                    "existingCount": len(existing_ids),
                    "subPlanCount": len(active_by_sub),
                    "mainPlanRowVersion": main.row_version,
                }

            max_sort_no = max(
                (
                    binding.sort_no
                    for binding in bindings
                    if not binding.is_deleted
                ),
                default=0,
            )
            for offset, sub_plan_id in enumerate(created_ids, start=1):
                session.add(
                    SubPlanBinding(
                        main_document_id=main.id,
                        sub_document_id=sub_plan_id,
                        sort_no=max_sort_no + offset * 10,
                        created_by=context.user_id,
                        updated_by=context.user_id,
                    )
                )
            session.flush()
            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={
                        "scope": "subPlanMount",
                        "mainPlanId": serialize_id(main.id),
                        "requestedSubPlanIds": [
                            serialize_id(value) for value in sub_plan_ids
                        ],
                        "createdSubPlanIds": [
                            serialize_id(value) for value in created_ids
                        ],
                        "existingSubPlanIds": [
                            serialize_id(value) for value in existing_ids
                        ],
                        "createdCount": len(created_ids),
                        "existingCount": len(existing_ids),
                        "mainPlanRowVersion": main.row_version,
                    },
                )
            )
            return {
                "createdCount": len(created_ids),
                "existingCount": len(existing_ids),
                "subPlanCount": len(active_by_sub) + len(created_ids),
                "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_sub_plan(
    main_document_id: int,
    sub_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)
            require_plan_action(main, context, AllowedAction.UNBIND_ATTACHMENT)
            require_document_password(
                main,
                DocumentPasswordAction.UNBIND_ATTACHMENT.value,
                get_request_password(),
                context,
            )
            if main.row_version != expected_version:
                raise ConflictError(
                    "数据已被其他用户修改，请刷新后重试",
                    details={"currentRowVersion": main.row_version},
                )
            binding = session.scalar(
                select(SubPlanBinding)
                .where(
                    SubPlanBinding.main_document_id == main.id,
                    SubPlanBinding.sub_document_id == sub_document_id,
                    SubPlanBinding.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
            remaining = (
                session.scalar(
                    select(func.count(SubPlanBinding.id)).where(
                        SubPlanBinding.main_document_id == main.id,
                        SubPlanBinding.is_deleted.is_(False),
                    )
                )
                or 0
            )
            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={
                        "scope": "subPlanMount",
                        "bindingId": serialize_id(binding.id),
                        "mainPlanId": serialize_id(main.id),
                        "subPlanId": serialize_id(sub_document_id),
                        "subPlanCount": remaining,
                        "mainPlanRowVersion": main.row_version,
                    },
                )
            )
            return {
                "subPlanCount": remaining,
                "mainPlanRowVersion": main.row_version,
            }

    return execute_with_deadlock_retry(operation)


__all__ = [
    "bind_sub_plans",
    "unbind_sub_plan",
]