"""Frozen evidence bundles for the auditor evidence-query catalogue.

Revision ID: 485
Revises: 484

审计季取证今天是「现查现看」:auditor 打开页面跑一遍,看到的是**此刻**的
数据。这有两个问题 ——(1) 外部审计师在 11 月重开同一份材料时,底层行已经
变了,他看到的和 8 月给他的不是同一份;(2) SOC2 对 IPE(Information Produced
by Entity)要求能证明清单**完整**,而一张截图证明不了任何东西。

所以一次运行被**冻结**成不可变的 bundle:表头记下问的是什么
(`query_key` + `query_version` + 规范化后的 `params`)和对结果行取的
`content_hash`;行表逐行记下**出处**(`source_table`+`source_pk`,或审计行
来源时的 `audit_log_id`+`source_trace_id`)。

同一问题重跑产生**新** bundle 而不是覆盖旧的。两个 bundle 的 hash 不同,
说明底层数据在两次之间动了 —— 这本身就是审计信号,不是工具的 bug。这也是
为什么这两张表没有 UPDATE 路径。

行**落库而不是读时重算**:重算的结果会变,那就不叫冻结了。~500 人规模下
单次名册最多几百行,存储代价可忽略,而换来的是下钻、分页和证据包 CSV 渲染
全都变成普通查询。

**不建外键**(同 `audit_logs` / 迁移 484 的持有区间台账):证据行必须比它
引用的 person / 目录 / 授权行活得久 —— 离职者被 HRIS purge、软件目录行软删
之后,「他当时持有过」仍然是证据。列类型用 `id_type_for` 与父表对齐,
JOIN 不受影响。
"""
import sqlalchemy as sa

from alembic import op
from app.core.migration_helpers import id_type_for, table_kw_for
from app.core.sql_types import PreciseDateTime as _PRECISE_TS

revision = "485"
down_revision = "484"
branch_labels = None
depends_on = None

BUNDLE_TABLE = "auditor_evidence_bundles"
ROW_TABLE = "auditor_evidence_bundle_rows"


def _table_exists(bind, name: str) -> bool:
    return bool(
        bind.execute(
            sa.text(
                "SELECT 1 FROM information_schema.TABLES "
                "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :t"
            ),
            {"t": name},
        ).fetchone()
    )


def upgrade() -> None:
    bind = op.get_bind()
    row_id = id_type_for("identity_persons")
    kw = table_kw_for("identity_persons")

    if not _table_exists(bind, BUNDLE_TABLE):
        op.create_table(
            BUNDLE_TABLE,
            sa.Column("id", row_id, primary_key=True),
            sa.Column("query_key", sa.String(64), nullable=False),
            # Bumped whenever a handler's logic changes shape. Two
            # bundles with equal params but different versions are not
            # comparable and must never be presented as a diff.
            sa.Column("query_version", sa.String(16), nullable=False),
            sa.Column("params", sa.JSON(), nullable=False),
            # Lifted out of params so the evidence-pack builder and the
            # monthly pre-compile cron can filter by window without
            # parsing each query's own parameter vocabulary.
            sa.Column("period_start", sa.DateTime(timezone=True), nullable=True),
            sa.Column("period_end", sa.DateTime(timezone=True), nullable=True),
            sa.Column(
                "row_count", sa.Integer(), nullable=False, server_default="0",
            ),
            sa.Column("content_hash", sa.String(64), nullable=False),
            # Declared incompleteness (a range predating a ledger, an
            # unreadable source). Surfaced verbatim — silent gaps are
            # what would make this tool worse than no tool.
            sa.Column("data_gaps", sa.JSON(), nullable=True),
            sa.Column("generated_by_person_id", row_id, nullable=True),
            sa.Column(
                "generated_via", sa.String(16), nullable=False,
                server_default="manual",
            ),
            sa.Column("trace_id", row_id, nullable=True),
            sa.Column("duration_ms", sa.Integer(), nullable=True),
            sa.Column(
                "created_at", sa.DateTime(timezone=True), nullable=False,
                server_default=sa.func.now(),
            ),
            **kw,
        )
        op.create_index(
            "idx_auditor_evidence_bundle_query",
            BUNDLE_TABLE,
            ["query_key", "created_at"],
        )
        op.create_index(
            "idx_auditor_evidence_bundle_actor",
            BUNDLE_TABLE,
            ["generated_by_person_id"],
        )

    if not _table_exists(bind, ROW_TABLE):
        op.create_table(
            ROW_TABLE,
            sa.Column("id", row_id, primary_key=True),
            sa.Column("bundle_id", row_id, nullable=False),
            # Emission order. The content hash is computed over rows in
            # this order, so a reader can reproduce it.
            sa.Column("seq", sa.Integer(), nullable=False),
            sa.Column("subject_type", sa.String(32), nullable=False),
            sa.Column("subject_id", sa.String(64), nullable=True),
            sa.Column("subject_label", sa.String(255), nullable=True),
            # Provenance. One of (source_table, source_pk) or
            # audit_log_id is always populated — the service refuses to
            # persist a row that cites nothing.
            sa.Column("source_table", sa.String(64), nullable=True),
            sa.Column("source_pk", sa.String(64), nullable=True),
            sa.Column("audit_log_id", row_id, nullable=True),
            sa.Column("source_trace_id", row_id, nullable=True),
            # DATETIME(6), not DATETIME. ``content_hash`` digests
            # ``occurred_at.isoformat()``, and several sources
            # (``auth_user_role_cycles`` / ``user_permission_cycles``,
            # migration 484) carry microseconds. A whole-second column
            # would round the value UP on insert, so a reader who
            # rebuilds the rows and re-computes the digest would get a
            # different hash than the bundle stores — silently voiding
            # the reproducibility claim this whole table exists for.
            sa.Column("occurred_at", _PRECISE_TS, nullable=True),
            sa.Column("payload", sa.JSON(), nullable=False),
            sa.Column(
                "created_at", sa.DateTime(timezone=True), nullable=False,
                server_default=sa.func.now(),
            ),
            sa.UniqueConstraint(
                "bundle_id", "seq", name="uq_auditor_evidence_row_seq",
            ),
            **kw,
        )
        op.create_index(
            "idx_auditor_evidence_row_bundle", ROW_TABLE, ["bundle_id", "seq"],
        )


def downgrade() -> None:
    bind = op.get_bind()
    if _table_exists(bind, ROW_TABLE):
        op.drop_table(ROW_TABLE)
    if _table_exists(bind, BUNDLE_TABLE):
        op.drop_table(BUNDLE_TABLE)