"""HRIS 批次:把「初始播种」从 sync_mode='full' 里拆出来。

``sync_mode='full'`` 一直背着两个互相冲突的含义。

* **初始播种** —— 一个 HRIS 实例第一次进 Helix。整册人一次性落库,
  其中几百个是早就离职的历史人员。这时候派发协调器是灾难:会给
  500 个人建账号、给 130 个历史离职者开离职流程。抑制是对的。
* **例行全量对账** —— ``hris.full_sync`` 每周跑一次。它存在的**全部**
  理由就是捞增量同步捞不到的人(增量硬上限回看 30 天)。对这种批次
  抑制协调器,等于把它捞回来的东西又扔掉一次。

代码里只有一句 ``is_initial_seed = (batch.sync_mode or "").lower() == "full"``
(``service.py`` 两处),于是第二种被当成第一种。生产后果,直连 TiDB 取证:

* ``hr_lifecycle_events``:onboard 557、offboard 133、**role_change 1**。
  五次被检出的组织变动里,四次落在 ``sync_mode='full'`` 批次上 ——
  ``role_change`` 事件 0 条、云授权复核标记 0 条;唯一成功的那次
  (2026-07-13)落在 ``incremental`` 上。相关性是完美的。
* ``cmdb_cloud_access_grants.review_status`` 5273 行**全部为 NULL** ——
  转岗复核这条路从来没有被走过。
* ``trigger_source='initial_seed'`` 的 lifecycle 事件有 50 条,时间跨到
  2026-08-16 —— 全是每周例行全量同步写的,离真正的播种(2026-04-21)
  已经四个月。这些人的建号、邮箱预留、离职流程当时都没有派发。

修法不是删掉这个标志,是**把两个概念分开存**。``is_seed`` 由建批次时
的谓词算出(该 HRIS 实例还没有任何人、也没有任何生命周期事件 = 播种),
并允许运维显式指定 —— 历史数据回捞是一次刻意的操作,应该说出来,而不是
让「全量」这个词替它决定。

已有行一律回填 ``is_seed=1``:保持它们**原样**的历史语义(万一有
``applying`` 批次被恢复重放,行为一个字不变)。新批次走谓词。

同时改的还有质量刹车:``offboard_count > threshold and sync_mode != 'full'``
现在按 ``is_seed`` 判。生产复核过:最近的每周全量批次 deactivate 最多 16 条,
阈值是 ``max(25, 5%×444)=25``,不会误刹;而 2026-06-01 那次 59 条的历史
离职回捞正是应该让人看一眼的形状,那种批次现在由运维显式标 ``is_seed``
豁免。

加列 + 回填,``information_schema`` 按 ``DATABASE()`` 守卫,不建索引 ——
降级就是一次 drop,TiDB 没有索引要重建。

Revision ID: 565
Revises: 564
"""
from __future__ import annotations

import json as _json
import uuid

import sqlalchemy as sa

from alembic import op

revision = "565"
down_revision = "564"
branch_labels = None
depends_on = None

TABLE = "hr_sync_batches"
COLUMN = "is_seed"
RECONCILE_JOB_KEY = "role_change.dispatch_reconcile"

_RECONCILE_DESC = (
    "每天 03:20 (Asia/Shanghai) 复核最近 180 天内**已确认**的 HRIS 暂存事件:"
    "凡是暂存时记录了组织变动(manager / dept / title 任一为 true)、却没有对应 "
    "role_change 生命周期事件的,原样补发一次协调器 —— 写事件、把该员工的云授权"
    "标成 pending_review、跑一次坐席差异预览。\n"
    "为什么需要它:确认路径是**内联**调用协调器的,而协调器按平台惯例自吞异常。"
    "于是一次失败是完全隐形的 —— 暂存行照样变成 confirmed、人员行照样被改,"
    "但事件没写、授权没标。检测(暂存时落标记)与派发(这个任务)因此被分开:"
    "标记是持久的,这个扫描保证每一条被记录的变动最终都有一个事件与之对应。\n"
    "补发的事件按当时观察到的日期落 effective_date,不 backdate 到今天;"
    "trigger_source='reconcile' 与内联写入区分。按 (person_id, event_type, "
    "effective_date) 去重,重复运行是空转 —— 已被管理员判成 confirmed_keep 的"
    "授权不会被重新翻成待复核。只读外部:不打厂商接口、不调用模型、不发通知。"
)


def _has_column(conn: sa.engine.Connection, column: str) -> bool:
    # DATABASE()-qualified: an unqualified information_schema probe
    # matches a same-named table in ANY schema, and this box runs
    # hundreds of helix_test_* schemas side by side.
    return bool(
        conn.execute(
            sa.text(
                "SELECT COUNT(*) FROM information_schema.columns "
                "WHERE table_schema = DATABASE() "
                "AND table_name = :t AND column_name = :c",
            ),
            {"t": TABLE, "c": column},
        ).scalar()
    )


def upgrade() -> None:
    conn = op.get_bind()
    if not sa.inspect(conn).has_table(TABLE):
        return
    if _has_column(conn, COLUMN):
        return

    op.add_column(
        TABLE,
        sa.Column(
            COLUMN,
            sa.Boolean(),
            nullable=False,
            server_default=sa.text("0"),
            comment=(
                "True = 初始播种(该 HRIS 实例第一次整册入库):抑制全部"
                "生命周期协调器、豁免群体离职刹车。False = 例行同步"
                "(含每周全量对账):协调器照常派发。此前由 "
                "sync_mode='full' 顶替,导致每周全量对账捞回来的组织变动/"
                "新人/离职被静默丢弃。"
            ),
        ),
    )

    # Backfill: every EXISTING full batch keeps exactly the semantics it
    # ran under. This is deliberately not "guess which ones were真播种" —
    # a resumable ``applying`` batch must replay identically to how it
    # started, and a confirmed/cancelled batch's flag is only ever read
    # again by the resume path. New batches get the predicate.
    conn.execute(
        sa.text(
            f"UPDATE {TABLE} SET {COLUMN} = 1 "
            "WHERE LOWER(COALESCE(sync_mode, '')) = 'full'",
        ),
    )

    # The straggler reconciler, seeded ENABLED. The confirm path calls
    # the role-change coordinator INLINE and the coordinator swallows
    # its own exceptions, so a dispatch failure leaves no trace at all:
    # the staged row still reaches ``confirmed`` and the person still
    # moves, but no lifecycle event exists and no cloud grant is
    # flagged. This sweep asserts the invariant the inline call cannot:
    # every staged row that RECORDED a move eventually has an event to
    # show for it.
    #
    # It cannot storm — it writes at most one lifecycle event per
    # (person, effective_date), sends no notification, and drains to a
    # no-op once caught up.
    conn.execute(
        sa.text(
            "INSERT INTO scheduled_jobs "
            "(id, name, display_name, description, handler_key, "
            " trigger_type, cron_expr, timezone, enabled, config, "
            " run_count, failure_count, created_at, updated_at) "
            "SELECT :id, :name, :display, :desc, :handler, 'cron', "
            ":cron, 'Asia/Shanghai', 1, :config, 0, 0, NOW(), NOW() "
            "FROM DUAL WHERE NOT EXISTS "
            "(SELECT 1 FROM scheduled_jobs WHERE handler_key = :handler)",
        ),
        {
            "id": str(uuid.uuid4()),
            "name": RECONCILE_JOB_KEY,
            "display": "转岗派发对账(补发漏掉的组织变动)",
            "desc": _RECONCILE_DESC,
            "handler": RECONCILE_JOB_KEY,
            # 03:20 — after the nightly HRIS activity has settled and
            # well before the 09:00–10:00 digest band, so a replay's
            # freshly-flagged grants are visible to the same morning's
            # ``role_change.scope_review_sweep`` rather than waiting a
            # full day for their first reminder.
            "cron": "20 3 * * *",
            "config": _json.dumps({"lookback_days": 180, "limit": 200}),
        },
    )


def downgrade() -> None:
    conn = op.get_bind()
    # The cron row goes first and independently of the column: the
    # handler reads ``is_seed`` on no path, but leaving an enabled row
    # whose handler this revision introduced would fail the worker's
    # ``_assert_all_enabled_handlers_registered`` boot check on a
    # downgraded image.
    if sa.inspect(conn).has_table("scheduled_jobs"):
        conn.execute(
            sa.text("DELETE FROM scheduled_jobs WHERE handler_key = :k"),
            {"k": RECONCILE_JOB_KEY},
        )
    if not sa.inspect(conn).has_table(TABLE):
        return
    if _has_column(conn, COLUMN):
        op.drop_column(TABLE, COLUMN)