document-management/backend/migrations/versions/0018_sub_plan_binding.py

"""子案跨主案挂载：新表 doc_sub_plan_binding。

Revision ID: 0018_sub_plan_binding
Revises: 0017_category_type_display_name
Create Date: 2026-08-24

变更要点：
1. 新建 doc_sub_plan_binding：主案对其他主案子案的挂载关系；
2. 软删 + 生成列唯一约束（uq_doc_sub_plan_binding_active）保证同一主案
   对同一子案至多一条有效挂载；主案自有子案（parent_document_id 指向
   该主案）不允许建立挂载，由服务层校验；
3. 审计复用 BIND_ATTACHMENT / UNBIND_ATTACHMENT 动作（detail 标注
   scope=subPlanMount），不扩展 sys_audit_log 的 CHECK 约束；
4. 降级保护：已产生挂载数据时拒绝有损降级。
"""

from __future__ import annotations

from collections.abc import Sequence

from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import mysql

revision: str = "0018_sub_plan_binding"
down_revision: str | None = "0017_category_type_display_name"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None


def _business_columns() -> list[sa.Column]:
    return [
        sa.Column("id", sa.BigInteger(), autoincrement=True, nullable=False),
        sa.Column("created_by", sa.BigInteger(), nullable=True),
        sa.Column("updated_by", sa.BigInteger(), nullable=True),
        sa.Column(
            "created_at",
            mysql.DATETIME(fsp=3),
            server_default=sa.text("CURRENT_TIMESTAMP(3)"),
            nullable=False,
        ),
        sa.Column(
            "updated_at",
            mysql.DATETIME(fsp=3),
            server_default=sa.text("CURRENT_TIMESTAMP(3)"),
            nullable=False,
        ),
        sa.Column("deleted_at", mysql.DATETIME(fsp=3), nullable=True),
        sa.Column(
            "is_deleted",
            sa.Boolean(),
            server_default=sa.text("0"),
            nullable=False,
        ),
        sa.Column(
            "row_version",
            sa.Integer(),
            server_default=sa.text("0"),
            nullable=False,
        ),
        sa.PrimaryKeyConstraint("id"),
    ]


def _existing_tables() -> set[str]:
    inspector = sa.inspect(op.get_bind())
    return set(inspector.get_table_names())


def upgrade() -> None:
    if "doc_sub_plan_binding" in _existing_tables():
        return
    op.create_table(
        "doc_sub_plan_binding",
        *_business_columns(),
        sa.Column("main_document_id", sa.BigInteger(), nullable=False),
        sa.Column("sub_document_id", sa.BigInteger(), nullable=False),
        sa.Column(
            "sort_no",
            sa.Integer(),
            server_default=sa.text("0"),
            nullable=False,
        ),
        sa.Column(
            "active_marker",
            sa.Integer(),
            sa.Computed("IF(is_deleted = 0, 1, NULL)", persisted=False),
            nullable=True,
        ),
        sa.ForeignKeyConstraint(
            ["main_document_id"],
            ["doc_document.id"],
            name="fk_doc_sub_plan_binding_main_id_doc_document",
            ondelete="RESTRICT",
        ),
        sa.ForeignKeyConstraint(
            ["sub_document_id"],
            ["doc_document.id"],
            name="fk_doc_sub_plan_binding_sub_id_doc_document",
            ondelete="RESTRICT",
        ),
        sa.UniqueConstraint(
            "main_document_id",
            "sub_document_id",
            "active_marker",
            name="uq_doc_sub_plan_binding_active",
        ),
    )
    op.create_index(
        "ix_doc_sub_plan_binding_main_active_sort",
        "doc_sub_plan_binding",
        ["main_document_id", "is_deleted", "sort_no"],
    )
    op.create_index(
        "ix_doc_sub_plan_binding_sub_active",
        "doc_sub_plan_binding",
        ["sub_document_id", "is_deleted"],
    )


def downgrade() -> None:
    if "doc_sub_plan_binding" not in _existing_tables():
        return
    bind = op.get_bind()
    binding_rows = bind.scalar(
        sa.text("SELECT COUNT(*) FROM doc_sub_plan_binding")
    )
    if binding_rows:
        raise RuntimeError(
            "0018已产生子案挂载数据，拒绝有损降级；请先备份并按数据治理流程处理"
        )
    op.drop_table("doc_sub_plan_binding")