document-management/backend/migrations/versions/0036_notification.py

"""站内通知表 sys_notification（契约 1.34）。

Revision ID: 0036_notification
Revises: 0035_drop_audit_action_check
Create Date: 2026-09-15

变更要点：
1. 新建 sys_notification：用户站内通知（审批待办等），追加型表，
   不含业务表 mixin 的软删/乐观锁列；
2. 双引擎幂等：先探测表存在再建（MySQL 与达梦升级互不冲突）；
3. 降级保护：存在通知记录时拒绝有损降级。
"""

from __future__ import annotations

from collections.abc import Sequence

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

revision: str = "0036_notification"
down_revision: str | None = "0035_drop_audit_action_check"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None


def _table_exists(table: str) -> bool:
    # 达梦 inspector 返回大写表名，统一小写比较
    inspector = sa.inspect(op.get_bind())
    return table.lower() in {name.lower() for name in inspector.get_table_names()}


def upgrade() -> None:
    if _table_exists("sys_notification"):
        return

    op.create_table(
        "sys_notification",
        sa.Column("id", sa.BigInteger(), autoincrement=True, nullable=False),
        sa.Column("user_id", sa.BigInteger(), nullable=False),
        sa.Column("title", sa.String(length=200), nullable=False),
        sa.Column("content", sa.String(length=1000), nullable=True),
        sa.Column("notify_type", sa.String(length=32), nullable=False),
        sa.Column("action", sa.String(length=16), nullable=False),
        sa.Column("status", sa.String(length=16), nullable=False),
        sa.Column(
            "created_at",
            mysql.DATETIME(fsp=3),
            server_default=sa.text("CURRENT_TIMESTAMP(3)"),
            nullable=False,
        ),
        sa.Column("handled_at", mysql.DATETIME(fsp=3), nullable=True),
        sa.PrimaryKeyConstraint("id"),
        sa.CheckConstraint(
            "notify_type IN ('APPROVAL')",
            name="notification_notify_type",
        ),
        sa.CheckConstraint(
            "action IN ('NONE', 'PENDING', 'AGREE', 'REJECT')",
            name="notification_action",
        ),
        sa.CheckConstraint(
            "status IN ('UNREAD', 'READ')",
            name="notification_status",
        ),
    )
    op.create_index(
        "ix_sys_notification_user_id",
        "sys_notification",
        ["user_id", "id"],
    )


def downgrade() -> None:
    if not _table_exists("sys_notification"):
        return
    rows = op.get_bind().scalar(sa.text("SELECT COUNT(*) FROM sys_notification"))
    if rows:
        raise RuntimeError(
            "0036已存在通知记录，拒绝有损降级；请先备份并按数据治理流程处理"
        )
    op.drop_index("ix_sys_notification_user_id", table_name="sys_notification")
    op.drop_table("sys_notification")