"""Add data-access effect run history and reconcile rollups. Revision ID: 333 Revises: 332 Create Date: 2026-06-11 DAX-026 makes observe-run outcomes visible beyond the latest effect row. Each processed compiled effect gets a history row, while request/binding rows carry a compact reconciliation roll-up for admin surfaces. """ from __future__ import annotations import sqlalchemy as sa from alembic import op from app.core.migration_helpers import ( id_type_for, normalize_fk_column_collation, table_kw_for, ) revision = "336" down_revision = "335" branch_labels = None depends_on = None REQUEST_TABLE = "data_access_requests" BINDING_TABLE = "data_access_bindings" RUN_TABLE = "data_access_effect_runs" def _table_exists(bind, table: str) -> bool: return bool(bind.execute( sa.text( "SELECT 1 FROM information_schema.tables " "WHERE table_schema = DATABASE() AND table_name = :t LIMIT 1", ), {"t": table}, ).scalar()) def _column_exists(bind, table: str, column: str) -> bool: return bool(bind.execute( sa.text( "SELECT 1 FROM information_schema.columns " "WHERE table_schema = DATABASE() " "AND table_name = :t AND column_name = :c LIMIT 1", ), {"t": table, "c": column}, ).scalar()) def _index_exists(bind, table: str, index: str) -> bool: return bool(bind.execute( sa.text( "SELECT 1 FROM information_schema.statistics " "WHERE table_schema = DATABASE() AND table_name = :t " "AND index_name = :i LIMIT 1", ), {"t": table, "i": index}, ).scalar()) def _add_rollup_columns( bind, table: str, *, default_status: str, index_name: str, ) -> None: row_id = id_type_for("identity_persons") if not _column_exists(bind, table, "reconcile_status"): op.add_column( table, sa.Column( "reconcile_status", sa.String(32), nullable=False, server_default=default_status, ), ) if not _column_exists(bind, table, "last_reconcile_run_id"): op.add_column( table, sa.Column("last_reconcile_run_id", row_id, nullable=True), ) if not _column_exists(bind, table, "last_reconcile_at"): op.add_column( table, sa.Column("last_reconcile_at", sa.DateTime(timezone=True)), ) if not _column_exists(bind, table, "last_reconcile_error"): op.add_column(table, sa.Column("last_reconcile_error", sa.Text())) if not _index_exists(bind, table, index_name): op.create_index(index_name, table, ["reconcile_status"]) def _create_run_table(bind) -> None: if _table_exists(bind, RUN_TABLE): return row_id = id_type_for("identity_persons") effect_id = id_type_for("data_access_compiled_effects") binding_id = id_type_for("data_access_bindings") request_id = id_type_for("data_access_requests") resource_id = id_type_for("data_access_resources") integration_id = id_type_for("integrations") kw = table_kw_for("identity_persons") op.create_table( RUN_TABLE, sa.Column("id", row_id, primary_key=True), sa.Column("run_id", row_id, nullable=False), sa.Column("effect_id", effect_id, nullable=False), sa.Column("binding_id", binding_id, nullable=False), sa.Column("request_id", request_id, nullable=True), sa.Column("resource_id", resource_id, nullable=False), sa.Column("integration_id", integration_id, nullable=False), sa.Column("trace_id", sa.String(128), nullable=True), sa.Column( "mode", sa.String(16), nullable=False, server_default="observe", ), sa.Column("provider", sa.String(32), nullable=False), sa.Column("resource_type", sa.String(32), nullable=False), sa.Column( "target_project_id", sa.String(64, collation="utf8mb4_bin"), nullable=False, ), sa.Column( "target_dataset_id", sa.String(1024, collation="utf8mb4_bin"), nullable=False, ), sa.Column("provider_member", sa.String(320), nullable=False), sa.Column("provider_role", sa.String(128), nullable=False), sa.Column("desired_state", sa.String(16), nullable=False), sa.Column("effect_status", sa.String(32), nullable=False), sa.Column("provider_outcome", sa.String(64), nullable=False), sa.Column("provider_etag", sa.String(256), nullable=True), sa.Column("error_message", sa.Text(), nullable=True), sa.Column("started_at", sa.DateTime(timezone=True), nullable=False), sa.Column("completed_at", sa.DateTime(timezone=True), nullable=False), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), **kw, ) for column, parent, nullable in ( ("effect_id", "data_access_compiled_effects", False), ("binding_id", "data_access_bindings", False), ("request_id", "data_access_requests", True), ("resource_id", "data_access_resources", False), ("integration_id", "integrations", False), ): normalize_fk_column_collation( RUN_TABLE, column, parent, nullable=nullable, ) op.create_foreign_key( "fk_data_access_effect_runs_effect", RUN_TABLE, "data_access_compiled_effects", ["effect_id"], ["id"], ondelete="CASCADE", ) op.create_foreign_key( "fk_data_access_effect_runs_binding", RUN_TABLE, "data_access_bindings", ["binding_id"], ["id"], ) op.create_foreign_key( "fk_data_access_effect_runs_request", RUN_TABLE, "data_access_requests", ["request_id"], ["id"], ondelete="SET NULL", ) op.create_foreign_key( "fk_data_access_effect_runs_resource", RUN_TABLE, "data_access_resources", ["resource_id"], ["id"], ) op.create_foreign_key( "fk_data_access_effect_runs_integration", RUN_TABLE, "integrations", ["integration_id"], ["id"], ondelete="CASCADE", ) for index_name, columns in ( ("ix_data_access_effect_runs_run", ["run_id"]), ("ix_data_access_effect_runs_effect", ["effect_id"]), ("ix_data_access_effect_runs_binding", ["binding_id"]), ("ix_data_access_effect_runs_request", ["request_id"]), ("ix_data_access_effect_runs_status", ["effect_status"]), ("ix_data_access_effect_runs_trace", ["trace_id"]), ): op.create_index(index_name, RUN_TABLE, columns) def upgrade() -> None: bind = op.get_bind() _add_rollup_columns( bind, REQUEST_TABLE, default_status="not_started", index_name="ix_data_access_requests_reconcile_status", ) _add_rollup_columns( bind, BINDING_TABLE, default_status="pending_reconcile", index_name="ix_data_access_bindings_reconcile_status", ) _create_run_table(bind) def downgrade() -> None: bind = op.get_bind() if _table_exists(bind, RUN_TABLE): op.drop_table(RUN_TABLE) for table, index_name in ( (BINDING_TABLE, "ix_data_access_bindings_reconcile_status"), (REQUEST_TABLE, "ix_data_access_requests_reconcile_status"), ): if _index_exists(bind, table, index_name): op.drop_index(index_name, table_name=table) for column in ( "last_reconcile_error", "last_reconcile_at", "last_reconcile_run_id", "reconcile_status", ): if _column_exists(bind, table, column): op.drop_column(table, column)