document-management/backend/migrations/versions/0014_document_relations.py

"""方案关联关系：新表 doc_document_relation。

Revision ID: 0014_document_relations
Revises: 0013_document_password_lock
Create Date: 2026-08-22

变更要点：
1. 新建 doc_document_relation：方案（MAIN/SUB_PLAN）声明的关联关系，
   relation_type 存最终文本（预置 label 或用户自定义文字，1~64 字符）；
2. 软删 + 生成列唯一约束（uq_doc_document_relation_active）保证同一发起方
   对同一目标至多一条有效关联；CHECK 禁止自关联；
3. 审计复用 EDIT_DOCUMENT 动作（detail 标注 scope=relations），
   不扩展 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 = "0014_document_relations"
down_revision: str | None = "0013_document_password_lock"
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 upgrade() -> None:
    op.create_table(
        "doc_document_relation",
        *_business_columns(),
        sa.Column("document_id", sa.BigInteger(), nullable=False),
        sa.Column("related_document_id", sa.BigInteger(), nullable=False),
        sa.Column("relation_type", sa.String(length=64), 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.CheckConstraint(
            "document_id <> related_document_id",
            name="document_relation_no_self",
        ),
        sa.ForeignKeyConstraint(
            ["document_id"],
            ["doc_document.id"],
            name="fk_doc_document_relation_document_id_doc_document",
            ondelete="RESTRICT",
        ),
        sa.ForeignKeyConstraint(
            ["related_document_id"],
            ["doc_document.id"],
            name="fk_doc_document_relation_related_document_id_doc_document",
            ondelete="RESTRICT",
        ),
        sa.UniqueConstraint(
            "document_id",
            "related_document_id",
            "active_marker",
            name="uq_doc_document_relation_active",
        ),
    )
    op.create_index(
        "ix_doc_document_relation_document_active_sort",
        "doc_document_relation",
        ["document_id", "is_deleted", "sort_no"],
    )
    op.create_index(
        "ix_doc_document_relation_related_active",
        "doc_document_relation",
        ["related_document_id", "is_deleted"],
    )


def downgrade() -> None:
    bind = op.get_bind()
    relation_rows = bind.scalar(
        sa.text("SELECT COUNT(*) FROM doc_document_relation")
    )
    if relation_rows:
        raise RuntimeError(
            "0014已产生方案关联数据，拒绝有损降级；请先备份并按数据治理流程处理"
        )
    op.drop_table("doc_document_relation")