"""software_external_holder + software_seat_sync_stat — seat reconciliation. Revision ID: 210 Revises: 209 Create Date: 2026-05-28 Why this exists --------------- Before this, the "vendor bills N license holders but Helix only shows M employee seats" gap lived ONLY as a capped, email-only sample inside each ``*.seat_sync`` run's ``result_summary`` JSON — per-connector, inconsistent shape, no drill-down, and Slack didn't even compute it. The UI surface was a raw JSON blob buried in the scheduled-job runs page. This migration adds the two tables that turn the gap into a first-class, explorable reconciliation surface across all five seat-sync vendors (Jira/JSM/Confluence, Zoom, Google, Feishu, Slack): * ``software_external_holder`` — one row per (plan, external email): a vendor license holder that matched no Helix employee. Carries vendor account id / status / last-active so the per-software detail drawer can drill down and deep-link to the vendor admin console. * ``software_seat_sync_stat`` — one row per plan: the authoritative headline snapshot (vendor_license_holders / matched / external / no_email) + ``synced_at`` provenance for the "72 vs 69 vs 3" display. Schema notes ------------ * All UUID FKs (``integration_id`` / ``software_id`` / ``plan_id``) mirror the target column's data type + collation at runtime via ``_fk_source_col_type`` — same TiDB strict-FK guard established by migrations 191/192. A naive ``sa.CHAR(36)`` inherits the new table's connection-time collation, which TiDB rejects with (3780, "Referencing column ... incompatible."). * ``run_id`` on the stat table is a bare UUID (no FK) — ``scheduled_job_runs`` rows are pruned on their own retention clock and we don't want a snapshot to vanish when its originating run is pruned. Idempotency ----------- Each table creation guarded via ``information_schema.TABLES`` probe. Safe to re-run. """ from __future__ import annotations import sqlalchemy as sa from alembic import op revision = "210" down_revision = "209" branch_labels = None depends_on = None def _table_exists(bind, table: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.TABLES " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :t LIMIT 1" ), {"t": table}, ).scalar() is not None def _fk_source_col_type( bind, target_table: str, target_column: str, *, fallback_type: str = "char", fallback_collation: str | None = "utf8mb4_unicode_ci", length: int = 36, ): """Mirror the target FK column's data type + collation (TiDB strict-FK). See migration 192's docstring for the full failure mode — a new table's ``sa.CHAR(36)`` inherits the connection collation and TiDB rejects FK creation with (3780, incompatible) when it differs from the ancient target column's ``utf8mb4_unicode_ci``. """ row = bind.execute( sa.text( "SELECT DATA_TYPE, COLLATION_NAME " "FROM information_schema.COLUMNS " "WHERE TABLE_SCHEMA = DATABASE() " " AND TABLE_NAME = :t AND COLUMN_NAME = :c" ), {"t": target_table, "c": target_column}, ).fetchone() if row: data_type = (row[0] or fallback_type).lower() collation = row[1] or fallback_collation else: data_type = fallback_type.lower() collation = fallback_collation type_cls = sa.CHAR if data_type == "char" else sa.String if collation: return type_cls(length, collation=collation) return type_cls(length) def upgrade() -> None: bind = op.get_bind() if not _table_exists(bind, "software_external_holder"): integration_col = _fk_source_col_type(bind, "integrations", "id") software_col = _fk_source_col_type(bind, "asset_software_catalog", "id") plan_col = _fk_source_col_type(bind, "software_license_plans", "id") op.create_table( "software_external_holder", sa.Column("id", sa.CHAR(36), primary_key=True), sa.Column( "integration_id", integration_col, sa.ForeignKey("integrations.id"), nullable=False, ), sa.Column( "software_id", software_col, sa.ForeignKey("asset_software_catalog.id"), nullable=False, ), sa.Column( "plan_id", plan_col, sa.ForeignKey("software_license_plans.id"), nullable=False, ), sa.Column("provider", sa.String(32), nullable=False), sa.Column("product_key", sa.String(64), nullable=True), sa.Column("site", sa.String(255), nullable=True), sa.Column("email", sa.String(255), nullable=False), sa.Column("display_name", sa.String(255), nullable=True), sa.Column("vendor_account_id", sa.String(128), nullable=True), sa.Column("account_status", sa.String(32), nullable=True), sa.Column("last_active_on", sa.Date(), nullable=True), sa.Column( "first_observed_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column( "last_observed_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column( "resolved", sa.Boolean(), nullable=False, server_default=sa.text("0"), ), sa.Column("resolved_at", sa.DateTime(timezone=True), 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.Index( "uq_software_external_holder_plan_email", "plan_id", "email", unique=True, ), sa.Index( "ix_software_external_holder_software_resolved", "software_id", "resolved", ), sa.Index( "ix_software_external_holder_plan_resolved", "plan_id", "resolved", ), sa.Index( "ix_software_external_holder_integration_id", "integration_id", ), ) if not _table_exists(bind, "software_seat_sync_stat"): integration_col = _fk_source_col_type(bind, "integrations", "id") software_col = _fk_source_col_type(bind, "asset_software_catalog", "id") plan_col = _fk_source_col_type(bind, "software_license_plans", "id") op.create_table( "software_seat_sync_stat", sa.Column("id", sa.CHAR(36), primary_key=True), sa.Column( "integration_id", integration_col, sa.ForeignKey("integrations.id"), nullable=False, ), sa.Column( "software_id", software_col, sa.ForeignKey("asset_software_catalog.id"), nullable=False, ), sa.Column( "plan_id", plan_col, sa.ForeignKey("software_license_plans.id"), nullable=False, ), sa.Column("provider", sa.String(32), nullable=False), sa.Column("product_key", sa.String(64), nullable=True), sa.Column( "vendor_license_holders", sa.Integer(), nullable=False, server_default=sa.text("0"), ), sa.Column( "holders_matched", sa.Integer(), nullable=False, server_default=sa.text("0"), ), sa.Column( "holders_external", sa.Integer(), nullable=False, server_default=sa.text("0"), ), sa.Column( "holders_no_email", sa.Integer(), nullable=False, server_default=sa.text("0"), ), sa.Column( "synced_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column("run_id", sa.CHAR(36), 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.Index( "uq_software_seat_sync_stat_plan", "plan_id", unique=True, ), sa.Index("ix_software_seat_sync_stat_software", "software_id"), sa.Index( "ix_software_seat_sync_stat_integration_id", "integration_id", ), ) def downgrade() -> None: bind = op.get_bind() if _table_exists(bind, "software_seat_sync_stat"): op.drop_table("software_seat_sync_stat") if _table_exists(bind, "software_external_holder"): op.drop_table("software_external_holder")