"""Introduce immutable data-access principals and GCP SA catalog profiles. Revision ID: 459 Revises: 458 Create Date: 2026-08-10 The first Service Account rollout deliberately keeps the legacy Google Group foreign keys as read-compatible projections. Existing groups are projected into ``data_access_principals`` and every historic request/binding/effect/run is backfilled before the new semantic key is enabled. This lets the runtime dual-read Group history while Service Account rows use an immutable Google service-account ``unique_id`` from the trusted catalog. No provider mutation is performed by this migration. """ 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 = "459" down_revision = "458" branch_labels = None depends_on = None PRINCIPAL_TABLE = "data_access_principals" PROFILE_TABLE = "data_access_service_account_profiles" 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 = :table LIMIT 1", ), {"table": 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 = :table " "AND column_name = :column LIMIT 1", ), {"table": table, "column": 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 = :table " "AND index_name = :index LIMIT 1", ), {"table": table, "index": index}).scalar()) def _foreign_key_exists(bind, table: str, constraint: str) -> bool: return bool(bind.execute(sa.text( "SELECT 1 FROM information_schema.table_constraints " "WHERE table_schema = DATABASE() AND table_name = :table " "AND constraint_name = :constraint AND constraint_type = 'FOREIGN KEY' " "LIMIT 1", ), {"table": table, "constraint": constraint}).scalar()) def _column_is_nullable(bind, table: str, column: str) -> bool: value = bind.execute(sa.text( "SELECT is_nullable FROM information_schema.columns " "WHERE table_schema = DATABASE() AND table_name = :table " "AND column_name = :column LIMIT 1", ), {"table": table, "column": column}).scalar() return str(value).upper() == "YES" def _create_foreign_key_if_missing( bind, constraint: str, source_table: str, referent_table: str, local_cols: list[str], remote_cols: list[str], **kwargs, ) -> None: if not _foreign_key_exists(bind, source_table, constraint): op.create_foreign_key( constraint, source_table, referent_table, local_cols, remote_cols, **kwargs, ) def _create_index_if_missing( bind, index: str, table: str, columns: list[str], *, unique: bool = False, ) -> None: if not _index_exists(bind, table, index): op.create_index(index, table, columns, unique=unique) def _add_column_if_missing(table: str, column: sa.Column) -> None: bind = op.get_bind() if not _column_exists(bind, table, column.name): op.add_column(table, column) def _create_principal_tables(bind) -> None: if not _table_exists(bind, PRINCIPAL_TABLE): row_id = id_type_for("identity_persons") group_id = id_type_for("email_groups") op.create_table( PRINCIPAL_TABLE, sa.Column("id", row_id, primary_key=True), sa.Column("principal_type", sa.String(32), nullable=False), sa.Column( "external_subject_id", sa.String(128, collation="utf8mb4_bin"), nullable=False, ), sa.Column( "email_snapshot", sa.String(255, collation="utf8mb4_bin"), nullable=False, ), sa.Column("status", sa.String(32), nullable=False, server_default="active"), sa.Column("catalog_id", sa.String(128, collation="utf8mb4_bin")), sa.Column("source_version", sa.String(128)), sa.Column("source_etag", sa.String(256)), sa.Column("source_payload_digest", sa.String(128)), sa.Column("email_group_id", group_id), 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.UniqueConstraint( "principal_type", "external_subject_id", name="uq_data_access_principals_type_subject", ), sa.UniqueConstraint("email_group_id", name="uq_data_access_principals_group"), **table_kw_for("identity_persons"), ) # TiDB DDL is non-transactional. A failed deployment can leave this table # behind without its trailing FK/index DDL, while Alembic remains on 458. # Always resume those independent steps when the table is already present. normalize_fk_column_collation( PRINCIPAL_TABLE, "email_group_id", "email_groups", nullable=True, ) _create_foreign_key_if_missing( bind, "fk_data_access_principals_email_group", PRINCIPAL_TABLE, "email_groups", ["email_group_id"], ["id"], ondelete="SET NULL", ) _create_index_if_missing( bind, "ix_data_access_principals_type_status", PRINCIPAL_TABLE, ["principal_type", "status"], ) _create_index_if_missing( bind, "ix_data_access_principals_email", PRINCIPAL_TABLE, ["email_snapshot"], ) _create_index_if_missing( bind, "ix_data_access_principals_catalog", PRINCIPAL_TABLE, ["catalog_id"], ) if not _table_exists(bind, PROFILE_TABLE): principal_id = id_type_for(PRINCIPAL_TABLE) person_id = id_type_for("identity_persons") op.create_table( PROFILE_TABLE, sa.Column("principal_id", principal_id, primary_key=True), sa.Column( "catalog_service_account_id", sa.String(128, collation="utf8mb4_bin"), nullable=False, ), sa.Column( "gcp_project_id", sa.String(64, collation="utf8mb4_bin"), nullable=False, ), sa.Column("environment", sa.String(16), nullable=False), sa.Column("workload_ref", sa.String(255), nullable=False), sa.Column("technical_owner_person_id", person_id), sa.Column("lifecycle_provenance", sa.String(128)), 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.UniqueConstraint( "catalog_service_account_id", name="uq_dax_sa_profile_catalog", ), **table_kw_for("identity_persons"), ) normalize_fk_column_collation( PROFILE_TABLE, "principal_id", PRINCIPAL_TABLE, nullable=False, ) normalize_fk_column_collation( PROFILE_TABLE, "technical_owner_person_id", "identity_persons", nullable=True, ) _create_foreign_key_if_missing( bind, "fk_dax_sa_profile_principal", PROFILE_TABLE, PRINCIPAL_TABLE, ["principal_id"], ["id"], ondelete="CASCADE", ) _create_foreign_key_if_missing( bind, "fk_dax_sa_profile_owner", PROFILE_TABLE, "identity_persons", ["technical_owner_person_id"], ["id"], ondelete="SET NULL", ) _create_index_if_missing( bind, "ix_dax_sa_profile_owner", PROFILE_TABLE, ["technical_owner_person_id"], ) _create_index_if_missing( bind, "ix_dax_sa_profile_project_env", PROFILE_TABLE, ["gcp_project_id", "environment"], ) def _project_legacy_groups(bind) -> None: if not _table_exists(bind, "email_groups"): return bind.execute(sa.text( "INSERT INTO data_access_principals " "(id, principal_type, external_subject_id, email_snapshot, status, email_group_id, created_at, updated_at) " "SELECT UUID(), 'google_group', " "COALESCE(NULLIF(google_group_id, ''), CONCAT('email-group:', id)), " "LOWER(email), " "CASE WHEN status = 'active' THEN 'active' " "WHEN status = 'deleted' THEN 'deleted_confirmed' ELSE 'suspended' END, " "id, NOW(), NOW() " "FROM email_groups g " "WHERE NOT EXISTS (" "SELECT 1 FROM email_groups preferred " "WHERE COALESCE(NULLIF(preferred.google_group_id, ''), CONCAT('email-group:', preferred.id)) " "= COALESCE(NULLIF(g.google_group_id, ''), CONCAT('email-group:', g.id)) " "AND ((preferred.status = 'active' AND g.status <> 'active') " "OR (preferred.status = g.status AND preferred.id < g.id))" ") " "AND NOT EXISTS (" "SELECT 1 FROM data_access_principals p " "WHERE p.principal_type = 'google_group' " "AND p.external_subject_id = COALESCE(NULLIF(g.google_group_id, ''), CONCAT('email-group:', g.id))" ")", )) def _add_principal_projection_columns(bind) -> None: for table in ( "data_access_requests", "data_access_bindings", "data_access_compiled_effects", "data_access_effect_runs", ): if not _table_exists(bind, table): continue _add_column_if_missing(table, sa.Column("principal_id", id_type_for(PRINCIPAL_TABLE))) _add_column_if_missing( table, sa.Column("principal_type_snapshot", sa.String(32), nullable=False, server_default="google_group"), ) _add_column_if_missing(table, sa.Column("principal_subject_snapshot", sa.String(128))) normalize_fk_column_collation(table, "principal_id", PRINCIPAL_TABLE, nullable=True) _create_foreign_key_if_missing( bind, f"fk_{table}_principal_v2", table, PRINCIPAL_TABLE, ["principal_id"], ["id"], ondelete="RESTRICT", ) _create_index_if_missing( bind, f"ix_{table}_principal_v2", table, ["principal_id"], ) if _table_exists(bind, "data_access_compiled_effects"): _add_column_if_missing( "data_access_compiled_effects", sa.Column("provider_access_kind", sa.String(32), nullable=False, server_default="groupByEmail"), ) if _table_exists(bind, "data_access_effect_runs"): _add_column_if_missing( "data_access_effect_runs", sa.Column("provider_access_kind", sa.String(32), nullable=False, server_default="groupByEmail"), ) def _backfill_principal_projection(bind) -> None: mappings = ( ("data_access_requests", "principal_email_group_id"), ("data_access_bindings", "principal_email_group_id"), ("data_access_compiled_effects", "principal_email_group_id"), ) for table, group_column in mappings: if _table_exists(bind, table) and _column_exists(bind, table, group_column): bind.execute(sa.text( f"UPDATE {table} dax_row " f"JOIN email_groups group_row ON group_row.id = dax_row.{group_column} " "JOIN data_access_principals p ON p.principal_type = 'google_group' " "AND p.external_subject_id = COALESCE(" "NULLIF(group_row.google_group_id, ''), CONCAT('email-group:', group_row.id)) " "SET dax_row.principal_id = p.id, " "dax_row.principal_type_snapshot = 'google_group', " "dax_row.principal_subject_snapshot = p.external_subject_id " "WHERE dax_row.principal_id IS NULL", )) if _table_exists(bind, "data_access_effect_runs"): bind.execute(sa.text( "UPDATE data_access_effect_runs run " "JOIN data_access_compiled_effects effect ON effect.id = run.effect_id " "SET run.principal_id = effect.principal_id, " "run.principal_type_snapshot = effect.principal_type_snapshot, " "run.principal_subject_snapshot = effect.principal_subject_snapshot, " "run.provider_access_kind = effect.provider_access_kind " "WHERE run.principal_id IS NULL", )) def _make_legacy_group_columns_nullable(bind) -> None: for table in ( "data_access_requests", "data_access_compiled_effects", ): if ( _table_exists(bind, table) and _column_exists(bind, table, "principal_email_group_id") and not _column_is_nullable(bind, table, "principal_email_group_id") ): op.alter_column( table, "principal_email_group_id", existing_type=id_type_for("email_groups"), nullable=True, ) def _assert_no_binding_semantic_collisions(bind) -> None: """Fail before replacing the generated unique key if projections collapse. A historic duplicate Google group may project to one immutable principal. Replacing the legacy Group-id key would then make two active bindings share one semantic key. Detect that read-only, before dropping the old unique index, so the operator gets a safe, actionable migration failure instead of a partially rewritten uniqueness constraint. """ collision = bind.execute(sa.text( "SELECT 1 FROM data_access_bindings " "WHERE status IN ('active', 'revoked_pending') " "GROUP BY resource_id, COALESCE(principal_id, principal_email_group_id), " "entitlement_id HAVING COUNT(*) > 1 LIMIT 1" )).scalar() if collision: raise RuntimeError( "Cannot migrate data_access_bindings: legacy Group projections " "would collapse active semantic keys; resolve duplicate bindings first", ) def _switch_binding_semantic_key(bind) -> None: table = "data_access_bindings" if not _table_exists(bind, table): return # TiDB forbids MODIFY on a base column while a generated column references # it (error 3106). The old binding key depends on the legacy Group id, so # remove its unique index and generated column before making that Group id # nullable, then add the v2 principal-aware expression back. Every stage # is independently resumable because a failed DDL does not roll back 459. _assert_no_binding_semantic_collisions(bind) if _index_exists(bind, table, "uq_data_access_bindings_active_semantic"): op.drop_index("uq_data_access_bindings_active_semantic", table_name=table) if _column_exists(bind, table, "active_semantic_key"): op.drop_column(table, "active_semantic_key") if not _column_is_nullable(bind, table, "principal_email_group_id"): op.alter_column( table, "principal_email_group_id", existing_type=id_type_for("email_groups"), nullable=True, ) if not _column_exists(bind, table, "active_semantic_key"): op.add_column( table, sa.Column( "active_semantic_key", sa.String(64), sa.Computed( "CASE WHEN status IN ('active', 'revoked_pending') THEN " "SHA2(CONCAT_WS('|', resource_id, " "COALESCE(principal_id, principal_email_group_id), entitlement_id), 256) " "ELSE NULL END", ), nullable=True, ), ) _create_index_if_missing( bind, "uq_data_access_bindings_active_semantic", table, ["active_semantic_key"], unique=True, ) def _scope_idempotency_to_requester(bind) -> None: table = "data_access_requests" if not _table_exists(bind, table): return if _index_exists(bind, table, "uq_data_access_requests_idempotency_key"): op.drop_constraint( "uq_data_access_requests_idempotency_key", table, type_="unique", ) if not _index_exists(bind, table, "uq_data_access_requests_requester_idempotency"): op.create_unique_constraint( "uq_data_access_requests_requester_idempotency", table, ["requester_person_id", "idempotency_key"], ) def upgrade() -> None: bind = op.get_bind() _create_principal_tables(bind) _project_legacy_groups(bind) _add_principal_projection_columns(bind) _backfill_principal_projection(bind) _make_legacy_group_columns_nullable(bind) _switch_binding_semantic_key(bind) _scope_idempotency_to_requester(bind) def downgrade() -> None: # This migration changes historic semantic keys from a Group projection to # immutable DAX principal ids. Reversing it after Service Account rows # exist would either lose the rows or corrupt uniqueness, so rollback is # deliberately by deploy/gate closure rather than a destructive schema # downgrade. Fresh environments can downgrade the preceding chain. raise NotImplementedError( "459 is a non-destructive principal-identity migration; use gate rollback", )