"""Add ``cmdb_resource_creators`` — per-resource attribution from cloud-provider audit logs (CloudTrail / Cloud Audit Logs / ActionTrail / Activity Log). The cost dashboard's `tags` column (added 2026-05-22) surfaces *static* attribution from billing data — ``Owner=alice`` only works when the org enforces a tagging policy. Resources created in the console by an engineer who forgot to tag, or AWS-managed resources (CloudFormation-generated stacks, auto-scaling launches) carry no user tags. For those, the source of truth is the audit log: * AWS CloudTrail ``LookupEvents`` → ``userIdentity.arn`` on ``RunInstances`` / ``CreateBucket`` / etc. * GCP Cloud Audit Logs → ``protoPayload.authenticationInfo. principalEmail`` on the matching insert/create method. * Aliyun ActionTrail ``LookupEvents`` → ``UserIdentity.UserName``. * Azure Activity Log → ``caller`` on ``Microsoft.*/write``. Each lookup is rate-limited (CloudTrail 2 req/s, others similar) and the audit window is provider-dependent (AWS / Aliyun / Azure default 90 days; GCP 400 days). Resolving on every dashboard render would slam the audit APIs and burn through rate budgets. This table **caches** the resolution so: 1. First resolve attempt for a resource → API call → row written. 2. Subsequent reads → cached row, zero API cost. 3. Cron sweep (``cmdb.resource_creator_sweep``) walks ``cmdb_cost_facts.resource_id`` for newly-seen resources and pre-resolves so the dashboard drawer is never the first hit. Schema: * ``id BIGINT AUTO_INCREMENT`` — same rationale as ``cmdb_cost_facts``: the table grows ~one row per resource ever discovered, easily into the millions on multi-account orgs; UUID PKs would add 24 bytes per row of index space for no analytical benefit. * ``UNIQUE(integration_id, resource_id)`` — the natural key. Cloud resource IDs aren't globally unique (two Aliyun tenants can both have ``i-bp1...``); the integration prefix disambiguates. * ``status`` is a 6-state enum stored as VARCHAR (not native ENUM) so adding new states later doesn't require an ALTER TYPE on TiDB: ``pending`` (row created, attempt not yet made) / ``resolved`` (creator found and persisted) / ``not_found`` (audit log returned zero events for this resource — typically means it pre-dates the audit window) / ``window_exceeded`` (explicit provider signal that the lookup window was the cause) / ``error`` (transient API failure — retry up to attempt_count cap) / ``unsupported`` (driver doesn't implement get_resource_creator for this provider yet; reserved for future drivers). * ``source`` records which audit API produced the answer (``cloudtrail`` / ``gcp_audit`` / ``aliyun_actiontrail`` / ``azure_activity``). Helps debug a mis-attributed row when an operator wants to know "which API did this come from." * ``attempt_count`` + ``last_attempt_at`` + ``last_error`` are the retry-with-backoff state. Cron skips rows with ``attempt_count >= 5`` to avoid retry storms on perma-missing resources. * ``raw JSON`` carries the full audit event for forensics. Capped in Python at ~16 KiB before persist (same as ``cmdb_cost_facts.raw``). * ``creator_person_id`` FK with ``ondelete=SET NULL`` — hard-deleting a Person row (orphan cleanup, GDPR right-to-be-forgotten) blanks this column without losing the resource attribution itself (``creator_email`` / ``creator_principal_id`` survive). Matches the ``cost_owner_person_id`` pattern on ``cmdb_cost_facts``. * ``integration_id`` FK with ``ondelete=RESTRICT`` — same reasoning as ``cmdb_cost_facts``: a ``DELETE FROM integrations`` would silently wipe attribution history. Operators soft-disable via ``Integration.enabled=False``. Indexes: * ``ix_resource_creators_sweep`` ``(status, last_attempt_at)`` — drives the cron sweep's "fetch next batch to retry" query. Leading with status keeps the index range tight (most rows are ``resolved`` and won't appear in the cron's queue). * ``ix_resource_creators_person`` ``(creator_person_id, resolved_at)`` — reverse lookup "what did Alice create?" for the Person detail page (future surface; the column exists now so the index is ready). * ``ix_resource_creators_email`` ``(creator_email)`` — same reverse lookup keyed on email (used when person_id is NULL because the IAM ARN didn't match any Helix Person). Revision ID: 198 Revises: 197 Create Date: 2026-05-22 """ from __future__ import annotations import sqlalchemy as sa from alembic import op from app.core.migration_helpers import normalize_fk_column_collation # revision identifiers, used by Alembic. revision = "198" down_revision = "197" branch_labels = None depends_on = None def _table_exists(bind, name: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.TABLES " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :n" ), {"n": name}, ).scalar() is not None def _index_exists(bind, table: str, name: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.STATISTICS " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :t " "AND INDEX_NAME = :n" ), {"t": table, "n": name}, ).scalar() is not None def upgrade() -> None: bind = op.get_bind() if not _table_exists(bind, "cmdb_resource_creators"): op.create_table( "cmdb_resource_creators", sa.Column( "id", sa.BigInteger, primary_key=True, autoincrement=True, ), sa.Column("integration_id", sa.String(36), nullable=False), # Mirrors ``cmdb_cost_facts.resource_id`` length (512) so the # FK-by-value join doesn't truncate Azure ARM ids # (~200 chars on Foundry accounts). sa.Column("resource_id", sa.String(512), nullable=False), sa.Column("resource_type", sa.String(128), nullable=True), # IAM ARN / SSO ARN / UPN / Aliyun RAM user id — driver- # opaque string carrying the provider-native principal. # Stored even when ``creator_email`` is NULL so an # operator can manually map federated-identity ARNs to # people later. sa.Column("creator_principal_id", sa.String(512), nullable=True), sa.Column("creator_email", sa.String(255), nullable=True), sa.Column("creator_display_name", sa.String(255), nullable=True), sa.Column("creator_person_id", sa.String(36), nullable=True), # When the resource was created per the audit log # (NOT when Helix ingested the row). The two diverge by # the audit-API call latency + the cron sweep cadence. sa.Column( "created_at_cloud", sa.DateTime(timezone=True), nullable=True, ), sa.Column( "resolved_at", sa.DateTime(timezone=True), nullable=True, ), # ``cloudtrail`` / ``gcp_audit`` / ``aliyun_actiontrail`` / # ``azure_activity``. VARCHAR not ENUM so future providers # don't require an ALTER TYPE on TiDB. sa.Column("source", sa.String(32), nullable=False), # 6-state status — see module docstring. sa.Column( "status", sa.String(16), nullable=False, server_default="pending", ), sa.Column( "attempt_count", sa.Integer, nullable=False, server_default=sa.text("0"), ), sa.Column( "last_attempt_at", sa.DateTime(timezone=True), nullable=True, ), sa.Column("last_error", sa.Text, nullable=True), sa.Column("raw", sa.JSON, nullable=True), sa.Column( "created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False, ), sa.Column( "updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), onupdate=sa.func.now(), nullable=False, ), sa.UniqueConstraint( "integration_id", "resource_id", name="uq_resource_creators_integ_rid", ), mysql_engine="InnoDB", mysql_charset="utf8mb4", mysql_collate="utf8mb4_general_ci", ) # Collation normalization BEFORE FK creation so TiDB accepts # the cross-table reference (memory: # ``feedback_tidb_fk_strictness``). normalize_fk_column_collation( table="cmdb_resource_creators", column="integration_id", parent_table="integrations", nullable=False, ) normalize_fk_column_collation( table="cmdb_resource_creators", column="creator_person_id", parent_table="identity_persons", nullable=True, ) op.create_foreign_key( "fk_resource_creators_integration", source_table="cmdb_resource_creators", referent_table="integrations", local_cols=["integration_id"], remote_cols=["id"], ondelete="RESTRICT", ) op.create_foreign_key( "fk_resource_creators_person", source_table="cmdb_resource_creators", referent_table="identity_persons", local_cols=["creator_person_id"], remote_cols=["id"], ondelete="SET NULL", ) if not _index_exists( bind, "cmdb_resource_creators", "ix_resource_creators_sweep", ): op.create_index( "ix_resource_creators_sweep", "cmdb_resource_creators", ["status", "last_attempt_at"], ) if not _index_exists( bind, "cmdb_resource_creators", "ix_resource_creators_person", ): op.create_index( "ix_resource_creators_person", "cmdb_resource_creators", ["creator_person_id", "resolved_at"], ) if not _index_exists( bind, "cmdb_resource_creators", "ix_resource_creators_email", ): op.create_index( "ix_resource_creators_email", "cmdb_resource_creators", ["creator_email"], ) def downgrade() -> None: bind = op.get_bind() if _table_exists(bind, "cmdb_resource_creators"): op.drop_table("cmdb_resource_creators")