"""Lifecycle automation: notification outbox, compliance incidents, IT-info push. Revision ID: 038 Revises: 037 Create Date: 2026-04-20 Makes the HRIS → platform → employee loop production-grade: 1. ``notification_outbox`` — every outbound dispatch writes one row here FIRST, then tries each channel (feishu / email / ...). Per-channel delivery tracked; failures retry hourly with exponential backoff. Guarantees "nothing silently drops" — every send has an audit row and a retry path. 2. ``compliance_incidents`` — when post-leaver rescan finds a resource still pointing at someone whose ``employment_status = 'left'``, a row is created here. This is the hard safeguard behind the platform commitment "no orphan resources after leaver". 3. ``identity_persons`` gains ``it_info_sent_at`` + ``it_info_sent_channels`` so the hire-day push handler is idempotent. 4. Drop unused tables ``wf_instances`` / ``wf_tasks`` / ``wf_leaver_cases`` (migration 001 artefacts, never wired) and the stale columns ``hr_lifecycle_events.workflow_id`` and ``offboarding_processes.summary``. 5. Seed five new scheduled_jobs rows: - ``hr.hire_day_it_notify`` (daily 09:00 Asia/Shanghai) - ``hr.process_advance`` (every 15 min) — T+0 force transfer + atomic transition to ``left`` - ``hr.post_leaver_verify`` (daily 02:00) — T+1 rescan - ``hr.daily_digest`` (daily 09:00) — IT admin morning summary - ``notification.outbox_retry`` (hourly) — reship failed dispatches """ from __future__ import annotations import sqlalchemy as sa from alembic import op from app.core.migration_helpers import id_type_for, table_kw_for revision = "038" down_revision = "037" 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 _col_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 _safe_drop_wf_tables(bind) -> None: """Drop the 3 unused ``wf_*`` legacy tables from migration 001. On TiDB prod an out-of-codebase table ``wf_offboarding_checklist_items`` FK-references ``wf_leaver_cases``, so a bare DROP errors with 3730. We: 1. Query ``information_schema.REFERENTIAL_CONSTRAINTS`` for every inbound FK on each target and drop those constraints. 2. DROP TABLE IF EXISTS — any still-blocked table is logged and skipped (cleanup is cosmetic; leaving a dead table in place is fine). """ import logging log = logging.getLogger("alembic") for tbl in ("wf_leaver_cases", "wf_tasks", "wf_instances"): # Must check existence *on the bind* — the table may be absent # in dev/CI DBs where migration 001 rolled back or was stripped. exists = bind.execute( sa.text( "SELECT 1 FROM information_schema.TABLES " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :n" ), {"n": tbl}, ).scalar() is not None if not exists: continue # Drop inbound FK constraints try: rows = bind.execute( sa.text( "SELECT TABLE_NAME, CONSTRAINT_NAME " "FROM information_schema.REFERENTIAL_CONSTRAINTS " "WHERE CONSTRAINT_SCHEMA = DATABASE() " "AND REFERENCED_TABLE_NAME = :t" ), {"t": tbl}, ).fetchall() except Exception: # noqa: BLE001 rows = [] for child_table, fk_name in rows: try: op.drop_constraint(fk_name, child_table, type_="foreignkey") log.info( "dropped inbound FK %s on %s (was referencing %s)", fk_name, child_table, tbl, ) except Exception as exc: # noqa: BLE001 log.warning( "failed to drop FK %s on %s (may block DROP %s): %s", fk_name, child_table, tbl, exc, ) # Now try to drop the unused table itself try: op.execute(f"DROP TABLE IF EXISTS `{tbl}`") except Exception as exc: # noqa: BLE001 log.warning("could not drop unused wf table %s: %s", tbl, exc) def upgrade() -> None: """Idempotent — re-entrant after partial failure. MySQL/TiDB DDL isn't transactional, so earlier failures of this migration (error 3780 on FK creation before the charset-aware rewrite) may leave the schema in a half-applied state. Guard each CREATE / ADD with information_schema checks. """ bind = op.get_bind() person_id_type = id_type_for("identity_persons") person_table_kw = table_kw_for("identity_persons") # ── notification_outbox ───────────────────────────────────── if not _table_exists(bind, "notification_outbox"): _create_notification_outbox(person_id_type, person_table_kw) # ── compliance_incidents ──────────────────────────────────── if not _table_exists(bind, "compliance_incidents"): _create_compliance_incidents(person_id_type, person_table_kw) # ── identity_persons: IT-info idempotency ────────────────── if not _col_exists(bind, "identity_persons", "it_info_sent_at"): op.add_column( "identity_persons", sa.Column("it_info_sent_at", sa.DateTime(timezone=True), nullable=True), ) if not _col_exists(bind, "identity_persons", "it_info_sent_channels"): op.add_column( "identity_persons", sa.Column("it_info_sent_channels", sa.JSON, nullable=True), ) # ── Drop unused wf_* tables + stale columns ──────────────── # migration 001 created wf_instances / wf_tasks / wf_leaver_cases # for a workflow engine that was never wired. Prod TiDB has an # additional table ``wf_offboarding_checklist_items`` (outside # this codebase) whose FK references ``wf_leaver_cases`` — a # naive DROP hits MySQL error 3730. Walk information_schema, # drop inbound FKs first, then try the DROP. Any stubborn # leftover is logged and swallowed: dead tables are cleanup, # not a hard requirement. _safe_drop_wf_tables(bind) if _col_exists(bind, "hr_lifecycle_events", "workflow_id"): with op.batch_alter_table("hr_lifecycle_events") as batch: batch.drop_column("workflow_id") if _col_exists(bind, "offboarding_processes", "summary"): with op.batch_alter_table("offboarding_processes") as batch: batch.drop_column("summary") _seed_scheduled_jobs() def _create_notification_outbox(person_id_type, table_kw) -> None: op.create_table( "notification_outbox", sa.Column("id", sa.CHAR(36), primary_key=True), sa.Column( "recipient_person_id", person_id_type, sa.ForeignKey("identity_persons.id"), nullable=True, ), sa.Column("recipient_email", sa.String(255), nullable=True), sa.Column("template_key", sa.String(64), nullable=True), sa.Column("title", sa.String(512), nullable=False), sa.Column("body_text", sa.Text, nullable=False), sa.Column("body_html", sa.Text, nullable=True), sa.Column("url", sa.String(512), nullable=True), sa.Column("context", sa.JSON, nullable=True), sa.Column("channels", sa.JSON, nullable=False), sa.Column("delivered_channels", sa.JSON, nullable=True), sa.Column("failed_channels", sa.JSON, nullable=True), sa.Column( "status", sa.String(16), nullable=False, server_default="pending", ), sa.Column( "attempts", sa.Integer, nullable=False, server_default="0", ), sa.Column( "max_attempts", sa.Integer, nullable=False, server_default="5", ), sa.Column("last_error", sa.Text, nullable=True), sa.Column("next_attempt_at", sa.DateTime(timezone=True), nullable=True), sa.Column("delivered_at", sa.DateTime(timezone=True), nullable=True), sa.Column("related_resource_type", sa.String(64), nullable=True), sa.Column("related_resource_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(), ), **table_kw, ) op.create_index( "ix_notification_outbox_status_next", "notification_outbox", ["status", "next_attempt_at"], ) op.create_index( "ix_notification_outbox_recipient", "notification_outbox", ["recipient_person_id"], ) def _create_compliance_incidents(person_id_type, table_kw) -> None: op.create_table( "compliance_incidents", sa.Column("id", sa.CHAR(36), primary_key=True), sa.Column( "incident_type", sa.String(64), nullable=False, ), sa.Column( "severity", sa.String(16), nullable=False, server_default="critical", ), sa.Column( "status", sa.String(16), nullable=False, server_default="open", ), sa.Column( "person_id", person_id_type, sa.ForeignKey("identity_persons.id"), nullable=True, ), sa.Column("resource_type", sa.String(64), nullable=True), sa.Column("resource_id", sa.CHAR(36), nullable=True), sa.Column("resource_label", sa.String(255), nullable=True), sa.Column("title", sa.String(255), nullable=False), sa.Column("detail", sa.JSON, nullable=True), sa.Column("detected_at", sa.DateTime(timezone=True), nullable=False), sa.Column("resolved_at", sa.DateTime(timezone=True), nullable=True), sa.Column( "resolved_by_person_id", person_id_type, sa.ForeignKey("identity_persons.id"), nullable=True, ), sa.Column("resolution_note", sa.Text, 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(), ), **table_kw, ) op.create_index( "ix_compliance_incidents_status", "compliance_incidents", ["status"], ) op.create_index( "ix_compliance_incidents_person", "compliance_incidents", ["person_id"], ) def _seed_scheduled_jobs() -> None: _seed_job( name="hr.hire_day_it_notify", display_name="入职当天 IT 信息推送", description=( "在入职日 09:00 给当天入职的员工推送 IT 账号信息(登录入口 / " "已分配软件清单 / 设备领取 / 2FA / IT 支持邮箱),Email + Feishu " "双通道。已发送的通过 identity_persons.it_info_sent_at 幂等。" ), trigger_type="cron", cron_expr="0 9 * * *", timezone="Asia/Shanghai", config_json=None, ) _seed_job( name="hr.process_advance", display_name="离职流程自动推进", description=( "数据保鲜的核心:到 leaver_date 当天,对仍有未解 blocker 的离职人" "立即强制转让(级联 manager → manager.manager → IT Admin 兜底人)," "全部 owner 迁移完成后原子地触发 suspend_sso + 状态转 left。" "每 15 分钟跑一次,保证 status=left 的人在 DB 里没有任何字段指向他们。" ), trigger_type="interval", interval_seconds=900, timezone="UTC", config_json='{"escalation_fallback_email": null}', ) _seed_job( name="hr.post_leaver_verify", display_name="离职后 T+1 资源复查", description=( "对 24 小时前转为 left 状态的人做 rescan,发现仍然指向他们的" "资源(资产 / 软件 seat / 权限 / API Key / 云账号)时生成 critical " "compliance_incident 并推送 IT Admin。每周一另外跑一次全量 left 人员" "扫描(通过 config.weekly_full_scan)。" ), trigger_type="cron", cron_expr="0 2 * * *", timezone="Asia/Shanghai", config_json='{"recent_days": 7}', ) _seed_job( name="hr.daily_digest", display_name="入离职每日摘要", description=( "每天 09:00 给 IT Admin 发日报:今日应离 N 人 / 3 日内入职 K 人 / " "当前未解 blocker M 项 / 昨日新 compliance incidents L 项。" "Email + Feishu 双通道。" ), trigger_type="cron", cron_expr="0 9 * * *", timezone="Asia/Shanghai", config_json=None, ) _seed_job( name="notification.outbox_retry", display_name="通知失败重投", description=( "notification_outbox 里 status=pending 且 next_attempt_at <= now 的" "行按指数退避(5m / 30m / 2h / 8h / 24h)重投,max_attempts 次后" "标记 dead。保证所有通知要么送达,要么留痕可手动补发。" ), trigger_type="interval", interval_seconds=3600, timezone="UTC", config_json=None, ) def _seed_job( *, name: str, display_name: str, description: str, trigger_type: str, interval_seconds: int | None = None, cron_expr: str | None = None, timezone: str = "UTC", config_json: str | None = None, ) -> None: # Idempotent insert — only seeds if absent. parts = [ "INSERT INTO scheduled_jobs", " (id, name, display_name, description, handler_key, trigger_type,", " interval_seconds, cron_expr, timezone, enabled, config,", " run_count, failure_count, created_at, updated_at)", "SELECT UUID(), :name, :display_name, :description, :name, :trigger_type,", ] if trigger_type == "interval": parts.append(f" {int(interval_seconds)}, NULL,") else: parts.append( f" NULL, '{cron_expr.replace(chr(39), chr(39) * 2)}'," ) parts.append(f" '{timezone}', 1,") if config_json: parts.append(f" CAST('{config_json}' AS JSON),") else: parts.append(" NULL,") parts.append(" 0, 0, NOW(), NOW()") parts.append("FROM DUAL") parts.append( "WHERE NOT EXISTS (SELECT 1 FROM scheduled_jobs WHERE name = :name)" ) sql = "\n".join(parts) op.execute( sa.text(sql).bindparams( name=name, display_name=display_name, description=description, trigger_type=trigger_type, ) ) def downgrade() -> None: # Remove seeded rows for n in ( "hr.hire_day_it_notify", "hr.process_advance", "hr.post_leaver_verify", "hr.daily_digest", "notification.outbox_retry", ): op.execute( sa.text("DELETE FROM scheduled_jobs WHERE name = :n").bindparams(n=n) ) # NOTE: we don't re-create the dropped wf_* tables nor the # workflow_id / summary columns on downgrade — they were dead code. op.drop_column("identity_persons", "it_info_sent_channels") op.drop_column("identity_persons", "it_info_sent_at") op.drop_index( "ix_compliance_incidents_person", table_name="compliance_incidents", ) op.drop_index( "ix_compliance_incidents_status", table_name="compliance_incidents", ) op.drop_table("compliance_incidents") op.drop_index( "ix_notification_outbox_recipient", table_name="notification_outbox", ) op.drop_index( "ix_notification_outbox_status_next", table_name="notification_outbox", ) op.drop_table("notification_outbox")