"""identity_sync_state: auto_included_reason + notified_at + Navan preset rewrite. Revision ID: 202 Revises: 201 Create Date: 2026-05-26 Two changes, both in service of the Navan optimisation PR (PingCAP's Navan tenant only enrols foreign employees, but their managers may be domestic — we auto-pull those managers in and notify them once on creation): 1. ``identity_sync_state.auto_included_reason VARCHAR(64) NULL`` Set to ``'manager_of_foreign'`` for state rows the sweep created because the person is the direct manager of an in-scope foreign employee. Sweep treats these as permanently in-scope: they never transition to the out-of-scope / deactivate path even if the foreign report leaves or transfers. Free-form column (not enum) so future auto-include reasons (e.g. ``'finance_partner_of_country'``) can be added without another migration. 2. ``identity_sync_state.notified_at DATETIME NULL`` When a CN manager is first auto-included AND the upsert succeeds, the sweep enqueues a Feishu notification to that manager explaining why the account was created. ``notified_at`` is set at enqueue time to prevent re-notification on subsequent sweeps. NULL = no notification ever sent for this state row. 3. **Data migration**: rewrite the three Navan-specific rules in the ``mapping_config`` of every existing ``kind='navan'`` target so the already-live target doesn't keep pushing the old shape. Targeted per-rule edit (not a full-config overwrite) so admin's other customisations on the same row are preserved: * Drop the rule with ``target_field='title'`` (job_title pull-out) * Replace ``enterprise.department`` source with ``{department_combined}`` * Add ``strip_company_prefix`` transform to ``enterprise.division`` Idempotent — re-running on a DB whose Navan target already carries the new rule shape is a no-op (the helper functions check before writing). """ from __future__ import annotations import json import sqlalchemy as sa from alembic import op revision = "202" down_revision = "201" branch_labels = None depends_on = None _ENT_DEPT = ( "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User.department" ) _ENT_DIV = ( "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User.division" ) def _column_exists(bind, table: str, column: str) -> bool: row = 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}, ).fetchone() return row is not None def _rewrite_navan_rules(rules: list[dict]) -> tuple[list[dict], bool]: """Apply the three Navan-specific edits to a mapping rule list. Returns ``(new_rules, changed)``. ``changed`` is False when the list already matches the post-migration shape — so re-running this migration on a partially-migrated env is a no-op for that target. """ changed = False out: list[dict] = [] for r in rules: if not isinstance(r, dict): out.append(r) continue tgt = r.get("target_field") # 1. Drop job_title if tgt == "title": changed = True continue # 2. department source rewrite if tgt == _ENT_DEPT: if r.get("source_template") != "{department_combined}": changed = True out.append({**r, "source_template": "{department_combined}"}) continue # 3. division add prefix-strip transform if tgt == _ENT_DIV: existing_xf = r.get("transform") if existing_xf == "strip_company_prefix" or ( isinstance(existing_xf, list) and "strip_company_prefix" in existing_xf ): out.append(r) continue changed = True new_xf: list[str] if existing_xf is None or existing_xf == []: new_xf = ["strip_company_prefix"] elif isinstance(existing_xf, str): new_xf = [existing_xf, "strip_company_prefix"] elif isinstance(existing_xf, list): new_xf = [*existing_xf, "strip_company_prefix"] else: # Unknown shape — preserve, don't crash the migration. out.append(r) continue new_rule = dict(r) new_rule["transform"] = new_xf out.append(new_rule) continue out.append(r) return out, changed def upgrade() -> None: bind = op.get_bind() # ── Schema: two nullable columns. ──────────────────────────────── if not _column_exists(bind, "identity_sync_state", "auto_included_reason"): op.add_column( "identity_sync_state", sa.Column( "auto_included_reason", sa.String(64), nullable=True, comment=( "Free-form reason a state row was auto-pulled into " "scope by the sweep (e.g. 'manager_of_foreign'). NULL " "for normal filter-matched rows. Auto-included rows " "skip the out-of-scope deactivate path." ), ), ) # Filtered index — only the auto-included subset is queried. op.create_index( "ix_identity_sync_state_auto_included", "identity_sync_state", ["target_id", "auto_included_reason"], ) if not _column_exists(bind, "identity_sync_state", "notified_at"): op.add_column( "identity_sync_state", sa.Column( "notified_at", sa.DateTime(timezone=True), nullable=True, comment=( "Time the first-create notification was enqueued. " "NULL = never sent. Set once and never cleared so " "re-renders of the same state row do not re-notify." ), ), ) # ── Data: rewrite Navan target mapping_config rows. ────────────── # ``IdentitySyncTarget.mapping_config`` is a JSON column with # ``{"rules": [...]}`` shape. We pull every kind='navan' target, # apply the three targeted edits, and write back only when the # result actually differs from the stored config. rows = bind.execute( sa.text( "SELECT id, mapping_config FROM identity_sync_targets " "WHERE kind = 'navan' AND deleted_at IS NULL" ) ).fetchall() for row in rows: target_id = row[0] mc = row[1] if isinstance(mc, str): try: mc = json.loads(mc) except (TypeError, ValueError): # Unparseable config — leave alone, admin will need to # re-pick the preset. continue if not isinstance(mc, dict): continue rules = mc.get("rules") or [] if not isinstance(rules, list): continue new_rules, changed = _rewrite_navan_rules(rules) if not changed: continue new_mc = dict(mc) new_mc["rules"] = new_rules bind.execute( sa.text( "UPDATE identity_sync_targets " "SET mapping_config = :mc " "WHERE id = :id" ), {"mc": json.dumps(new_mc), "id": target_id}, ) # Bust per-person payload-hash cache so the next sweep actually # re-pushes with the new shape. Matches the eager invalidation # in ``IdentitySyncService.update_target``. bind.execute( sa.text( "UPDATE identity_sync_state " "SET last_payload_hash = NULL " "WHERE target_id = :id" ), {"id": target_id}, ) def downgrade() -> None: bind = op.get_bind() if _column_exists(bind, "identity_sync_state", "auto_included_reason"): op.drop_index( "ix_identity_sync_state_auto_included", table_name="identity_sync_state", ) op.drop_column("identity_sync_state", "auto_included_reason") if _column_exists(bind, "identity_sync_state", "notified_at"): op.drop_column("identity_sync_state", "notified_at") # Mapping-config rewrites are not reversed — admins who downgrade # can re-pick the preset from the UI.