"""seat-sync real-occupancy categorisation — stat row gets 3 new buckets + software_external_holder gets a category discriminator + person FK. Revision ID: 219 Revises: 218 Create Date: 2026-05-29 Why this exists (the 2026-05-29 prod investigation) --------------------------------------------------- The 「软件坐席对账」 drawer pre-2026-05-29 showed "365 = 318 + 46 + 0" (off by 1 — the alias-collapse bug fixed in the prior commit) but ALSO hid the most operationally-important truth: **the 365 in Helix didn't match the 372 Atlassian's Admin UI billed for.** The 7-account gap was silently absorbed by the existing filters: * ``is_active=True`` filter dropped vendor accounts whose status is ``inactive`` but who still occupy a billable seat (prod: lifu.wu, a recently-departed person whose HRIS-driven Helix offboarding hasn't propagated yet but Atlassian already deactivated); * ``production_sites`` filter dropped accounts holding the product on a non-paid site (sandbox / personal sites: prod has qizhi.wang on hibernate.atlassian.net, yanyanqing@ on yanyanqing.atlassian.net); * Even when matched to a Helix employee whose ``employment_status`` is ``left`` (offboarding leak — the seat should've been revoked but didn't), the seat was bucketed into ``holders_matched`` and looked exactly like a healthy in-use seat. User feedback (2026-05-29): "我要求显示和 jira 实际一样的数字 显示出来 真的占了坐席的". The fix: count **every Atlassian account holding the product family across the org** (no status / no site filter — matches what the vendor bills), and classify each into one of 6 mutually- exclusive buckets so the operator can see exactly who's taking up each seat. Schema changes -------------- **software_seat_sync_stat** — add 3 counter columns to the existing 4 so the headline identity becomes ``vendor_license_holders = holders_matched + holders_external + holders_no_email + holders_leaver_residual + holders_vendor_inactive + holders_off_paid_site``. All NOT NULL DEFAULT 0 so existing rows keep their existing values; the next ``*.seat_sync`` run rewrites them with the new semantics. **software_external_holder** — was strictly "external (no Helix match) per (plan, email)". Generalise to "anomalous holder of any category per (plan, email)" by adding: * ``category`` VARCHAR(32) NOT NULL DEFAULT ``'external'`` — one of {``external`` | ``leaver_residual`` | ``vendor_inactive`` | ``off_paid_site``}. The unique constraint stays on (plan_id, email) because the same email can transition between categories across runs (an external resolved to a Helix person who has since become a leaver → external row's category flips to ``leaver_residual``). * ``person_id`` CHAR(36) NULL FK ``identity_persons.id`` — populated when the holder matched a Helix person via the alias lookup (NULL for ``category='external'``). Lets the drawer deep-link from a leaver_residual row straight to the person's profile / offboarding process. Idempotency ----------- Each ``ADD COLUMN`` guarded via ``information_schema.COLUMNS`` probe. ``person_id`` FK created only when both the column AND its index can be added cleanly — re-runs are safe. """ from __future__ import annotations import sqlalchemy as sa from alembic import op revision = "219" down_revision = "218" branch_labels = None depends_on = None def _column_exists(bind, table: str, column: str) -> bool: return 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() is not None def _index_exists(bind, table: str, index: str) -> bool: return 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() 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). Copy of the helper in migration 192/210 — duplicated here so this migration is self-contained (alembic upgrades don't share helpers across revision files). """ 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() # ── software_seat_sync_stat: 3 new bucket columns ───────────── for col_name in ( "holders_leaver_residual", "holders_vendor_inactive", "holders_off_paid_site", ): if not _column_exists(bind, "software_seat_sync_stat", col_name): op.add_column( "software_seat_sync_stat", sa.Column( col_name, sa.Integer(), nullable=False, server_default=sa.text("0"), ), ) # ── software_external_holder: category + person_id ───────────── if not _column_exists(bind, "software_external_holder", "category"): op.add_column( "software_external_holder", sa.Column( "category", sa.String(32), nullable=False, server_default=sa.text("'external'"), ), ) if not _column_exists(bind, "software_external_holder", "person_id"): person_col_type = _fk_source_col_type( bind, "identity_persons", "id", ) op.add_column( "software_external_holder", sa.Column( "person_id", person_col_type, sa.ForeignKey( "identity_persons.id", ondelete="SET NULL", ), nullable=True, ), ) # Index on (plan_id, category) for the drawer's per-category # filter — the existing (plan_id, resolved) index doesn't cover # "all leaver_residual on this plan" queries. if not _index_exists( bind, "software_external_holder", "ix_software_external_holder_plan_category", ): op.create_index( "ix_software_external_holder_plan_category", "software_external_holder", ["plan_id", "category"], ) # Index on person_id for the reverse lookup ("which seats does # this person hold residually after offboarding?") — used by the # offboarding service's per-person leaver-residual badge. if not _index_exists( bind, "software_external_holder", "ix_software_external_holder_person_id", ): op.create_index( "ix_software_external_holder_person_id", "software_external_holder", ["person_id"], ) def downgrade() -> None: bind = op.get_bind() if _index_exists( bind, "software_external_holder", "ix_software_external_holder_person_id", ): op.drop_index( "ix_software_external_holder_person_id", table_name="software_external_holder", ) if _index_exists( bind, "software_external_holder", "ix_software_external_holder_plan_category", ): op.drop_index( "ix_software_external_holder_plan_category", table_name="software_external_holder", ) # Drop FK first then column (MySQL needs explicit FK drop). if _column_exists(bind, "software_external_holder", "person_id"): fk_row = bind.execute( sa.text( "SELECT CONSTRAINT_NAME FROM information_schema." "KEY_COLUMN_USAGE WHERE TABLE_SCHEMA = DATABASE() " "AND TABLE_NAME = 'software_external_holder' " "AND COLUMN_NAME = 'person_id' " "AND REFERENCED_TABLE_NAME IS NOT NULL LIMIT 1" ), ).fetchone() if fk_row: op.drop_constraint( fk_row[0], "software_external_holder", type_="foreignkey", ) op.drop_column("software_external_holder", "person_id") if _column_exists(bind, "software_external_holder", "category"): op.drop_column("software_external_holder", "category") for col_name in ( "holders_off_paid_site", "holders_vendor_inactive", "holders_leaver_residual", ): if _column_exists(bind, "software_seat_sync_stat", col_name): op.drop_column("software_seat_sync_stat", col_name)