"""Onboarding domain scheduled-job handlers.

Three jobs (mirrors offboarding's reconcile + escalation + post-leaver
verify trio):

* ``onboarding.reconcile_processes`` — every 5 minutes. Rescans every
  active process, flips blockers whose underlying resource appeared, and
  re-pends ones whose resource disappeared.
* ``onboarding.escalation_tick`` — every 30 minutes. Computes
  ``sla_breach_level`` for every preparing/live process; future PRs
  layer Feishu push at the 7d/3d/1d/day-0 thresholds (PR-3).
* ``onboarding.t7_verify`` — daily at 02:30 Asia/Shanghai. For every
  ``live`` process whose ``live_at`` is ≥7 days ago, rescan; close as
  ``completed`` if green, or open ``onboarding_day1_gap`` compliance
  incident if any dimension is still pending (and not ``backfilled``).
  Also keeps those incidents honest in both directions — re-opens one
  that was resolved over a still-pending gap, auto-closes one whose
  process has since gone terminal.

Handlers are import-side-effect registered — see
``server/app/scheduled_jobs/registry.py``. The worker.py / main.py
import this module exactly once per process to make the registration
take effect (see ``feedback_worker_handler_imports``).
"""
from __future__ import annotations

import logging
import uuid
from typing import Any

from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.onboarding.advance_service import OnboardingT0AdvanceService
from app.domains.onboarding.service import OnboardingService
from app.domains.onboarding.t7_verify_service import OnboardingT7VerifyService
from app.scheduled_jobs.handlers import _run_in_fresh_session_with_lock_retry
from app.scheduled_jobs.registry import register_scheduled_job

logger = logging.getLogger(__name__)


@register_scheduled_job(
    key="onboarding.reconcile_processes",
    display_name="入职流程对账",
    description=(
        "每 5 分钟扫描所有 scanning / preparing / ready / live 状态的 "
        "OnboardingProcess,对每个 process 调用 reconcile_blockers:重新"
        "扫描 7+ 维度的 Day-1 readiness,把已经提供到位的资源对应的 "
        "blocker 翻为 provisioned,把新出现的缺口创建为 pending blocker。"
        "对称 offboarding.reconcile_processes。"
    ),
)
async def run_onboarding_reconcile(
    db: AsyncSession,
    config: dict[str, Any],
) -> dict[str, Any]:
    trace_id = f"system:cron:onboarding.reconcile:{uuid.uuid4()}"

    async def _work(session: AsyncSession) -> dict[str, Any]:
        svc = OnboardingService(session)
        # Also runs the daily T+0 advance — cheap, atomic, and a 5-min
        # cadence means we won't miss the hire_date boundary by more
        # than 5 minutes in the worst case. This consolidates the
        # cron count by 1 (no separate t0_advance job needed).
        advance = await OnboardingT0AdvanceService(
            session
        ).advance_all_due(trace_id=trace_id)
        recon = await svc.reconcile_all_active_processes()
        # Fold BOTH halves' per-process failures into the handler gate below
        # (previously only ``recon`` failures were surfaced, so a T+0 advance
        # that failed every tick left the run 'ok' with no failure-notifier).
        return {
            "blockers_flipped": recon["blockers_flipped"],
            "processes_advanced_to_live": advance["flipped"],
            "advance_failed_count": advance["failed_count"],
            "failed_count": recon["failed_count"] + advance["failed_count"],
            "errors": recon["errors"] + advance["errors"],
        }

    # Both halves re-scan live process state, so a fresh-session re-run
    # after a transient DB error is idempotent. The 2026-06-14 failure was
    # errno 2013 ("Lost connection ... connection reset by peer") — a TiDB
    # Cloud node RST'd an in-flight query; pool_pre_ping only probes at
    # checkout so it can't catch a mid-query drop. Retry on a fresh
    # connection is the canonical recovery.
    result = await _run_in_fresh_session_with_lock_retry(
        _work, label="onboarding.reconcile_processes"
    )
    # Handler-must-raise contract (mirrors offboarding.reconcile): per-process
    # failures are collected into ``errors`` rather than propagated so one bad
    # process can't abort the tick — but returning the summary verbatim would
    # land scheduled_job_runs.status='ok' even when processes failed every
    # tick, silently, with no failure-notifier page. Re-raise on any failure;
    # the successful work is already committed by the fresh-session wrapper, so
    # this only flips run status.
    failed = result.get("failed_count", 0) if isinstance(result, dict) else 0
    if failed:
        raise RuntimeError(
            f"onboarding.reconcile: {failed} process(es) failed to reconcile — "
            f"see errors in the run summary: {result.get('errors')}"
        )
    return result


@register_scheduled_job(
    key="onboarding.escalation_tick",
    display_name="入职 SLA 升级扫描 + IT 工单升级",
    description=(
        "每 30 分钟跑三件事:① 对每个 preparing/live 的 OnboardingProcess "
        "重新计算 sla_breach_level (none/warn/urgent/breach); ② day0_it_gaps:"
        "hire_date 恰为员工本地当天且仍有 BLOCKER 级 Day-1 项 pending 的流程,"
        "向 IT 推硬缺口工单卡(单发,深链流程详情;迟到/补开的流程交给 SLA "
        "梯队而非 day0 卡); ③ sla_urgent / sla_breach:"
        "SLA 进入 urgent / breach 均只推 IT(各单发)。所有推送均 IT-only "
        "且对应一个具体 IT 动作 —— 不向经理 / 员工发纯 FYI 提醒(入离职 "
        "提醒由 HRIS 负责;2026-06-15 移除经理面 7d/3d/1d/day0 提醒卡)。"
        "全部 idempotent via onboarding_reminder_log。"
    ),
)
async def run_onboarding_escalation_tick(
    db: AsyncSession,
    config: dict[str, Any],
) -> dict[str, Any]:
    from sqlalchemy import select

    from app.core.exceptions import is_transaction_aborting_db_error
    from app.domains.onboarding.models import (
        PROC_LIVE,
        PROC_PREPARING,
        OnboardingProcess,
    )

    trace_id = f"system:cron:onboarding.escalation:{uuid.uuid4()}"

    async def _work(session: AsyncSession) -> dict[str, Any]:
        svc = OnboardingService(session)
        # Part 1: SLA breach evaluation on preparing / live processes.
        procs = (
            (
                await session.execute(
                    select(OnboardingProcess).where(
                        OnboardingProcess.status.in_(
                            [PROC_PREPARING, PROC_LIVE]
                        )
                    )
                )
            )
            .scalars()
            .all()
        )
        breach_counts: dict[str, int] = {}
        failed_count = 0
        errors: list[dict[str, str]] = []
        for proc in procs:
            # Capture the id up front: evaluate_sla dirties + flushes proc
            # inside the SAVEPOINT, so after a rollback the ORM instance is
            # expired and reading proc.id in the except handler would emit a
            # sync lazy-load SELECT → MissingGreenlet in the async context.
            # Mirrors ProcessAdvanceService's guard.
            proc_id = str(proc.id)
            # Per-process SAVEPOINT isolation (mirrors reconcile's
            # per-process nesting): a single evaluate_sla flush failure must
            # not poison the session and discard the whole tick's commit.
            try:
                async with session.begin_nested():
                    level = await svc.evaluate_sla(proc)
            except Exception as exc:
                # A transaction-aborting error (deadlock 1213 / lock-wait
                # 1205 / lost-connection) kills the whole txn server-side
                # on TiDB; swallowing it here would cascade onto every
                # later begin_nested (orphaned-savepoint) and make the
                # trailing session.commit() raise PendingRollbackError,
                # masking the real cause and silently discarding the
                # entire tick — including the IT escalation pushes below.
                # Re-raise so the lock-retry wrapper restarts the whole
                # tick on a fresh session (it is idempotent: evaluate_sla
                # only levels-up and the dispatchers are reminder-log
                # deduped). Mirrors OnboardingService.
                # reconcile_all_active_processes.
                if is_transaction_aborting_db_error(exc):
                    raise
                logger.exception(
                    "onboarding.escalation: evaluate_sla failed for %s",
                    proc_id,
                )
                # Handler-must-raise contract (mirrors sibling
                # onboarding.reconcile / onboarding.t7_verify): a per-process
                # evaluate_sla failure must be COUNTED so the tick can raise
                # after _work — previously this only logged+continued, so a
                # process whose SLA eval failed every 30-min tick left the run
                # status='ok' with no failure-notifier page, and that hire's
                # sla_breach_level froze / never escalated (#1747).
                failed_count += 1
                if len(errors) < 50:
                    errors.append({
                        "process_id": proc_id,
                        "error": f"{type(exc).__name__}: {exc!s:.200}",
                    })
                continue
            breach_counts[level] = breach_counts.get(level, 0) + 1
        # Part 2 + 3 (onboarding-parity PR2): IT-facing escalation —
        # hard Day-1 gaps on arrival day, then urgent/breach tiers.
        # (Manager-facing 7d/3d/1d/day0 reminder cards were removed
        # 2026-06-15 — HRIS owns join/leave reminders; Helix only pushes
        # when there is a concrete IT action.)
        day0_gap_cards = await svc.dispatch_day0_it_gaps(trace_id=trace_id)
        sla_escalation_counts = await svc.dispatch_sla_escalations(
            trace_id=trace_id,
        )
        # No explicit commit here — the lock-retry helper commits the
        # fresh session after _work returns (matches sibling
        # run_onboarding_reconcile / run_onboarding_t7_verify).
        return {
            "sla_level_counts": breach_counts,
            "day0_it_gap_cards": day0_gap_cards,
            "sla_escalation_counts": sla_escalation_counts,
            "failed_count": failed_count,
            "errors": errors,
            "trace_id": trace_id,
        }

    # Wrap in the fresh-session lock-retry helper like the sibling
    # onboarding crons (reconcile / t7_verify). Previously this handler
    # used a bare session with no retry, so a transient TiDB error
    # (deadlock / node RST) dropped the whole SLA + IT-escalation tick
    # with no recovery until the next 30-min cadence.
    result = await _run_in_fresh_session_with_lock_retry(
        _work, label="onboarding.escalation_tick"
    )
    # Handler-must-raise contract (mirrors onboarding.reconcile /
    # onboarding.t7_verify): per-process evaluate_sla failures are collected
    # into ``errors`` so one bad process can't abort the tick — but returning
    # the summary verbatim would land scheduled_job_runs.status='ok' even when
    # a process fails every tick, silently, with no failure-notifier page. The
    # dispatched cards + committed SLA levels are already persisted by the
    # fresh-session wrapper; this only flips the run status (#1747).
    failed = result.get("failed_count", 0) if isinstance(result, dict) else 0
    if failed:
        raise RuntimeError(
            f"onboarding.escalation: {failed} process(es) failed SLA "
            f"evaluation — see errors in the run summary: "
            f"{result.get('errors')}"
        )
    return result


@register_scheduled_job(
    key="onboarding.t7_verify",
    display_name="入职 T+7 健康检查",
    description=(
        "每天 02:30 (Asia/Shanghai) 扫描所有 live 状态且 live_at <= now-7d "
        "的 OnboardingProcess,重新 rescan;全 clear 翻 completed,仍有 "
        "pending 开 compliance_incident (onboarding_day1_gap, warning) 并"
        "去重打 t7_verify_failed 标记。backfilled=1 的行直接 completed。"
        "同一次 tick 还负责事件生命周期的两端:已 resolved 但缺口仍在的"
        "(超过 3 天宽限期)清掉标记重开;process 已 completed/cancelled 的"
        "open 事件系统自动关闭。"
    ),
)
async def run_onboarding_t7_verify(
    db: AsyncSession,
    config: dict[str, Any],
) -> dict[str, Any]:
    trace_id = f"system:cron:onboarding.t7_verify:{uuid.uuid4()}"

    async def _work(session: AsyncSession) -> dict[str, Any]:
        return await OnboardingT7VerifyService(session).verify_all_due(
            trace_id=trace_id,
        )

    result = await _run_in_fresh_session_with_lock_retry(
        _work, label="onboarding.t7_verify"
    )
    result["trace_id"] = trace_id
    # Handler-must-raise contract (mirrors onboarding.reconcile): per-process
    # failures are swallowed into ``errors`` so one bad process can't abort the
    # tick, but returning the summary verbatim would land
    # scheduled_job_runs.status='ok' even when a process fails every night —
    # silently, with no failure-notifier page and the day-1-gap incident never
    # opened. The successful work is already committed; this only flips status.
    failed = result.get("failed_count", 0) if isinstance(result, dict) else 0
    if failed:
        raise RuntimeError(
            f"onboarding.t7_verify: {failed} process(es) failed to verify — "
            f"see errors in the run summary: {result.get('errors')}"
        )
    return result