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

"""add typed risk disposition state and append-only events

Revision ID: 20260716_0011
Revises: 20260716_0010
Create Date: 2026-07-16 16:20:00
"""

from __future__ import annotations

from collections.abc import Sequence

import sqlalchemy as sa

from alembic import op

revision: str = "20260716_0011"
down_revision: str | None = "20260716_0010"
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_0011 only supports PostgreSQL; "
            f"refusing to mutate {dialect_name} without transactional constraint DDL"
        )


def _require_empty_audit_chain_for_downgrade() -> None:
    bind = op.get_bind()
    disposition_count = int(bind.scalar(sa.text("SELECT COUNT(*) FROM risk_dispositions")) or 0)
    event_count = int(bind.scalar(sa.text("SELECT COUNT(*) FROM risk_disposition_events")) or 0)
    if disposition_count or event_count:
        raise RuntimeError(
            "cannot downgrade risk dispositions: audit chain is not empty "
            f"({disposition_count} dispositions, {event_count} events)"
        )


def upgrade() -> None:
    _require_postgresql()
    op.create_unique_constraint(
        "uq_risk_observations_tenant_id",
        "risk_observations",
        ["tenant_id", "id"],
    )
    op.create_table(
        "risk_dispositions",
        sa.Column("id", sa.String(length=36), nullable=False),
        sa.Column("tenant_id", sa.String(length=64), nullable=False),
        sa.Column("observation_id", sa.String(length=36), nullable=False),
        sa.Column(
            "adjudication",
            sa.String(length=24),
            nullable=False,
            server_default="unreviewed",
        ),
        sa.Column(
            "lifecycle_status",
            sa.String(length=32),
            nullable=False,
            server_default="open",
        ),
        sa.Column("version", sa.Integer(), nullable=False, server_default="0"),
        sa.Column("assignee", sa.String(length=120), nullable=True),
        sa.Column("due_at", sa.DateTime(timezone=True), nullable=True),
        sa.Column("resolution", sa.Text(), nullable=True),
        sa.Column(
            "created_at",
            sa.DateTime(timezone=True),
            nullable=False,
            server_default=sa.func.now(),
        ),
        sa.Column(
            "updated_at",
            sa.DateTime(timezone=True),
            nullable=False,
            server_default=sa.func.now(),
        ),
        sa.CheckConstraint(
            "adjudication IN ('unreviewed', 'confirmed', 'false_positive')",
            name="ck_risk_dispositions_adjudication",
        ),
        sa.CheckConstraint(
            "lifecycle_status IN ('open', 'supplement_requested', "
            "'remediation_in_progress', 'waiver_requested', 'resolved')",
            name="ck_risk_dispositions_lifecycle",
        ),
        sa.CheckConstraint("version >= 0", name="ck_risk_dispositions_version"),
        sa.ForeignKeyConstraint(
            ["tenant_id", "observation_id"],
            ["risk_observations.tenant_id", "risk_observations.id"],
            name="fk_risk_dispositions_tenant_observation",
            ondelete="RESTRICT",
        ),
        sa.PrimaryKeyConstraint("id"),
        sa.UniqueConstraint(
            "tenant_id",
            "id",
            name="uq_risk_dispositions_tenant_id",
        ),
        sa.UniqueConstraint(
            "tenant_id",
            "observation_id",
            name="uq_risk_dispositions_tenant_observation",
        ),
    )
    op.create_index(
        "ix_risk_dispositions_tenant_lifecycle_due",
        "risk_dispositions",
        ["tenant_id", "lifecycle_status", "due_at"],
        unique=False,
    )
    op.create_index(
        "ix_risk_dispositions_assignee",
        "risk_dispositions",
        ["tenant_id", "assignee"],
        unique=False,
    )

    op.create_table(
        "risk_disposition_events",
        sa.Column("id", sa.String(length=36), nullable=False),
        sa.Column("tenant_id", sa.String(length=64), nullable=False),
        sa.Column("disposition_id", sa.String(length=36), nullable=False),
        sa.Column("observation_id", sa.String(length=36), nullable=False),
        sa.Column("version", sa.Integer(), nullable=False),
        sa.Column("action", sa.String(length=32), nullable=False),
        sa.Column("actor_id", sa.String(length=120), nullable=False),
        sa.Column("actor_name", sa.String(length=120), nullable=False),
        sa.Column("request_id", sa.String(length=120), nullable=False),
        sa.Column("payload_fingerprint", sa.String(length=64), nullable=False),
        sa.Column("comment", sa.Text(), nullable=True),
        sa.Column("before_json", sa.JSON(), nullable=False),
        sa.Column("after_json", sa.JSON(), nullable=False),
        sa.Column(
            "created_at",
            sa.DateTime(timezone=True),
            nullable=False,
            server_default=sa.func.now(),
        ),
        sa.CheckConstraint(
            "action IN ('confirm', 'false_positive', 'request_supplement', "
            "'start_remediation', 'resolve', 'request_waiver')",
            name="ck_risk_disposition_events_action",
        ),
        sa.CheckConstraint(
            "version > 0",
            name="ck_risk_disposition_events_version",
        ),
        sa.ForeignKeyConstraint(
            ["tenant_id", "disposition_id"],
            ["risk_dispositions.tenant_id", "risk_dispositions.id"],
            name="fk_risk_disposition_events_tenant_disposition",
            ondelete="RESTRICT",
        ),
        sa.ForeignKeyConstraint(
            ["tenant_id", "observation_id"],
            ["risk_observations.tenant_id", "risk_observations.id"],
            name="fk_risk_disposition_events_tenant_observation",
            ondelete="RESTRICT",
        ),
        sa.PrimaryKeyConstraint("id"),
        sa.UniqueConstraint(
            "tenant_id",
            "request_id",
            name="uq_risk_disposition_events_tenant_request",
        ),
        sa.UniqueConstraint(
            "disposition_id",
            "version",
            name="uq_risk_disposition_events_version",
        ),
    )
    op.create_index(
        "ix_risk_disposition_events_disposition_id",
        "risk_disposition_events",
        ["disposition_id"],
        unique=False,
    )
    op.create_index(
        "ix_risk_disposition_events_tenant_observation_time",
        "risk_disposition_events",
        ["tenant_id", "observation_id", "created_at"],
        unique=False,
    )
    op.execute(
        "CREATE FUNCTION reject_risk_disposition_event_mutation() "
        "RETURNS trigger AS $$ "
        "BEGIN "
        "RAISE EXCEPTION 'risk_disposition_events is append-only'; "
        "RETURN OLD; "
        "END; "
        "$$ LANGUAGE plpgsql"
    )
    op.execute(
        "CREATE TRIGGER trg_risk_disposition_events_append_only "
        "BEFORE UPDATE OR DELETE ON risk_disposition_events "
        "FOR EACH ROW EXECUTE FUNCTION reject_risk_disposition_event_mutation()"
    )


def downgrade() -> None:
    _require_postgresql()
    # 审计链不可静默销毁；数据归档或迁移必须由显式运维流程完成。
    _require_empty_audit_chain_for_downgrade()
    op.execute("DROP TRIGGER trg_risk_disposition_events_append_only ON risk_disposition_events")
    op.execute("DROP FUNCTION reject_risk_disposition_event_mutation()")
    op.drop_index(
        "ix_risk_disposition_events_tenant_observation_time",
        table_name="risk_disposition_events",
    )
    op.drop_index(
        "ix_risk_disposition_events_disposition_id",
        table_name="risk_disposition_events",
    )
    op.drop_table("risk_disposition_events")
    op.drop_index(
        "ix_risk_dispositions_assignee",
        table_name="risk_dispositions",
    )
    op.drop_index(
        "ix_risk_dispositions_tenant_lifecycle_due",
        table_name="risk_dispositions",
    )
    op.drop_table("risk_dispositions")
    op.drop_constraint(
        "uq_risk_observations_tenant_id",
        "risk_observations",
        type_="unique",
    )