X-Financial/server/alembic/versions/20260716_0014_risk_waiver_decision.py

"""add risk waiver decision metadata and separation of duties state

Revision ID: 20260716_0014
Revises: 20260716_0013
Create Date: 2026-07-16 19:10:00
"""

from __future__ import annotations

from collections.abc import Sequence

import sqlalchemy as sa

from alembic import op

revision: str = "20260716_0014"
down_revision: str | None = "20260716_0013"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None


def _require_postgresql() -> None:
    dialect_name = op.get_bind().dialect.name
    if dialect_name != "postgresql":
        raise RuntimeError(
            "20260716_0014 only supports PostgreSQL; "
            f"refusing to mutate {dialect_name} without transactional constraint DDL"
        )


def _require_no_waiver_audit_data_for_downgrade() -> None:
    bind = op.get_bind()
    disposition_count = int(
        bind.scalar(
            sa.text(
                "SELECT COUNT(*) FROM risk_dispositions "
                "WHERE waiver_requester_id IS NOT NULL "
                "OR waiver_decision IS NOT NULL "
                "OR lifecycle_status IN ('waived', 'waiver_rejected')"
            )
        )
        or 0
    )
    event_count = int(
        bind.scalar(
            sa.text(
                "SELECT COUNT(*) FROM risk_disposition_events "
                "WHERE action IN ('approve_waiver', 'reject_waiver')"
            )
        )
        or 0
    )
    if disposition_count or event_count:
        raise RuntimeError(
            "cannot downgrade risk waiver decisions: immutable waiver audit data exists "
            f"({disposition_count} dispositions, {event_count} decision events)"
        )


def upgrade() -> None:
    _require_postgresql()
    op.add_column(
        "risk_dispositions",
        sa.Column("waiver_requester_id", sa.String(length=120), nullable=True),
    )
    op.add_column(
        "risk_dispositions",
        sa.Column("waiver_requester_name", sa.String(length=120), nullable=True),
    )
    op.add_column(
        "risk_dispositions",
        sa.Column("waiver_requested_at", sa.DateTime(timezone=True), nullable=True),
    )
    op.add_column(
        "risk_dispositions",
        sa.Column("waiver_reason", sa.Text(), nullable=True),
    )
    op.add_column(
        "risk_dispositions",
        sa.Column("waiver_scope", sa.Text(), nullable=True),
    )
    op.add_column(
        "risk_dispositions",
        sa.Column("waiver_expires_at", sa.DateTime(timezone=True), nullable=True),
    )
    op.add_column(
        "risk_dispositions",
        sa.Column(
            "waiver_conditions_json",
            sa.JSON(),
            nullable=False,
            server_default=sa.text("'[]'::json"),
        ),
    )
    op.add_column(
        "risk_dispositions",
        sa.Column("waiver_decision", sa.String(length=20), nullable=True),
    )
    op.add_column(
        "risk_dispositions",
        sa.Column("waiver_decider_id", sa.String(length=120), nullable=True),
    )
    op.add_column(
        "risk_dispositions",
        sa.Column("waiver_decider_name", sa.String(length=120), nullable=True),
    )
    op.add_column(
        "risk_dispositions",
        sa.Column("waiver_decided_at", sa.DateTime(timezone=True), nullable=True),
    )
    op.add_column(
        "risk_dispositions",
        sa.Column("waiver_decision_reason", sa.Text(), nullable=True),
    )

    op.drop_constraint(
        "ck_risk_dispositions_lifecycle",
        "risk_dispositions",
        type_="check",
    )
    op.create_check_constraint(
        "ck_risk_dispositions_lifecycle",
        "risk_dispositions",
        "lifecycle_status IN ('open', 'supplement_requested', "
        "'remediation_in_progress', 'waiver_requested', 'waived', "
        "'waiver_rejected', 'resolved')",
    )
    op.create_check_constraint(
        "ck_risk_dispositions_waiver_request",
        "risk_dispositions",
        "(waiver_requester_id IS NULL AND waiver_requester_name IS NULL "
        "AND waiver_requested_at IS NULL AND waiver_reason IS NULL "
        "AND waiver_scope IS NULL AND waiver_expires_at IS NULL) OR "
        "(waiver_requester_id IS NOT NULL "
        "AND length(trim(waiver_requester_id)) > 0 "
        "AND waiver_requester_name IS NOT NULL "
        "AND length(trim(waiver_requester_name)) > 0 "
        "AND waiver_requested_at IS NOT NULL "
        "AND waiver_reason IS NOT NULL AND length(trim(waiver_reason)) > 0 "
        "AND waiver_scope IS NOT NULL AND length(trim(waiver_scope)) > 0 "
        "AND waiver_expires_at IS NOT NULL "
        "AND waiver_expires_at > waiver_requested_at)",
    )
    op.create_check_constraint(
        "ck_risk_dispositions_waiver_decision",
        "risk_dispositions",
        "(waiver_decision IS NULL AND waiver_decider_id IS NULL "
        "AND waiver_decider_name IS NULL AND waiver_decided_at IS NULL "
        "AND waiver_decision_reason IS NULL) OR "
        "(waiver_decision IN ('approved', 'rejected') "
        "AND waiver_requester_id IS NOT NULL "
        "AND waiver_decider_id IS NOT NULL "
        "AND length(trim(waiver_decider_id)) > 0 "
        "AND waiver_decider_name IS NOT NULL "
        "AND length(trim(waiver_decider_name)) > 0 "
        "AND waiver_decided_at IS NOT NULL "
        "AND waiver_decision_reason IS NOT NULL "
        "AND length(trim(waiver_decision_reason)) > 0)",
    )
    op.create_check_constraint(
        "ck_risk_dispositions_waiver_lifecycle",
        "risk_dispositions",
        "(lifecycle_status != 'waived' OR waiver_decision = 'approved') AND "
        "(lifecycle_status != 'waiver_rejected' OR waiver_decision = 'rejected')",
    )
    op.create_index(
        "ix_risk_dispositions_tenant_waiver_expiry",
        "risk_dispositions",
        ["tenant_id", "lifecycle_status", "waiver_expires_at"],
        unique=False,
    )

    op.drop_constraint(
        "ck_risk_disposition_events_action",
        "risk_disposition_events",
        type_="check",
    )
    op.create_check_constraint(
        "ck_risk_disposition_events_action",
        "risk_disposition_events",
        "action IN ('confirm', 'false_positive', 'request_supplement', "
        "'start_remediation', 'resolve', 'request_waiver', "
        "'approve_waiver', 'reject_waiver')",
    )


def downgrade() -> None:
    _require_postgresql()
    # 申请与决定元数据已经进入不可变事件快照时，禁止静默丢弃审计链。
    _require_no_waiver_audit_data_for_downgrade()
    op.drop_constraint(
        "ck_risk_disposition_events_action",
        "risk_disposition_events",
        type_="check",
    )
    op.create_check_constraint(
        "ck_risk_disposition_events_action",
        "risk_disposition_events",
        "action IN ('confirm', 'false_positive', 'request_supplement', "
        "'start_remediation', 'resolve', 'request_waiver')",
    )

    op.drop_index(
        "ix_risk_dispositions_tenant_waiver_expiry",
        table_name="risk_dispositions",
    )
    op.drop_constraint(
        "ck_risk_dispositions_waiver_lifecycle",
        "risk_dispositions",
        type_="check",
    )
    op.drop_constraint(
        "ck_risk_dispositions_waiver_decision",
        "risk_dispositions",
        type_="check",
    )
    op.drop_constraint(
        "ck_risk_dispositions_waiver_request",
        "risk_dispositions",
        type_="check",
    )
    op.drop_constraint(
        "ck_risk_dispositions_lifecycle",
        "risk_dispositions",
        type_="check",
    )
    op.create_check_constraint(
        "ck_risk_dispositions_lifecycle",
        "risk_dispositions",
        "lifecycle_status IN ('open', 'supplement_requested', "
        "'remediation_in_progress', 'waiver_requested', 'resolved')",
    )

    for column_name in (
        "waiver_decision_reason",
        "waiver_decided_at",
        "waiver_decider_name",
        "waiver_decider_id",
        "waiver_decision",
        "waiver_conditions_json",
        "waiver_expires_at",
        "waiver_scope",
        "waiver_reason",
        "waiver_requested_at",
        "waiver_requester_name",
        "waiver_requester_id",
    ):
        op.drop_column("risk_dispositions", column_name)