"""Unified provisioning + network-asset ownership (additive). Revision ID: 039 Revises: 038 Create Date: 2026-04-21 Additive schema for the unified-connectors + provisioning-completeness PR. Everything here is backward-compatible — no drops, no renames — so prod deploy is safe to run before the code that *uses* these columns is released. Drops happen later in migration 041. Changes ------- 1. ``software_license_plans.max_seats`` (Integer, nullable) Quota tracking per seat plan. NULL = unlimited. Enforced in provisioning service when a grant would exceed the limit. 2. ``cmdb_dns_records.owner_person_id`` (CHAR(36), FK → identity_persons.id) Turns the previously-raw ``owner_raw`` string into a real FK so DNS records can participate in offboarding force-transfer. The old column stays for now — backfilled below, dropped in 041. 3. ``core_idempotency_records`` (new table) Backs the ``IdempotencyService`` defined in ``app/core/idempotency.py``. Stores (key, scope) reservations for POST / PATCH / DELETE endpoints that opt in to ``Idempotency-Key``. 4. Indexes for offboarding-scan performance: - ``cmdb_dns_records(owner_person_id)`` - ``cmdb_domains(owner_person_id)`` (if not already indexed) - ``cmdb_servers(owner_person_id)`` (if not already indexed) 5. Backfill pass: for every non-NULL ``owner_raw`` on domains / servers / dns records, try to match it against ``identity_persons.work_email`` (case-insensitive) or ``display_name``. On a unique match, populate ``owner_person_id``. Ambiguous / no-match rows are left NULL with a warning counter. """ 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 = "039" down_revision = "038" branch_labels = None depends_on = None # ─── Introspection helpers (idempotent re-runs) ──────────────────── 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 _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", ), {"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", ), {"t": table, "i": index}, ).scalar() is not None def _fk_exists(bind, table: str, constraint: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.TABLE_CONSTRAINTS " "WHERE TABLE_SCHEMA = DATABASE() " "AND TABLE_NAME = :t AND CONSTRAINT_NAME = :c " "AND CONSTRAINT_TYPE = 'FOREIGN KEY'", ), {"t": table, "c": constraint}, ).scalar() is not None # ─── Migration ───────────────────────────────────────────────────── def upgrade() -> None: bind = op.get_bind() # 1. software_license_plans.max_seats if _column_exists(bind, "software_license_plans", "max_seats") is False: op.add_column( "software_license_plans", sa.Column("max_seats", sa.Integer, nullable=True), ) # 2. cmdb_dns_records.owner_person_id (+ FK + index) if _column_exists(bind, "cmdb_dns_records", "owner_person_id") is False: op.add_column( "cmdb_dns_records", sa.Column("owner_person_id", id_type_for("identity_persons"), nullable=True), ) normalize_fk_column_collation( "cmdb_dns_records", "owner_person_id", "identity_persons", ) if _fk_exists(bind, "cmdb_dns_records", "fk_cmdb_dns_records_owner_person_id") is False: op.create_foreign_key( "fk_cmdb_dns_records_owner_person_id", "cmdb_dns_records", "identity_persons", ["owner_person_id"], ["id"], ondelete="SET NULL", ) if _index_exists(bind, "cmdb_dns_records", "ix_cmdb_dns_records_owner_person_id") is False: op.create_index( "ix_cmdb_dns_records_owner_person_id", "cmdb_dns_records", ["owner_person_id"], ) # 3. Helpful indexes on existing owner_person_id columns (for the # offboarding force-transfer scan). create_index is NOT idempotent # on MySQL, so guard each one. for table in ("cmdb_domains", "cmdb_servers"): idx = f"ix_{table}_owner_person_id" if ( _column_exists(bind, table, "owner_person_id") and not _index_exists(bind, table, idx) ): op.create_index(idx, table, ["owner_person_id"]) # 4. core_idempotency_records if not _table_exists(bind, "core_idempotency_records"): op.create_table( "core_idempotency_records", sa.Column("id", id_type_for("identity_persons"), primary_key=True), sa.Column("key", sa.String(128), nullable=False), sa.Column("scope", sa.String(64), nullable=False), sa.Column( "state", sa.String(16), nullable=False, server_default="pending", ), sa.Column("resource_id", id_type_for("identity_persons"), nullable=True), sa.Column("response_status", sa.Integer, nullable=True), sa.Column("response_body_json", sa.Text, nullable=True), sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False), 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("key", "scope", name="uq_idempotency_key_scope"), **table_kw_for("identity_persons"), ) op.create_index( "ix_core_idempotency_records_expires_at", "core_idempotency_records", ["expires_at"], ) # 5. Backfill owner_person_id from owner_raw on DNS / domain / server. # Match policy (most specific first): # a. exact work_email match (case-insensitive) # b. single non-deleted person whose display_name == owner_raw # Ambiguous matches (multiple active persons w/ same name) are # left NULL so the admin can resolve them manually via the new # offboarding console. for table in ("cmdb_dns_records", "cmdb_domains", "cmdb_servers"): if not _column_exists(bind, table, "owner_raw"): continue if not _column_exists(bind, table, "owner_person_id"): # Older schema; the FK column isn't in place yet. Skip # (will be added by this same migration above, but if an # earlier partial run already backfilled we're fine). continue # --- email match --- bind.execute(sa.text(f""" UPDATE {table} t JOIN identity_persons p ON LOWER(t.owner_raw) = LOWER(p.work_email) AND p.deleted_at IS NULL SET t.owner_person_id = p.id WHERE t.owner_person_id IS NULL AND t.owner_raw IS NOT NULL AND t.owner_raw != '' """)) # --- display_name unique match --- bind.execute(sa.text(f""" UPDATE {table} t JOIN ( SELECT display_name, MIN(id) AS only_id, COUNT(*) AS c FROM identity_persons WHERE deleted_at IS NULL AND display_name IS NOT NULL GROUP BY display_name HAVING c = 1 ) p ON t.owner_raw = p.display_name SET t.owner_person_id = p.only_id WHERE t.owner_person_id IS NULL AND t.owner_raw IS NOT NULL AND t.owner_raw != '' """)) def downgrade() -> None: bind = op.get_bind() # Drop in reverse order if _table_exists(bind, "core_idempotency_records"): op.drop_table("core_idempotency_records") for table in ("cmdb_domains", "cmdb_servers"): idx = f"ix_{table}_owner_person_id" if _index_exists(bind, table, idx): op.drop_index(idx, table_name=table) if _index_exists(bind, "cmdb_dns_records", "ix_cmdb_dns_records_owner_person_id"): op.drop_index( "ix_cmdb_dns_records_owner_person_id", table_name="cmdb_dns_records", ) if _fk_exists(bind, "cmdb_dns_records", "fk_cmdb_dns_records_owner_person_id"): op.drop_constraint( "fk_cmdb_dns_records_owner_person_id", "cmdb_dns_records", type_="foreignkey", ) if _column_exists(bind, "cmdb_dns_records", "owner_person_id"): op.drop_column("cmdb_dns_records", "owner_person_id") if _column_exists(bind, "software_license_plans", "max_seats"): op.drop_column("software_license_plans", "max_seats")