"""Onboarding cockpit read-aggregator.

Renders the Day-1 readiness view (ten dimensions since 2026-06-12)
off the ``onboarding_processes`` + ``onboarding_blockers`` tables
instead of recomputing the dimensions per page-load. The state-machine
service (``OnboardingService``) is the single writer; the cockpit is a
pure reader.

Response shape is backward-compatible with the original cockpit so the
existing frontend ``OnboardingCockpitPage`` keeps working with no
changes beyond the eventual tab consolidation.
"""
from __future__ import annotations

from datetime import date, datetime, timedelta
from typing import Any
from uuid import UUID
from zoneinfo import ZoneInfo

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

from app.core.coordinators.onboarding_hardware_coordinator import (
    HireHardware,
    resolve_hire_hardware_states,
)
from app.domains.identity_sync.models import IdentitySyncTarget
from app.domains.onboarding.models import (
    PROC_OPEN_STATUSES,
    RES_AI_TOOL_DEFAULT,
    RES_EMAIL_GROUP,
    RES_FEISHU_CONTACT,
    RES_HARDWARE_ALLOCATION,
    RES_SOFTWARE_SEAT,
    RES_SSO_PROVISION,
    RES_TRAINING_ACK,
    RES_VPN,
    RES_WORK_EMAIL,
    RESO_PENDING,
    OnboardingBlocker,
    OnboardingProcess,
)
from app.identity.identity_reconcile import LOGIN_PROVIDERS
from app.identity.person_models import (
    IdentityExternalIdentity,
    IdentityPerson,
)

# Forward window for upcoming hires — the cockpit shows every OPEN
# onboarding process (scanning/preparing/ready/live) whose hire_date is
# no more than this many days in the FUTURE. There is intentionally NO
# backward bound: a hire who already started (status ``live``) but is
# still missing Day-1 dimensions must stay on the board until the
# process completes — that's the IT worklist. Capping the future side at
# +14d keeps far-out "preparing" rows (hire scheduled months ahead) off
# the urgent board without hiding anyone already in-flight. (Before
# 2026-06-29 the query also required hire_date >= today AND excluded
# ``live``, which silently dropped every already-onboard hire — the
# board read 0 while real open cases existed.)
ONBOARDING_WINDOW_DAYS = 14

# Fallback for a person the batch resolver didn't cover (filtered
# out mid-loop). Module-level constant so the row loop allocates nothing.
_MISSING_HARDWARE = HireHardware(state="missing")

_FEISHU_TARGET_KIND = "feishu_contact"
_HEALTHY_RUN_STATUSES = ("ok", "success")

# Per-dimension render state. Most dimensions are binary (READY /
# PENDING). ``work_email`` gets a third RESERVED state: the SF-planned
# address is written back to the person but the live mailbox hasn't been
# minted yet (no active Google/Feishu login identity). Before 2026-07-06
# the cockpit greened 账号/SSO/邮箱 off ``employment_status='active'`` and
# a merely-reserved address, so a US hire with no Google account (P10002)
# read fully-ready. account/sso/work_email now derive from the live login
# identity — the real "is there an account" signal.
DIM_STATE_READY = "ready"
DIM_STATE_RESERVED = "reserved"
DIM_STATE_PENDING = "pending"

# Cockpit dimensions (UI display order). Ten since 2026-06-11: the
# three formerly-stubbed dimensions (email_group / ai_tool / training)
# went live with the policy-aware readiness scanner — email_group and
# ai_tool reflect what active ProvisionPolicy rows promise the person;
# training reflects the 入职须知 acknowledgement checklist.
DIMENSIONS = (
    "account",
    "sso",
    "work_email",
    "feishu",
    "seat",
    "computer",
    "vpn",
    "email_group",
    "ai_tool",
    "training",
)

# Map the BLOCKER-DRIVEN cockpit dimensions to their canonical
# resource_type. ``account`` / ``sso`` / ``work_email`` are intentionally
# NOT read from this map anymore — they derive live from the person's
# login identities (see the row loop) rather than the blocker set, so a
# hire with no real account can't false-green off a resolved/absent
# blocker. They stay listed here only for documentation of the underlying
# resource_type the state machine still tracks.
_DIM_TO_RESOURCE: dict[str, str | None] = {
    "account": None,
    "sso": RES_SSO_PROVISION,
    "work_email": RES_WORK_EMAIL,
    "feishu": RES_FEISHU_CONTACT,
    "seat": RES_SOFTWARE_SEAT,
    "computer": RES_HARDWARE_ALLOCATION,
    "vpn": RES_VPN,
    "email_group": RES_EMAIL_GROUP,
    "ai_tool": RES_AI_TOOL_DEFAULT,
    "training": RES_TRAINING_ACK,
}


def _shanghai_today() -> date:
    # Intentionally app-tz (single-viewpoint IT DASHBOARD aggregate): the
    # cockpit board presents ONE "today" for the whole queue, not a
    # per-employee state guard — so per-person local-today does NOT apply
    # here (contrast service._local_today_for_person).
    return datetime.now(ZoneInfo("Asia/Shanghai")).date()


class OnboardingCockpitService:
    def __init__(self, db: AsyncSession) -> None:
        self.db = db

    async def build_queue(self) -> dict[str, Any]:
        """Return ``{"kpis": {...}, "feishu": {...}, "rows": [...]}``.

        Reads onboarding_processes + onboarding_blockers, joins
        identity_persons for display, computes the same dimensions as
        the pre-PR-1 cockpit. Rows are sorted soonest-to-onboard first
        (upcoming hires nearest-first, then already-started stragglers),
        readiness asc as the tiebreak.

        Scope = every OPEN process (scanning/preparing/ready/**live**)
        with ``hire_date <= today + 14d``. ``live`` (already-onboard)
        rows are included on purpose: a hire in-seat but still missing a
        Day-1 dimension is exactly what IT needs to chase. Only the
        FUTURE side is bounded (+14d) — there is no backward cutoff, so a
        stuck/overdue live process never silently drops off the board.
        """
        today = _shanghai_today()
        window_end = today + timedelta(days=ONBOARDING_WINDOW_DAYS)

        procs = (
            (
                await self.db.execute(
                    select(OnboardingProcess).where(
                        OnboardingProcess.status.in_(PROC_OPEN_STATUSES),
                        OnboardingProcess.hire_date <= window_end,
                    )
                )
            )
            .scalars()
            .all()
        )

        feishu_meta = await self._build_feishu_meta()
        if not procs:
            return {
                "kpis": _empty_kpis(),
                "feishu": feishu_meta,
                "rows": [],
            }

        person_ids = [p.person_id for p in procs]
        proc_ids = [p.id for p in procs]
        # Exclude soft-deleted people up front. A departed person whose
        # onboarding process is still open (nothing cancels it on
        # offboard — the row only auto-cancels when the person row is
        # HARD-deleted, via _rescan) would otherwise surface here now
        # that the window has no backward bound. ``employment_status ==
        # 'left'`` is filtered in the row loop below (same invariant as
        # the platform headcount / leaver-display surfaces: exclude left).
        persons = {
            row.id: row
            for row in (
                await self.db.execute(
                    select(IdentityPerson).where(
                        IdentityPerson.id.in_(person_ids),
                        IdentityPerson.deleted_at.is_(None),
                    )
                )
            ).scalars()
        }

        # Live login identities across the person set — the real "does an
        # account exist" signal for the 账号 / SSO / 邮箱 dimensions. Only
        # ``is_login_identity=True`` rows (google / feishu) count; the
        # ``successfactors`` HRIS-source row is a directory record, not a
        # login account, and previously false-greened these three dims.
        # provider ∈ LOGIN_PROVIDERS is a redundant guard on top of the
        # flag (belt-and-suspenders — a mis-flagged HRIS row can't slip in).
        # Reserved-vs-assigned for the 电脑 dimension. Both satisfy
        # Day-1 (the T+0 advance converts the earmark into a claim), but
        # an admin must be able to see which is which: 「已预留」 means
        # the machine is set aside, not yet in the person's hands. Kept
        # OUT of ``dimension_states`` on purpose — that field's
        # ``reserved`` value means "not ready" (reserved-only mailbox),
        # and overloading it would make the readiness count disagree with
        # the state machine.
        hardware_states = await resolve_hire_hardware_states(
            self.db, person_ids,
        )

        login_person_ids: set[UUID] = set(
            (
                await self.db.execute(
                    select(IdentityExternalIdentity.person_id).where(
                        IdentityExternalIdentity.person_id.in_(person_ids),
                        IdentityExternalIdentity.is_login_identity.is_(True),
                        IdentityExternalIdentity.identity_status == "active",
                        IdentityExternalIdentity.provider.in_(LOGIN_PROVIDERS),
                        IdentityExternalIdentity.deleted_at.is_(None),
                    )
                )
            ).scalars()
        )

        # Manager display names for the rows we'll render.
        mgr_ids = {
            persons[pid].manager_person_id
            for pid in person_ids
            if pid in persons and persons[pid].manager_person_id is not None
        }
        mgr_names: dict[UUID, str] = {}
        if mgr_ids:
            mgr_rows = (
                await self.db.execute(
                    select(
                        IdentityPerson.id,
                        IdentityPerson.display_name,
                        IdentityPerson.work_email,
                    ).where(IdentityPerson.id.in_(mgr_ids))
                )
            ).all()
            mgr_names = {
                r.id: (r.display_name or r.work_email) for r in mgr_rows
            }

        # Pending blockers per process, indexed by resource_type so each
        # process's dim flags resolve in O(1).
        pending_blockers = (
            (
                await self.db.execute(
                    select(OnboardingBlocker).where(
                        OnboardingBlocker.process_id.in_(proc_ids),
                        OnboardingBlocker.resolution == RESO_PENDING,
                    )
                )
            )
            .scalars()
            .all()
        )
        pending_by_proc: dict[UUID, set[str]] = {}
        for b in pending_blockers:
            pending_by_proc.setdefault(b.process_id, set()).add(
                b.resource_type
            )

        rows: list[dict[str, Any]] = []
        for proc in procs:
            person = persons.get(proc.person_id)
            # Skip soft-deleted (absent from ``persons``) and departed
            # (``left``) people — a leaver's onboarding board entry is
            # noise; their case is owned by offboarding now.
            if person is None or person.employment_status == "left":
                continue
            pending_types = pending_by_proc.get(proc.id, set())
            has_login = person.id in login_person_ids
            has_reserved_email = bool(
                person.work_email and "@" in person.work_email
            )
            dims: dict[str, bool] = {}
            dim_states: dict[str, str] = {}
            for dim_key in DIMENSIONS:
                if dim_key in ("account", "sso"):
                    # 账号 / SSO: green only when a real login account
                    # exists. These no longer read the blocker set or
                    # employment_status — the live login identity is the
                    # single source of truth.
                    state = (
                        DIM_STATE_READY if has_login else DIM_STATE_PENDING
                    )
                elif dim_key == "work_email":
                    # 邮箱: live mailbox (login identity) → ready; a
                    # planned SF address with no live account → reserved
                    # (distinct amber chip, NOT green); neither → pending.
                    if has_login:
                        state = DIM_STATE_READY
                    elif has_reserved_email:
                        state = DIM_STATE_RESERVED
                    else:
                        state = DIM_STATE_PENDING
                else:
                    # Blocker-driven dims (feishu/seat/computer/vpn/
                    # email_group/ai_tool/training) — ready iff no pending
                    # blocker for the resource_type. Their checks are real.
                    resource_type = _DIM_TO_RESOURCE[dim_key]
                    state = (
                        DIM_STATE_READY
                        if resource_type not in pending_types
                        else DIM_STATE_PENDING
                    )
                dim_states[dim_key] = state
                # ``dimensions`` stays a strict ready/not-ready bool for
                # backward compat (filters, KPIs, scorecard). RESERVED is
                # NOT ready — a reserved-only mailbox isn't Day-1 live.
                dims[dim_key] = state == DIM_STATE_READY
            ready_count = sum(1 for v in dims.values() if v)
            hw = hardware_states.get(
                person.id, _MISSING_HARDWARE,
            )
            rows.append(
                {
                    "process_id": str(proc.id),
                    "person_id": str(person.id),
                    "display_name": person.display_name or person.work_email,
                    "work_email": person.work_email,
                    "employee_no": person.employee_no,
                    "employment_status": person.employment_status,
                    "job_title": person.job_title,
                    "department": person.company_name,
                    "hire_date": (
                        person.hire_date.isoformat()
                        if person.hire_date
                        else None
                    ),
                    "manager_name": (
                        mgr_names.get(person.manager_person_id)
                        if person.manager_person_id
                        else None
                    ),
                    "status": proc.status,
                    "sla_breach_level": proc.sla_breach_level,
                    "hardware_state": hw.state,
                    "hardware_asset_tag": hw.asset_tag,
                    "dimensions": dims,
                    "dimension_states": dim_states,
                    "ready_count": ready_count,
                    "total_dimensions": len(DIMENSIONS),
                    "fully_ready": ready_count == len(DIMENSIONS),
                    "days_to_hire": (
                        (person.hire_date - today).days
                        if person.hire_date
                        else None
                    ),
                }
            )

        # Order by "how soon is Day-1", not raw calendar date. Upcoming
        # hires (days_to_hire >= 0) come first, nearest-first, so the
        # person 马上就要入职 sits at the top. Already-started rows
        # (days_to_hire < 0 — live stragglers still missing a dimension)
        # fall below them, most-recent first, instead of monopolising the
        # top under a plain hire_date-asc sort. None hire_date sorts last.
        def _sort_key(r: dict[str, Any]) -> tuple[int, int, int]:
            d = r["days_to_hire"]
            if d is None:
                return (2, 0, r["ready_count"])
            if d >= 0:
                return (0, d, r["ready_count"])
            return (1, -d, r["ready_count"])

        rows.sort(key=_sort_key)
        return {
            "kpis": _compute_kpis(rows, today),
            "feishu": feishu_meta,
            "rows": rows,
        }

    async def _build_feishu_meta(self) -> dict[str, Any]:
        feishu_target = (
            await self.db.execute(
                select(IdentitySyncTarget)
                .where(
                    IdentitySyncTarget.kind == _FEISHU_TARGET_KIND,
                    IdentitySyncTarget.deleted_at.is_(None),
                )
                .order_by(IdentitySyncTarget.enabled.desc())
                .limit(1)
            )
        ).scalar_one_or_none()
        feishu_ready = bool(
            feishu_target is not None
            and feishu_target.enabled
            and (
                feishu_target.last_run_status in _HEALTHY_RUN_STATUSES
            )
        )
        return {
            "target_id": (
                str(feishu_target.id) if feishu_target is not None else None
            ),
            "enabled": (
                bool(feishu_target.enabled) if feishu_target else False
            ),
            "last_run_status": (
                feishu_target.last_run_status if feishu_target else None
            ),
            "ready": feishu_ready,
        }


def _empty_kpis() -> dict[str, Any]:
    return {
        "total": 0,
        "this_week": 0,
        "fully_ready": 0,
        "needs_attention": 0,
        "avg_readiness_pct": 0,
        "dimension_ready": {d: 0 for d in DIMENSIONS},
    }


def _compute_kpis(rows: list[dict[str, Any]], today: date) -> dict[str, Any]:
    total = len(rows)
    # "This week" = hires landing within ±7 days of today (just arrived
    # OR arriving within the week). The board now includes already-onboard
    # ``live`` rows whose hire_date may be well in the past, so the old
    # ``hire_date <= today+7`` test (which a hire from 30 days ago also
    # satisfies) would over-count — bound both sides via days_to_hire.
    this_week = sum(
        1
        for r in rows
        if r["days_to_hire"] is not None and -7 <= r["days_to_hire"] <= 7
    )
    fully_ready = sum(1 for r in rows if r["fully_ready"])
    needs_attention = total - fully_ready
    total_dims = len(DIMENSIONS)
    avg_pct = (
        round(sum(r["ready_count"] for r in rows) / (total * total_dims) * 100)
        if total
        else 0
    )
    dimension_ready = {
        d: sum(1 for r in rows if r["dimensions"][d]) for d in DIMENSIONS
    }
    return {
        "total": total,
        "this_week": this_week,
        "fully_ready": fully_ready,
        "needs_attention": needs_attention,
        "avg_readiness_pct": avg_pct,
        "dimension_ready": dimension_ready,
    }