"""Onboarding domain: first-class ``onboarding_processes`` + ``onboarding_blockers`` + ``onboarding_reminder_log`` tables. Revision ID: 229 Revises: 228 Create Date: 2026-06-01 Background ========== Pre-229 the ``onboarding`` domain was a 352-line read-only aggregator over ``identity_persons`` / ``identity_external_identities`` / ``software_seat_allocations`` / ``asset_items`` / ``vpn_applications`` / ``identity_sync_targets``. Each cockpit page-load recomputed the 7-dimension Day-1 readiness fan-out from scratch — no state machine, no SLA, no per- blocker tracking, no T+N case-file verification. Asymmetric to the offboarding domain (9293 LoC, ``offboarding_processes`` + ``offboarding_blockers`` + ``offboarding_reminder_log`` since migration 023). PR-1 of the onboarding-symmetry series promotes onboarding to a first-class domain. This migration creates the three mirror tables. Subsequent migrations in later PRs will add: per-bundle FK on ``provision_policies``, etc. Tables ====== ``onboarding_processes`` (one row per person undergoing onboarding) - ``id`` (CHAR(36)/VARCHAR(36) matching identity_persons.id) - ``person_id`` (FK → identity_persons, UNIQUE — at most one row per person; a new hire-event creates the row, ``cancelled`` / ``completed`` rows remain for audit) - ``hire_date`` (Date) - ``status``: scanning / preparing / ready / live / completed / cancelled - Stage timestamps: ``scanned_at`` / ``ready_at`` / ``live_at`` / ``completed_at`` / ``cancelled_at`` - ``current_step`` + ``step_log`` JSON tail - ``first_ungated_at`` + ``sla_breach_level`` (none / warn / urgent / breach) — measures how long Day-1 has been blocked past hire_date - ``workflow_run_id`` — pointer into workflow_journal - ``default_buddy_person_id`` (FK → identity_persons, SET NULL) — set by manager in PR-3; nullable in PR-1 - ``hardware_allocation`` JSON — per-asset source map (PR-3) - ``backfilled`` BOOLEAN — rows seeded by this migration are flagged so the T+7 verify cron skips compliance_incident emission ``onboarding_blockers`` - ``process_id`` (CASCADE) - ``resource_type`` — work_email / sso_provision / feishu_contact / software_seat / hardware_allocation / vpn / buddy_assigned / email_group / ai_tool_default / training_ack - ``resource_id`` (nullable — blockers like "no SSO identity exists at all" have no concrete row to point at yet) - ``severity`` (blocker / advisory) - ``resolution`` (pending / provisioned / skipped / identity_sync_managed) - ``provisioning_task_id`` (FK → provisioning_tasks) - ``last_error_message`` / ``last_error_at`` / ``error_count`` (mirrors offboarding migration 148 storage) ``onboarding_reminder_log`` - ``process_id`` (CASCADE), ``reminder_kind`` — UNIQUE pair - Kinds: reminder_7d / reminder_3d / reminder_1d / day0_manager / t7_verify_failed Backfill ======== For every ``identity_persons`` with ``hire_date`` in [today-60d, today+60d], ``employment_status='active'``, ``deleted_at IS NULL``, and no existing ``onboarding_processes`` row: insert one row marked ``backfilled=TRUE`` with status bucketed by distance from today: - hire_date < today - 14d → ``completed`` (already settled) - hire_date in [today - 14d, today + 14d] → ``live`` - hire_date > today + 14d → ``scanning`` Backfill is idempotent — inner SELECT excludes any person who already has a row. Downgrade drops all three tables. """ 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 = "229" down_revision = "228" branch_labels = None depends_on = None def upgrade() -> None: persons_id = id_type_for("identity_persons") prov_id = id_type_for("provisioning_tasks") kw = table_kw_for("identity_persons") op.create_table( "onboarding_processes", sa.Column("id", persons_id, primary_key=True), sa.Column("person_id", persons_id, nullable=False), sa.Column("hire_date", sa.Date(), nullable=False), sa.Column( "status", sa.String(32), nullable=False, server_default="scanning", ), sa.Column("scanned_at", sa.DateTime(timezone=True), nullable=True), sa.Column("ready_at", sa.DateTime(timezone=True), nullable=True), sa.Column("live_at", sa.DateTime(timezone=True), nullable=True), sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True), sa.Column("cancelled_at", sa.DateTime(timezone=True), nullable=True), sa.Column("current_step", sa.String(128), nullable=True), sa.Column("step_log", sa.JSON(), nullable=True), sa.Column("first_ungated_at", sa.DateTime(timezone=True), nullable=True), sa.Column( "sla_breach_level", sa.String(16), nullable=False, server_default="none", ), sa.Column("workflow_run_id", persons_id, nullable=True), sa.Column("default_buddy_person_id", persons_id, nullable=True), sa.Column("hardware_allocation", sa.JSON(), nullable=True), sa.Column( "backfilled", sa.Boolean(), nullable=False, server_default=sa.text("0"), ), 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.ForeignKeyConstraint( ["person_id"], ["identity_persons.id"], name="fk_onboarding_processes_person", ), sa.ForeignKeyConstraint( ["default_buddy_person_id"], ["identity_persons.id"], name="fk_onboarding_processes_buddy", ondelete="SET NULL", ), sa.UniqueConstraint( "person_id", name="uq_onboarding_processes_person", ), **kw, ) op.create_index( "ix_onboarding_processes_status_hire_date", "onboarding_processes", ["status", "hire_date"], ) op.create_index( "ix_onboarding_processes_hire_date", "onboarding_processes", ["hire_date"], ) op.create_table( "onboarding_blockers", sa.Column("id", persons_id, primary_key=True), sa.Column("process_id", persons_id, nullable=False), sa.Column("resource_type", sa.String(32), nullable=False), sa.Column("resource_id", persons_id, nullable=True), sa.Column("label", sa.String(255), nullable=False), sa.Column( "severity", sa.String(16), nullable=False, server_default="blocker", ), sa.Column( "resolution", sa.String(32), nullable=False, server_default="pending", ), sa.Column("resolution_ref", sa.String(255), nullable=True), sa.Column("provisioning_task_id", prov_id, nullable=True), sa.Column("notes", sa.Text(), nullable=True), sa.Column("last_error_message", sa.Text(), nullable=True), sa.Column("last_error_at", sa.DateTime(timezone=True), nullable=True), sa.Column( "error_count", sa.Integer(), nullable=False, server_default="0", ), 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.ForeignKeyConstraint( ["process_id"], ["onboarding_processes.id"], name="fk_onboarding_blockers_process", ondelete="CASCADE", ), sa.ForeignKeyConstraint( ["provisioning_task_id"], ["provisioning_tasks.id"], name="fk_onboarding_blockers_task", ), **kw, ) op.create_index( "ix_onboarding_blockers_process", "onboarding_blockers", ["process_id"], ) op.create_index( "ix_onboarding_blockers_resource_type", "onboarding_blockers", ["resource_type", "resolution"], ) op.create_table( "onboarding_reminder_log", sa.Column("id", persons_id, primary_key=True), sa.Column("process_id", persons_id, nullable=False), sa.Column("reminder_kind", sa.String(32), nullable=False), sa.Column( "sent_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column("channels", sa.JSON(), nullable=True), sa.Column("detail", sa.JSON(), nullable=True), sa.ForeignKeyConstraint( ["process_id"], ["onboarding_processes.id"], name="fk_onboarding_reminder_log_process", ondelete="CASCADE", ), sa.UniqueConstraint( "process_id", "reminder_kind", name="uq_onboarding_reminder_log_process_kind", ), **kw, ) # Backfill: one row per active person with hire_date in the # ±60-day window and no existing process. Status is bucketed by # distance from today; ``backfilled=1`` lets the T+7 verify cron # skip compliance_incident emission for legacy hires we never # tracked end-to-end. op.execute( sa.text( """ INSERT INTO onboarding_processes ( id, person_id, hire_date, status, scanned_at, ready_at, live_at, completed_at, sla_breach_level, backfilled, created_at, updated_at ) SELECT LOWER(REPLACE(UUID(), '-', '-')) AS id, p.id AS person_id, p.hire_date, CASE WHEN p.hire_date < DATE_SUB(CURDATE(), INTERVAL 14 DAY) THEN 'completed' WHEN p.hire_date <= DATE_ADD(CURDATE(), INTERVAL 14 DAY) THEN 'live' ELSE 'scanning' END AS status, NOW() AS scanned_at, CASE WHEN p.hire_date <= DATE_ADD(CURDATE(), INTERVAL 14 DAY) THEN NOW() ELSE NULL END AS ready_at, CASE WHEN p.hire_date <= CURDATE() THEN NOW() ELSE NULL END AS live_at, CASE WHEN p.hire_date < DATE_SUB(CURDATE(), INTERVAL 14 DAY) THEN NOW() ELSE NULL END AS completed_at, 'none' AS sla_breach_level, 1 AS backfilled, NOW() AS created_at, NOW() AS updated_at FROM identity_persons p WHERE p.hire_date IS NOT NULL AND p.hire_date BETWEEN DATE_SUB(CURDATE(), INTERVAL 60 DAY) AND DATE_ADD(CURDATE(), INTERVAL 60 DAY) AND p.employment_status = 'active' AND p.deleted_at IS NULL AND NOT EXISTS ( SELECT 1 FROM onboarding_processes op WHERE op.person_id = p.id ) """ ) ) # ── Seed the three new scheduled_jobs rows ── # Each is idempotent (NOT EXISTS guard) so an operator who already # added the handler row by hand doesn't get a duplicate. for name, display_name, description, trigger_type, interval, cron in [ ( "onboarding.reconcile_processes", "入职流程对账", ( "每 5 分钟扫描所有 scanning / preparing / ready / live 状态的 " "OnboardingProcess,对每个 process 调用 reconcile_blockers:" "重新扫描 7+ 维度的 Day-1 readiness,把已经提供到位的资源对应" "的 blocker 翻为 provisioned,把新出现的缺口创建为 pending " "blocker。与 offboarding.reconcile_processes 对称。" ), "interval", 300, None, ), ( "onboarding.escalation_tick", "入职 SLA 升级扫描", ( "每 30 分钟扫描所有 preparing / live 状态的 OnboardingProcess," "计算 hire_date 过去后还有多少 pending blocker,更新 " "sla_breach_level (none → warn @ 0-24h → urgent @ 24-72h → " "breach @ >72h),并按 7d/3d/1d/day0_manager 节点推送 Feishu " "卡片到经理 + IT。reminder_log 去重,同一节点不重复推送。" ), "interval", 1800, None, ), ( "onboarding.t7_verify", "入职 T+7 健康检查", ( "每天 02:30 (Asia/Shanghai) 扫描所有 live 状态的 OnboardingProcess " "且 live_at <= now - 7d,重新 rescan;若所有 blocker 已结清" "则翻为 completed,若仍有 pending 则开 compliance_incident " "(type=onboarding_day1_gap, severity=warning) 并标记 " "t7_verify_failed 提醒(去重)。backfilled=1 的行直接 " "completed,不发 incident。" ), "cron", None, "30 2 * * *", ), ]: exists = ( op.get_bind() .execute( sa.text( "SELECT 1 FROM scheduled_jobs WHERE name = :n LIMIT 1" ), {"n": name}, ) .scalar() ) if exists: continue op.get_bind().execute( sa.text( """ 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) VALUES (UUID(), :n, :dn, :desc, :hk, :tt, :ivl, :cron, :tz, 1, JSON_OBJECT(), 0, 0, NOW(), NOW()) """ ), { "n": name, "dn": display_name, "desc": description, "hk": name, "tt": trigger_type, "ivl": interval, "cron": cron, "tz": "Asia/Shanghai" if cron else "UTC", }, ) def downgrade() -> None: op.get_bind().execute( sa.text( "DELETE FROM scheduled_jobs WHERE name IN " "('onboarding.reconcile_processes', " "'onboarding.escalation_tick', " "'onboarding.t7_verify')" ) ) op.drop_index( "ix_onboarding_blockers_resource_type", table_name="onboarding_blockers", ) op.drop_index( "ix_onboarding_blockers_process", table_name="onboarding_blockers", ) op.drop_table("onboarding_reminder_log") op.drop_table("onboarding_blockers") op.drop_index( "ix_onboarding_processes_hire_date", table_name="onboarding_processes", ) op.drop_index( "ix_onboarding_processes_status_hire_date", table_name="onboarding_processes", ) op.drop_table("onboarding_processes")