"""T+0 advance service.

Daily cron entry: ``OnboardingT0AdvanceService.advance_all_due()``.

Finds every onboarding process whose ``hire_date`` has arrived in the
hire's OWN local timezone (per-employee, not a single Asia/Shanghai
reference — see ``_local_today_for_person``) and whose status is
``ready`` or ``preparing``, and flips them to
``live``. The ``preparing`` case is on purpose — the hire showed up
today but some Day-1 dimension is still pending. We mark the process
``live`` (the person IS in seat now), but leave the pending blockers
outstanding so the 30-min escalation tick can DM IT / manager. SLA
tracking continues to run from ``first_ungated_at``.
"""
from __future__ import annotations

import logging
from datetime import UTC, datetime, timedelta
from zoneinfo import ZoneInfo

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.audit.service import AuditService
from app.core.coordinators.onboarding_hardware_coordinator import (
    autoclaim_reserved_for_hire,
)
from app.core.exceptions import is_transaction_aborting_db_error
from app.domains.onboarding.models import (
    PROC_LIVE,
    PROC_PREPARING,
    PROC_READY,
    OnboardingProcess,
)
from app.domains.onboarding.service import (
    _local_today_for_person,
    _person_onboarding_active,
)
from app.identity.person_models import IdentityPerson

logger = logging.getLogger(__name__)

_APP_TZ = ZoneInfo("Asia/Shanghai")


class OnboardingT0AdvanceService:
    def __init__(self, db: AsyncSession) -> None:
        self.db = db
        self.audit = AuditService(db)

    async def advance_all_due(self, *, trace_id: str) -> dict:
        """Flip every due process to ``live``.

        Returns ``{"flipped": N, "failed_count": M, "errors": [...]}`` — the
        same shape as the sibling crons (``reconcile_all_active_processes``,
        ``t7_verify.verify_all_due``, ``escalation_tick``) so the scheduled-job
        handler can fold advance failures into its ``handler-must-raise`` gate.
        The historical bare-int return hid per-process failures: a hire that
        failed to flip on every tick (e.g. a persistent non-aborting audit
        insert error on its row) left the run ``status='ok'``, the
        failure-notifier never fired, and the due hire silently never went
        live.
        """
        # Widen the candidate window by +1 day: hires in TZs EAST of
        # Shanghai may already be on their hire_date locally while HQ is
        # still a calendar day behind. We fetch the superset, then trim
        # per-person below against each hire's OWN local "today" so we
        # never flip a US/EU hire to live a full day early.
        today = datetime.now(_APP_TZ).date()
        procs = (
            (
                await self.db.execute(
                    select(OnboardingProcess).where(
                        OnboardingProcess.status.in_(
                            [PROC_READY, PROC_PREPARING]
                        ),
                        OnboardingProcess.hire_date <= today + timedelta(days=1),
                    )
                )
            )
            .scalars()
            .all()
        )
        # Batch-load the candidate persons in ONE query (avoid N+1) for
        # the per-employee local-today gate inside the loop. Key the map
        # by ``str(id)`` and look up by ``str(person_id)`` so the gate
        # survives any UUID-vs-str drift between the two columns: this
        # cron shares ONE session with ``reconcile_all_active_processes``
        # (expire_on_commit=False), so an ``OnboardingProcess`` already in
        # the identity map can hand back a str ``person_id`` while a
        # freshly-loaded ``IdentityPerson.id`` is a ``uuid.UUID`` — a raw
        # ``people.get(proc.person_id)`` then misses, returns None, and
        # ``_local_today_for_person(None)`` falls back to the UTC date,
        # advancing a US/EU hire a full day early at UTC windows where the
        # UTC date is ahead of their local date. Normalising both sides to
        # str closes that gap.
        person_ids = {str(p.person_id) for p in procs}
        people: dict = {}
        if person_ids:
            people = {
                str(p.id): p
                for p in (
                    await self.db.execute(
                        select(IdentityPerson).where(
                            IdentityPerson.id.in_(person_ids)
                        )
                    )
                )
                .scalars()
                .all()
            }
        flipped = 0
        failed_count = 0
        errors: list[str] = []
        for proc in procs:
            person = people.get(str(proc.person_id))
            # #1782: never flip a rescinded-offer / early-departure hire
            # (HRIS row ``employment_status='left'`` or soft-deleted) to
            # ``live``. All sibling paths (rescan auto-cancel, evaluate_sla,
            # both escalation dispatchers) already gate on
            # ``_person_onboarding_active``; T+0 advance was the one loop
            # that unconditionally flipped, so a leaver whose process was
            # still ready/preparing got flipped to live (+ an
            # ``onboarding.advanced_to_live`` audit row) on hire_date, and
            # only the same tick's *later* reconcile auto-cancelled it — if
            # that reconcile failed, the leaver stayed live until the next
            # tick. The reconcile still does the actual cancel; this just
            # stops the bogus flip in the first place.
            if not _person_onboarding_active(person):
                continue
            # Skip any hire whose hire_date hasn't actually arrived in
            # their local wall-clock — trims the +1-day superset above.
            if proc.hire_date > _local_today_for_person(person):
                continue
            # Capture the id up front: after a SAVEPOINT rollback the
            # ORM instance is expired, so reading proc.id in the except
            # handler would emit a sync SELECT (MissingGreenlet).
            proc_id = str(proc.id)
            # Per-process SAVEPOINT isolation: this cron shares ONE
            # AsyncSession with reconcile_all_active_processes (same
            # _work tick) under a single outer commit. Without
            # begin_nested, one proc's audit-row flush failure would
            # poison the whole session — discarding every legitimate T+0
            # flip and making the subsequent reconcile raise
            # PendingRollbackError that masks the real error, leaving a
            # due new hire stuck in ready/preparing until a later tick.
            # Mirrors reconcile_all_active_processes / t7_verify /
            # ProcessAdvanceService.run_tick.
            try:
                async with self.db.begin_nested():
                    # Locking re-read: re-fetch the row FOR UPDATE and
                    # re-assert its pre-state under the lock before flipping.
                    # Our candidate SELECT above is lock-free, so between it
                    # and here a concurrent admin ``cancel_process`` /
                    # ``advance_now`` (both take this same row lock via
                    # ``_get_or_raise(for_update=True)``) may have moved the
                    # row. Without this re-check the plain
                    # ``UPDATE ... SET status='live'`` would block on the
                    # admin's lock, then OVERWRITE the just-committed
                    # ``cancelled`` — resurrecting a rescinded-offer hire to
                    # ``live`` with a bogus ``advanced_to_live`` audit row.
                    # The #1782 person-level guard above does not cover a
                    # concurrently-cancelled *process*. ``populate_existing``
                    # refreshes the identity-mapped instance in place.
                    locked = (
                        await self.db.execute(
                            select(OnboardingProcess)
                            .where(OnboardingProcess.id == proc.id)
                            .with_for_update()
                            .execution_options(populate_existing=True)
                        )
                    ).scalar_one_or_none()
                    if locked is None or locked.status not in (
                        PROC_READY, PROC_PREPARING,
                    ):
                        # Concurrently advanced or cancelled — nothing to do.
                        continue
                    prior = locked.status
                    # Day-1 ledger truth: convert the hire's earmarked
                    # computer into a real claim at the moment they go
                    # live. Inside the same SAVEPOINT as the flip, so a
                    # failed claim rolls back the flip too and the next
                    # tick retries both together — never a live hire whose
                    # device silently stayed "reserved".
                    claimed_asset_id = await autoclaim_reserved_for_hire(
                        self.db,
                        locked.person_id,
                        trace_id=trace_id,
                        process_id=locked.id,
                    )
                    locked.status = PROC_LIVE
                    now = datetime.now(UTC)
                    if locked.live_at is None:
                        locked.live_at = now
                    if locked.ready_at is None and prior == PROC_READY:
                        locked.ready_at = now
                    await self.audit.log(
                        trace_id=trace_id,
                        actor_type="system",
                        action="onboarding.advanced_to_live",
                        status="success",
                        resource_type="onboarding_process",
                        resource_id=str(locked.id),
                        detail={
                            "prior_status": prior,
                            "hire_date": locked.hire_date.isoformat(),
                            "autoclaimed_asset_id": (
                                str(claimed_asset_id)
                                if claimed_asset_id
                                else None
                            ),
                        },
                    )
                flipped += 1
            except Exception as exc:
                # A transaction-aborting error (deadlock 1213 / lock-wait
                # 1205 / orphaned-savepoint 1305) kills the WHOLE txn on
                # TiDB, not just this savepoint — swallowing it would
                # cascade SAVEPOINT failures onto every later proc and make
                # the outer commit raise PendingRollbackError, masking the
                # real error. Re-raise so the handler's lock-retry wrapper
                # can recover in a fresh session (the work is idempotent).
                if is_transaction_aborting_db_error(exc):
                    raise
                logger.exception(
                    "onboarding.t0_advance failed for process %s", proc_id
                )
                failed_count += 1
                errors.append(f"{proc_id}: {exc}"[:500])
                continue
        if flipped:
            await self.db.flush()
        logger.info(
            "onboarding.t0_advance: flipped=%d failed=%d (today=%s)",
            flipped, failed_count, today,
        )
        return {
            "flipped": flipped,
            "failed_count": failed_count,
            "errors": errors,
        }