from __future__ import annotations

import io
import json
import threading
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timezone
from pathlib import Path

import pytest


PDF_BYTES = b"%PDF-1.4\n1 0 obj\n<<>>\nendobj\n%%EOF\n"


def _headers(token: str) -> dict[str, str]:
    return {"Authorization": f"Bearer {token}"}


def _category_id(app) -> str:
    from dms.extensions import db
    from dms.models import Category

    with app.app_context():
        category = db.session.scalar(
            db.select(Category).where(Category.category_code == "CHILD_A1")
        )
        assert category is not None
        return str(category.id)


def _upload_plan(
    client,
    token: str,
    category_id: str,
    name: str,
    *,
    document_type: str = "MAIN",
    parent_id: str | None = None,
    visibility: str = "ALL_AUTHENTICATED",
):
    metadata = {
        "documentName": name,
        "documentType": document_type,
        "summary": f"{name}摘要",
        "securityLevel": "INTERNAL",
        "tags": ["B8"],
    }
    if document_type == "MAIN":
        metadata["categoryId"] = category_id
    if document_type == "SUB_PLAN" and parent_id is not None:
        metadata["parentDocumentId"] = parent_id
    response = client.post(
        "/api/v1/documents",
        data={
            "file": (io.BytesIO(PDF_BYTES), f"{name}.pdf"),
            "metadata": json.dumps(metadata, ensure_ascii=False),
        },
        headers=_headers(token),
        content_type="multipart/form-data",
    )
    assert response.status_code == 201, response.get_json()
    document = response.get_json()["data"]
    if document_type == "MAIN" and visibility == "ALL_AUTHENTICATED":
        permission = client.put(
            f"/api/v1/documents/{document['id']}/permissions",
            json={
                "visibilityType": "ALL_AUTHENTICATED",
                "documentRowVersion": document["rowVersion"],
                "entries": [],
            },
            headers=_headers(token),
        )
        assert permission.status_code == 200, permission.get_json()
        document["rowVersion"] = permission.get_json()["data"][
            "documentRowVersion"
        ]
        document["visibilityType"] = "ALL_AUTHENTICATED"
    return document


def _upload_attachment(client, token: str, name: str):
    response = client.post(
        "/api/v1/attachments",
        data={
            "file": (io.BytesIO(PDF_BYTES), f"{name}.pdf"),
            "metadata": json.dumps(
                {
                    "documentName": name,
                    "attachmentType": "OTHER",
                    "summary": f"{name}摘要",
                    "tags": ["B8"],
                },
                ensure_ascii=False,
            ),
        },
        headers=_headers(token),
        content_type="multipart/form-data",
    )
    assert response.status_code == 201, response.get_json()
    return response.get_json()["data"]


def _delete(client, token: str, document: dict, *, attachment: bool = False):
    prefix = "/api/v1/attachments" if attachment else "/api/v1/documents"
    response = client.delete(
        f"{prefix}/{document['id']}?rowVersion={document['rowVersion']}",
        headers=_headers(token),
    )
    assert response.status_code == 200, response.get_json()
    return response.get_json()["data"]


def _deleted_document(app, document_id: str):
    from dms.extensions import db
    from dms.models import Document

    with app.app_context():
        document = db.session.get(Document, int(document_id))
        assert document is not None and document.is_deleted
        return {
            "rowVersion": document.row_version,
            "deletedAt": document.deleted_at,
            "fileRelativePath": document.file_relative_path,
            "viewCount": document.view_count,
        }


def _restore(client, token: str, document_id: str, row_version: int):
    return client.post(
        f"/api/v1/recycle-bin/documents/{document_id}/restore",
        json={"rowVersion": row_version},
        headers=_headers(token),
    )


def test_recycle_bin_auth_query_filters_and_deleted_by(
    b3_app, b3_client, login_token, token_for
):
    category_id = _category_id(b3_app)
    normal = _upload_plan(
        b3_client, login_token, category_id, "B8正常文档"
    )
    deleted = _upload_plan(
        b3_client, login_token, category_id, "B8_百分号%_下划线_"
    )
    _delete(b3_client, login_token, deleted)

    assert b3_client.get("/api/v1/recycle-bin/documents").status_code == 401
    for username in ("user",):
        response = b3_client.get(
            "/api/v1/recycle-bin/documents",
            headers=_headers(token_for(username)),
        )
        assert response.status_code == 403
        assert response.get_json()["code"] == "FORBIDDEN"

    response = b3_client.get(
        "/api/v1/recycle-bin/documents",
        query_string={
            "keyword": "百分号%",
            "documentType": "MAIN",
            "categoryId": category_id,
            "page": "1",
            "pageSize": "20",
            "sortField": "documentName",
            "sortDirection": "asc",
        },
        headers=_headers(login_token),
    )
    assert response.status_code == 200, response.get_json()
    body = response.get_json()
    assert body["data"]["total"] == 1
    item = body["data"]["items"][0]
    assert item["id"] == deleted["id"]
    assert item["id"] != normal["id"]
    assert item["documentStatus"] == "PUBLISHED"
    assert item["deletedBy"]["username"] == "admin"
    assert "fileRelativePath" not in item
    assert "fileHash" not in item
    assert isinstance(item["id"], str)


@pytest.mark.parametrize(
    ("query", "message_part"),
    [
        ("unknown=x", "未知"),
        ("page=1&page=2", "重复"),
        ("documentType=OTHER", "documentType"),
        ("categoryId=0", "categoryId"),
        ("deletedFrom=2026-01-01T00:00:00%2B08:00", "UTC Z"),
        ("deletedFrom=2026-01-02T00:00:00Z&deletedTo=2026-01-01T00:00:00Z", "早于"),
        ("pageSize=101", "100"),
        ("sortField=id", "sortField"),
        ("sortDirection=sideways", "sortDirection"),
    ],
)
def test_recycle_bin_rejects_invalid_query(
    b3_client, login_token, query, message_part
):
    response = b3_client.get(
        f"/api/v1/recycle-bin/documents?{query}",
        headers=_headers(login_token),
    )
    assert response.status_code == 400
    assert response.get_json()["code"] == "INVALID_ARGUMENT"
    assert message_part in response.get_json()["message"]


def test_restore_main_restores_delete_batch_acl_binding_counts_and_audit(
    b3_app, b3_client, login_token
):
    from dms.extensions import db
    from dms.models import (
        AttachmentBinding,
        AuditLog,
        Category,
        Document,
        Organization,
        Permission,
    )

    category_id = _category_id(b3_app)
    main = _upload_plan(
        b3_client,
        login_token,
        category_id,
        "B8关系主案",
        visibility="ORGANIZATION",
    )
    attachment = _upload_attachment(b3_client, login_token, "B8关系附件")
    with b3_app.app_context():
        organization = db.session.scalar(
            db.select(Organization).where(Organization.org_code == "ORG_ROOT")
        )
        assert organization is not None
        organization_id = str(organization.id)

    permission_response = b3_client.put(
        f"/api/v1/documents/{main['id']}/permissions",
        json={
            "visibilityType": "ORGANIZATION",
            "documentRowVersion": main["rowVersion"],
            "entries": [
                {
                    "subjectType": "ORG",
                    "subjectId": organization_id,
                    "canView": True,
                    "canDownload": True,
                    "canEdit": False,
                    "canManagePermission": False,
                    "canDelete": False,
                }
            ],
        },
        headers=_headers(login_token),
    )
    assert permission_response.status_code == 200, permission_response.get_json()
    main_version = permission_response.get_json()["data"]["documentRowVersion"]
    bind_response = b3_client.post(
        f"/api/v1/main-plans/{main['id']}/attachments/bind",
        json={
            "attachmentIds": [attachment["id"]],
            "mainPlanRowVersion": main_version,
        },
        headers=_headers(login_token),
    )
    assert bind_response.status_code == 200, bind_response.get_json()
    main["rowVersion"] = bind_response.get_json()["data"]["mainPlanRowVersion"]
    _delete(b3_client, login_token, main)
    deleted = _deleted_document(b3_app, main["id"])

    response = _restore(
        b3_client, login_token, main["id"], deleted["rowVersion"]
    )
    assert response.status_code == 200, response.get_json()
    data = response.get_json()["data"]
    assert data["restoredPermissionCount"] == 1
    assert data["skippedPermissionCount"] == 0
    assert data["restoredBindingCount"] == 1
    assert data["skippedBindingCount"] == 0
    assert data["document"]["attachmentCount"] == 1
    assert data["document"]["viewCount"] == deleted["viewCount"]
    assert "fileHash" not in data["document"]

    with b3_app.app_context():
        document = db.session.get(Document, int(main["id"]))
        category = db.session.get(Category, int(category_id))
        permission = db.session.scalar(
            db.select(Permission).where(
                Permission.document_id == int(main["id"]),
                Permission.is_deleted.is_(False),
            )
        )
        binding = db.session.scalar(
            db.select(AttachmentBinding).where(
                AttachmentBinding.main_document_id == int(main["id"]),
                AttachmentBinding.is_deleted.is_(False),
            )
        )
        audit = db.session.scalar(
            db.select(AuditLog)
            .where(
                AuditLog.action_type == "RESTORE_DOCUMENT",
                AuditLog.target_id == int(main["id"]),
            )
            .order_by(AuditLog.id.desc())
        )
        view_audits = db.session.scalar(
            db.select(db.func.count(AuditLog.id)).where(
                AuditLog.action_type == "VIEW_DOCUMENT",
                AuditLog.target_id == int(main["id"]),
            )
        )
        assert document is not None and not document.is_deleted
        assert document.row_version == deleted["rowVersion"] + 1
        assert category is not None and category.document_count >= 1
        assert permission is not None and permission.row_version == 2
        assert binding is not None and binding.row_version == 2
        assert audit is not None
        assert audit.operation_detail["restoredPermissionCount"] == 1
        assert "fileHash" not in audit.operation_detail
        assert view_audits == 0


def test_restore_sub_plan_recounts_parent_and_inherits_current_acl(
    b3_app, b3_client, login_token
):
    from dms.extensions import db
    from dms.models import Document, Permission

    category_id = _category_id(b3_app)
    main = _upload_plan(
        b3_client, login_token, category_id, "B8子案父主案"
    )
    sub = _upload_plan(
        b3_client,
        login_token,
        category_id,
        "B8待恢复子案",
        document_type="SUB_PLAN",
        parent_id=main["id"],
    )
    _delete(b3_client, login_token, sub)
    deleted = _deleted_document(b3_app, sub["id"])
    with b3_app.app_context():
        parent_before = db.session.get(Document, int(main["id"])).row_version

    response = _restore(
        b3_client, login_token, sub["id"], deleted["rowVersion"]
    )
    assert response.status_code == 200, response.get_json()
    data = response.get_json()["data"]
    assert data["restoredPermissionCount"] == 0
    with b3_app.app_context():
        restored = db.session.get(Document, int(sub["id"]))
        parent = db.session.get(Document, int(main["id"]))
        acl_count = db.session.scalar(
            db.select(db.func.count(Permission.id)).where(
                Permission.document_id == restored.id,
                Permission.is_deleted.is_(False),
            )
        )
        assert restored.row_version == deleted["rowVersion"] + 1
        assert parent.child_count == 1
        assert parent.row_version == parent_before + 1
        assert acl_count == 0


def test_restore_attachment_does_not_restore_historical_binding(
    b3_app, b3_client, login_token
):
    from dms.extensions import db
    from dms.models import AttachmentBinding, Document

    category_id = _category_id(b3_app)
    main = _upload_plan(
        b3_client, login_token, category_id, "B8附件挂载主案"
    )
    attachment = _upload_attachment(b3_client, login_token, "B8待恢复附件")
    bind = b3_client.post(
        f"/api/v1/main-plans/{main['id']}/attachments/bind",
        json={
            "attachmentIds": [attachment["id"]],
            "mainPlanRowVersion": main["rowVersion"],
        },
        headers=_headers(login_token),
    ).get_json()["data"]
    unbind = b3_client.delete(
        f"/api/v1/main-plans/{main['id']}/attachments/{attachment['id']}"
        f"?mainPlanRowVersion={bind['mainPlanRowVersion']}",
        headers=_headers(login_token),
    )
    assert unbind.status_code == 200
    _delete(b3_client, login_token, attachment, attachment=True)
    deleted = _deleted_document(b3_app, attachment["id"])
    response = _restore(
        b3_client, login_token, attachment["id"], deleted["rowVersion"]
    )
    assert response.status_code == 200, response.get_json()
    data = response.get_json()["data"]
    assert data["restoredPermissionCount"] == 0
    assert data["restoredBindingCount"] == 0
    assert data["document"]["mountedPlanCount"] == 0
    with b3_app.app_context():
        restored = db.session.get(Document, int(attachment["id"]))
        binding = db.session.scalar(
            db.select(AttachmentBinding).where(
                AttachmentBinding.attachment_document_id == restored.id
            )
        )
        assert restored.security_level == "PUBLIC"
        assert restored.visibility_type == "ALL_AUTHENTICATED"
        assert restored.category_id is None
        assert binding is not None and binding.is_deleted


def test_restore_validates_body_state_and_old_path_stays_404(
    b3_app, b3_client, login_token
):
    category_id = _category_id(b3_app)
    document = _upload_plan(
        b3_client, login_token, category_id, "B8请求校验"
    )
    response = _restore(
        b3_client, login_token, document["id"], document["rowVersion"]
    )
    assert response.status_code == 409
    assert response.get_json()["code"] == "DOCUMENT_NOT_DELETED"
    assert (
        b3_client.post(
            f"/api/v1/documents/{document['id']}/restore",
            json={"rowVersion": document["rowVersion"]},
            headers=_headers(login_token),
        ).status_code
        == 404
    )
    _delete(b3_client, login_token, document)
    deleted = _deleted_document(b3_app, document["id"])
    for payload in (
        {},
        {"rowVersion": True},
        {"rowVersion": "1"},
        {"rowVersion": 1.5},
        {"rowVersion": -1},
        {"rowVersion": deleted["rowVersion"], "extra": 1},
    ):
        invalid = b3_client.post(
            f"/api/v1/recycle-bin/documents/{document['id']}/restore",
            json=payload,
            headers=_headers(login_token),
        )
        assert invalid.status_code == 400
        assert invalid.get_json()["code"] == "INVALID_ARGUMENT"
    conflict = _restore(
        b3_client, login_token, document["id"], deleted["rowVersion"] - 1
    )
    assert conflict.status_code == 409
    assert conflict.get_json()["code"] == "DATA_VERSION_CONFLICT"
    assert (
        conflict.get_json()["details"]["currentRowVersion"]
        == deleted["rowVersion"]
    )


@pytest.mark.parametrize(
    ("mutation", "expected_code", "expected_status"),
    [
        ("missing", "FILE_NOT_FOUND", 404),
        ("size", "FILE_INTEGRITY_MISMATCH", 409),
        ("hash", "FILE_INTEGRITY_MISMATCH", 409),
        ("path", "FILE_PATH_INVALID", 500),
        ("type", "FILE_INTEGRITY_MISMATCH", 409),
    ],
)
def test_restore_file_failures_are_safe_and_rollback(
    b3_app, b3_client, login_token, mutation, expected_code, expected_status
):
    from dms.extensions import db
    from dms.models import Document
    from dms.storage.paths import resolve_storage_path

    category_id = _category_id(b3_app)
    document = _upload_plan(
        b3_client, login_token, category_id, f"B8文件失败{mutation}"
    )
    _delete(b3_client, login_token, document)
    deleted = _deleted_document(b3_app, document["id"])
    with b3_app.app_context():
        row = db.session.get(Document, int(document["id"]))
        path = resolve_storage_path(
            row.file_relative_path, b3_app.config["DMS_STORAGE_ROOT"]
        )
        if mutation == "missing":
            path.unlink()
        elif mutation == "size":
            path.write_bytes(PDF_BYTES + b"x")
        elif mutation == "hash":
            changed = bytearray(PDF_BYTES)
            changed[-2] = ord("X")
            path.write_bytes(changed)
        elif mutation == "path":
            row.file_relative_path = "../outside.pdf"
            db.session.commit()
        elif mutation == "type":
            path.write_bytes(b"NOT_A_PDF" + PDF_BYTES)

    response = _restore(
        b3_client, login_token, document["id"], deleted["rowVersion"]
    )
    assert response.status_code == expected_status
    body = response.get_json()
    assert body["code"] == expected_code
    serialized = json.dumps(body, ensure_ascii=False)
    assert str(Path(b3_app.config["DMS_STORAGE_ROOT"])) not in serialized
    assert "a" * 64 not in serialized
    with b3_app.app_context():
        assert db.session.get(Document, int(document["id"])).is_deleted


def test_restore_file_recheck_and_audit_failure_roll_back(
    b3_app, b3_client, login_token, monkeypatch
):
    from dms.common.errors import RestoreFileChangedError
    from dms.extensions import db
    from dms.models import AuditLog, Document
    from dms.services import recycle_bin_service

    category_id = _category_id(b3_app)
    first = _upload_plan(
        b3_client, login_token, category_id, "B8文件竞态"
    )
    _delete(b3_client, login_token, first)
    first_deleted = _deleted_document(b3_app, first["id"])
    monkeypatch.setattr(
        recycle_bin_service,
        "_recheck_file",
        lambda *_args: (_ for _ in ()).throw(RestoreFileChangedError()),
    )
    response = _restore(
        b3_client, login_token, first["id"], first_deleted["rowVersion"]
    )
    assert response.status_code == 409
    assert response.get_json()["code"] == "RESTORE_FILE_CHANGED"
    with b3_app.app_context():
        assert db.session.get(Document, int(first["id"])).is_deleted
    monkeypatch.undo()

    second = _upload_plan(
        b3_client, login_token, category_id, "B8审计回滚"
    )
    _delete(b3_client, login_token, second)
    second_deleted = _deleted_document(b3_app, second["id"])

    def fail_audit(**_kwargs):
        raise RuntimeError("audit failed")

    monkeypatch.setattr(recycle_bin_service, "business_audit", fail_audit)
    audit_failure = _restore(
        b3_client, login_token, second["id"], second_deleted["rowVersion"]
    )
    assert audit_failure.status_code == 500
    assert audit_failure.get_json()["code"] == "INTERNAL_ERROR"
    with b3_app.app_context():
        assert db.session.get(Document, int(second["id"])).is_deleted
        count = db.session.scalar(
            db.select(db.func.count(AuditLog.id)).where(
                AuditLog.action_type == "RESTORE_DOCUMENT",
                AuditLog.target_id == int(second["id"]),
            )
        )
        assert count == 0


def test_restore_invalid_category_and_parent_roll_back(
    b3_app, b3_client, login_token
):
    from dms.extensions import db
    from dms.models import Category, Document

    category_id = _category_id(b3_app)
    main = _upload_plan(
        b3_client, login_token, category_id, "B8失效分类"
    )
    _delete(b3_client, login_token, main)
    deleted_main = _deleted_document(b3_app, main["id"])
    with b3_app.app_context():
        category = db.session.get(Category, int(category_id))
        category.status = "DISABLED"
        db.session.commit()
    invalid_category = _restore(
        b3_client, login_token, main["id"], deleted_main["rowVersion"]
    )
    assert invalid_category.status_code == 409
    assert invalid_category.get_json()["code"] == "RESTORE_CATEGORY_INVALID"
    with b3_app.app_context():
        category = db.session.get(Category, int(category_id))
        category.status = "ENABLED"
        db.session.commit()

    parent = _upload_plan(
        b3_client, login_token, category_id, "B8失效父主案"
    )
    sub = _upload_plan(
        b3_client,
        login_token,
        category_id,
        "B8父级失效子案",
        document_type="SUB_PLAN",
        parent_id=parent["id"],
    )
    _delete(b3_client, login_token, sub)
    deleted_sub = _deleted_document(b3_app, sub["id"])
    with b3_app.app_context():
        parent_row = db.session.get(Document, int(parent["id"]))
        parent_row.is_deleted = True
        parent_row.deleted_at = datetime.now(timezone.utc).replace(tzinfo=None)
        db.session.commit()
    invalid_parent = _restore(
        b3_client, login_token, sub["id"], deleted_sub["rowVersion"]
    )
    assert invalid_parent.status_code == 409
    assert invalid_parent.get_json()["code"] == "RESTORE_PARENT_INVALID"


def test_restore_custom_skips_disabled_subject_but_organization_rejects(
    b3_app, b3_client, login_token
):
    from dms.extensions import db
    from dms.models import Organization, User

    category_id = _category_id(b3_app)
    with b3_app.app_context():
        user = db.session.scalar(db.select(User).where(User.username == "user"))
        organization = db.session.scalar(
            db.select(Organization).where(Organization.org_code == "ORG_ROOT")
        )
        assert user is not None and organization is not None
        user_id = str(user.id)
        organization_id = str(organization.id)

    def permission(document, visibility, subject_type, subject_id):
        response = b3_client.put(
            f"/api/v1/documents/{document['id']}/permissions",
            json={
                "visibilityType": visibility,
                "documentRowVersion": document["rowVersion"],
                "entries": [
                    {
                        "subjectType": subject_type,
                        "subjectId": subject_id,
                        "canView": True,
                        "canDownload": False,
                        "canEdit": False,
                        "canManagePermission": False,
                        "canDelete": False,
                    }
                ],
            },
            headers=_headers(login_token),
        )
        assert response.status_code == 200
        document["rowVersion"] = response.get_json()["data"]["documentRowVersion"]

    custom = _upload_plan(
        b3_client,
        login_token,
        category_id,
        "B8自定义跳过",
        visibility="CUSTOM",
    )
    permission(custom, "CUSTOM", "USER", user_id)
    _delete(b3_client, login_token, custom)
    custom_deleted = _deleted_document(b3_app, custom["id"])
    with b3_app.app_context():
        user = db.session.get(User, int(user_id))
        user.status = "DISABLED"
        db.session.commit()
    custom_restore = _restore(
        b3_client, login_token, custom["id"], custom_deleted["rowVersion"]
    )
    assert custom_restore.status_code == 200, custom_restore.get_json()
    assert custom_restore.get_json()["data"]["skippedPermissionCount"] == 1

    organization_main = _upload_plan(
        b3_client,
        login_token,
        category_id,
        "B8组织权限拒绝",
        visibility="ORGANIZATION",
    )
    permission(
        organization_main, "ORGANIZATION", "ORG", organization_id
    )
    _delete(b3_client, login_token, organization_main)
    organization_deleted = _deleted_document(
        b3_app, organization_main["id"]
    )
    with b3_app.app_context():
        organization = db.session.get(Organization, int(organization_id))
        organization.status = "DISABLED"
        db.session.commit()
    rejected = _restore(
        b3_client,
        login_token,
        organization_main["id"],
        organization_deleted["rowVersion"],
    )
    assert rejected.status_code == 409
    assert rejected.get_json()["code"] == "RESTORE_PERMISSION_INVALID"


def test_deleted_by_is_null_without_reliable_audit_and_query_has_no_n_plus_one(
    b3_app, b3_client, login_token
):
    from sqlalchemy import event

    from dms.extensions import db
    from dms.models import AuditLog

    category_id = _category_id(b3_app)
    documents = [
        _upload_plan(
            b3_client, login_token, category_id, f"B8批量删除人{i}"
        )
        for i in range(3)
    ]
    for document in documents:
        _delete(b3_client, login_token, document)
    with b3_app.app_context():
        db.session.execute(
            db.delete(AuditLog).where(
                AuditLog.action_type == "DELETE_DOCUMENT",
                AuditLog.target_id == int(documents[0]["id"]),
            )
        )
        db.session.commit()

    select_count = 0

    def count_selects(
        _conn, _cursor, statement, _parameters, _context, _executemany
    ):
        nonlocal select_count
        if statement.lstrip().upper().startswith("SELECT"):
            select_count += 1

    with b3_app.app_context():
        event.listen(db.engine, "before_cursor_execute", count_selects)
    try:
        response = b3_client.get(
            "/api/v1/recycle-bin/documents?pageSize=100",
            headers=_headers(login_token),
        )
    finally:
        with b3_app.app_context():
            event.remove(db.engine, "before_cursor_execute", count_selects)
    assert response.status_code == 200
    items = {
        item["id"]: item for item in response.get_json()["data"]["items"]
    }
    assert items[documents[0]["id"]]["deletedBy"] is None
    assert items[documents[1]["id"]]["deletedBy"]["username"] == "admin"
    # Token校验1次、分页count 1次、文档页1次、窗口审计1次。
    assert select_count <= 4


def test_restore_relation_conflicts_and_all_authenticated_acl_roll_back(
    b3_app, b3_client, login_token
):
    from dms.extensions import db
    from dms.models import Document, Organization, Permission

    category_id = _category_id(b3_app)
    with b3_app.app_context():
        organization = db.session.scalar(
            db.select(Organization).where(Organization.org_code == "ORG_ROOT")
        )
        assert organization is not None
        organization_id = str(organization.id)

    main = _upload_plan(
        b3_client,
        login_token,
        category_id,
        "B8权限关系冲突",
        visibility="ORGANIZATION",
    )
    permission = b3_client.put(
        f"/api/v1/documents/{main['id']}/permissions",
        json={
            "visibilityType": "ORGANIZATION",
            "documentRowVersion": main["rowVersion"],
            "entries": [
                {
                    "subjectType": "ORG",
                    "subjectId": organization_id,
                    "canView": True,
                    "canDownload": False,
                    "canEdit": False,
                    "canManagePermission": False,
                    "canDelete": False,
                }
            ],
        },
        headers=_headers(login_token),
    )
    main["rowVersion"] = permission.get_json()["data"]["documentRowVersion"]
    _delete(b3_client, login_token, main)
    deleted = _deleted_document(b3_app, main["id"])
    with b3_app.app_context():
        db.session.add(
            Permission(
                document_id=int(main["id"]),
                subject_type="ORG",
                subject_id=int(organization_id),
                subject_name="并发新权限",
                can_view=True,
            )
        )
        db.session.commit()
    conflict = _restore(
        b3_client, login_token, main["id"], deleted["rowVersion"]
    )
    assert conflict.status_code == 409
    assert conflict.get_json()["code"] == "RESTORE_RELATION_CONFLICT"
    assert conflict.get_json()["details"]["permissionConflictCount"] == 1
    with b3_app.app_context():
        assert db.session.get(Document, int(main["id"])).is_deleted

    all_main = _upload_plan(
        b3_client,
        login_token,
        category_id,
        "B8全员历史ACL",
        visibility="CUSTOM",
    )
    saved = b3_client.put(
        f"/api/v1/documents/{all_main['id']}/permissions",
        json={
            "visibilityType": "CUSTOM",
            "documentRowVersion": all_main["rowVersion"],
            "entries": [
                {
                    "subjectType": "ORG",
                    "subjectId": organization_id,
                    "canView": True,
                    "canDownload": False,
                    "canEdit": False,
                    "canManagePermission": False,
                    "canDelete": False,
                }
            ],
        },
        headers=_headers(login_token),
    ).get_json()["data"]
    # 构造历史异常数据：全员可见文档仍残留有效ACL。
    with b3_app.app_context():
        document = db.session.get(Document, int(all_main["id"]))
        document.visibility_type = "ALL_AUTHENTICATED"
        document.row_version = saved["documentRowVersion"] + 1
        db.session.commit()
        all_main["rowVersion"] = document.row_version
        all_main["visibilityType"] = document.visibility_type
    _delete(b3_client, login_token, all_main)
    all_deleted = _deleted_document(b3_app, all_main["id"])
    invalid = _restore(
        b3_client, login_token, all_main["id"], all_deleted["rowVersion"]
    )
    assert invalid.status_code == 409
    assert invalid.get_json()["code"] == "RESTORE_PERMISSION_INVALID"
    assert (
        invalid.get_json()["details"]["reason"]
        == "ALL_AUTHENTICATED_HAS_PERMISSIONS"
    )


def test_restore_attachment_rejects_anomalous_active_binding(
    b3_app, b3_client, login_token
):
    from dms.extensions import db
    from dms.models import AttachmentBinding, Document

    category_id = _category_id(b3_app)
    main = _upload_plan(
        b3_client, login_token, category_id, "B8异常挂载主案"
    )
    attachment = _upload_attachment(
        b3_client, login_token, "B8异常挂载附件"
    )
    _delete(b3_client, login_token, attachment, attachment=True)
    deleted = _deleted_document(b3_app, attachment["id"])
    with b3_app.app_context():
        db.session.add(
            AttachmentBinding(
                main_document_id=int(main["id"]),
                attachment_document_id=int(attachment["id"]),
                sort_no=10,
            )
        )
        db.session.commit()
    response = _restore(
        b3_client, login_token, attachment["id"], deleted["rowVersion"]
    )
    assert response.status_code == 409
    assert response.get_json()["code"] == "RESTORE_RELATION_CONFLICT"
    assert response.get_json()["details"]["bindingConflictCount"] == 1
    with b3_app.app_context():
        assert db.session.get(Document, int(attachment["id"])).is_deleted


def test_concurrent_double_restore_only_one_succeeds(
    b3_app, b3_client, login_token, monkeypatch
):
    from dms.services import recycle_bin_service

    category_id = _category_id(b3_app)
    document = _upload_plan(
        b3_client, login_token, category_id, "B8并发双恢复"
    )
    _delete(b3_client, login_token, document)
    deleted = _deleted_document(b3_app, document["id"])
    original = recycle_bin_service._restore_operation
    barrier = threading.Barrier(2)

    def synchronized(*args, **kwargs):
        barrier.wait(timeout=10)
        return original(*args, **kwargs)

    monkeypatch.setattr(
        recycle_bin_service, "_restore_operation", synchronized
    )

    def invoke():
        with b3_app.test_client() as client:
            response = _restore(
                client, login_token, document["id"], deleted["rowVersion"]
            )
            return response.status_code, response.get_json()["code"]

    with ThreadPoolExecutor(max_workers=2) as executor:
        results = list(executor.map(lambda _index: invoke(), range(2)))
    assert sorted(results) == [
        (200, "OK"),
        (409, "DOCUMENT_NOT_DELETED"),
    ]