"""备机闭环 — hardware prep worklist for incoming hires (PR3, 2026-06-12).

Closes the gap between "the manager declared what hardware the hire
needs" (``onboarding_processes.hardware_allocation`` JSON, set on
/my/team-joiners) and "a device is assigned on Day-1" (the
``hardware_allocation`` readiness blocker only resolves when an
``asset_items`` row with a computer category lands on the person).

Three pieces, all asset-domain (asset admins are the audience; reads of
``onboarding_*`` are model-level only — no service import):

* ``build_queue``      — the 待备机 worklist: every open onboarding
                         process inside the window with its hardware
                         readiness, manager declaration, reservation
                         state and office-stock hint; plus stale
                         reservations (process gone terminal but the
                         earmark survived).
* ``list_candidates``  — reservable in-stock devices for one hire.
* ``reserve / release``— earmark a specific device for the hire
                         (voucher: ``AssetOperation`` rows
                         ``reserve_for_hire`` / ``reserve_release``).
                         The reservation is enforced by
                         ``AssetService._ensure_not_reserved_for_other``
                         on every grant path and auto-clears when the
                         device is assigned to the reserved person.
"""
from __future__ import annotations

import logging
from datetime import UTC, datetime, timedelta
from typing import Any
from uuid import UUID
from zoneinfo import ZoneInfo

from sqlalchemy import and_, func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.audit.service import AuditService
from app.core.exceptions import AppException, NotFoundError
from app.domains.asset.models import AssetItem, AssetOperation
from app.domains.asset.service import _generate_operation_number
from app.domains.onboarding.models import (
    PROC_OPEN_STATUSES,
    RES_HARDWARE_ALLOCATION,
    RESO_PENDING,
    OnboardingBlocker,
    OnboardingProcess,
)
from app.identity.person_models import IdentityPerson

logger = logging.getLogger(__name__)

_APP_TZ = ZoneInfo("Asia/Shanghai")

# Mirrors the onboarding readiness scanner's computer categories.
COMPUTER_CATEGORIES = ("laptop", "desktop", "computer")

# Stock statuses a device can be reserved from. ``held_by_leaver`` is
# deliberately included — earmarking a departing colleague's machine
# for the incoming hire is the inherit_from_leaver flow.
RESERVABLE_STATUSES = (
    "idle", "available", "held_by_office", "held_by_leaver",
)


def _today():
    return datetime.now(_APP_TZ).date()


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

    # ───────────────────────── Worklist ─────────────────────────

    async def build_queue(self, *, window_days: int = 14) -> dict[str, Any]:
        """The 待备机 worklist.

        Rows = open onboarding processes whose ``hire_date`` is within
        ``[anything past, today + window_days]`` (past-due hires that
        still lack hardware stay on the list — they're the most urgent
        rows, not the least). Each row carries hardware readiness, the
        manager's declared allocation, the current reservation and a
        global in-stock count for at-a-glance feasibility.

        Visibility note (security review, 2026-06-12): the worklist is
        deliberately ORG-WIDE for every asset admin (incl. region-
        scoped ones) — prep often crosses offices (relocating hires,
        stock borrowed between sites) and the onboarding cockpit
        already shows the same hires to admin-read roles. Writes
        (reserve / release) stay region-gated.
        """
        today = _today()
        window_end = today + timedelta(days=window_days)

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

        person_ids = [p.person_id for p in procs]
        persons: dict[UUID, IdentityPerson] = {}
        if person_ids:
            persons = {
                p.id: p
                for p in (
                    await self.db.execute(
                        select(IdentityPerson).where(
                            IdentityPerson.id.in_(person_ids)
                        )
                    )
                ).scalars()
            }

        # Hardware blocker state per process.
        pending_hw: set[UUID] = set()
        if procs:
            rows = (
                await self.db.execute(
                    select(OnboardingBlocker.process_id).where(
                        OnboardingBlocker.process_id.in_(
                            [p.id for p in procs]
                        ),
                        OnboardingBlocker.resource_type
                        == RES_HARDWARE_ALLOCATION,
                        OnboardingBlocker.resolution == RESO_PENDING,
                    )
                )
            ).scalars()
            pending_hw = set(rows)

        # Reservations pointed at these hires.
        reserved_by_person: dict[UUID, AssetItem] = {}
        if person_ids:
            for a in (
                await self.db.execute(
                    select(AssetItem).where(
                        AssetItem.reserved_for_person_id.in_(person_ids),
                        AssetItem.deleted_at.is_(None),
                    )
                )
            ).scalars():
                reserved_by_person[a.reserved_for_person_id] = a

        # Global in-stock computer count (feasibility hint).
        stock_count = int(
            (
                await self.db.execute(
                    select(func.count())
                    .select_from(AssetItem)
                    .where(
                        AssetItem.deleted_at.is_(None),
                        AssetItem.category.in_(COMPUTER_CATEGORIES),
                        AssetItem.status.in_(
                            ("idle", "available", "held_by_office")
                        ),
                        AssetItem.reserved_for_person_id.is_(None),
                    )
                )
            ).scalar_one()
        )

        rows_out: list[dict[str, Any]] = []
        for proc in procs:
            person = persons.get(proc.person_id)
            if person is None:
                continue
            hw_pending = proc.id in pending_hw
            reserved = reserved_by_person.get(proc.person_id)
            # Hardware satisfied AND nothing reserved → nothing to do;
            # keep the row only while there's prep work or a live
            # reservation to track.
            if not hw_pending and reserved is None:
                continue
            rows_out.append(
                {
                    "process_id": str(proc.id),
                    "person_id": str(proc.person_id),
                    "display_name": person.display_name or person.work_email,
                    "work_email": person.work_email,
                    "department": person.company_name,
                    "job_title": person.job_title,
                    "work_location": person.work_location_name
                    or person.work_location_code,
                    "hire_date": proc.hire_date.isoformat(),
                    "days_to_hire": (proc.hire_date - today).days,
                    "process_status": proc.status,
                    "sla_breach_level": proc.sla_breach_level,
                    "hardware_pending": hw_pending,
                    "hardware_allocation": proc.hardware_allocation,
                    "reserved_asset": (
                        _serialize_asset_brief(reserved)
                        if reserved is not None
                        else None
                    ),
                }
            )
        rows_out.sort(key=lambda r: (r["hire_date"], r["display_name"] or ""))

        stale = await self.stale_reservations()

        unprepped = [
            r for r in rows_out
            if r["hardware_pending"] and r["reserved_asset"] is None
        ]
        kpis = {
            "total": len(rows_out),
            "needs_prep": len(unprepped),
            "within_3d_unprepped": sum(
                1 for r in unprepped if r["days_to_hire"] <= 3
            ),
            "reserved": sum(
                1 for r in rows_out if r["reserved_asset"] is not None
            ),
            "stock_available": stock_count,
            "stale_reservations": len(stale),
        }
        return {"kpis": kpis, "rows": rows_out, "stale_reservations": stale}

    async def stale_reservations(self) -> list[dict[str, Any]]:
        """Reservations whose hire no longer has an open process.

        Happens when an offer is rescinded (process cancelled) or the
        process completed without this device. Surfaced as a 可释放
        list so earmarked stock doesn't quietly rot.

        The active set is derived from ALL open processes — NOT the
        window-bounded queue list. ``reserve()`` accepts any open
        process regardless of hire_date, so judging staleness by the
        display window would mark a legitimate far-out earmark as
        releasable the moment the page is viewed at a narrower window
        (review-r1 finding, three independent reviewers).
        """
        active_person_ids = set(
            (
                await self.db.execute(
                    select(OnboardingProcess.person_id).where(
                        OnboardingProcess.status.in_(PROC_OPEN_STATUSES)
                    )
                )
            ).scalars()
        )
        query = select(AssetItem).where(
            AssetItem.reserved_for_person_id.isnot(None),
            AssetItem.deleted_at.is_(None),
        )
        if active_person_ids:
            query = query.where(
                AssetItem.reserved_for_person_id.notin_(active_person_ids)
            )
        assets = list((await self.db.execute(query)).scalars().all())
        if not assets:
            return []
        person_rows = {
            p.id: p
            for p in (
                await self.db.execute(
                    select(IdentityPerson).where(
                        IdentityPerson.id.in_(
                            {a.reserved_for_person_id for a in assets}
                        )
                    )
                )
            ).scalars()
        }
        out = []
        for a in assets:
            p = person_rows.get(a.reserved_for_person_id)
            brief = _serialize_asset_brief(a)
            brief["reserved_for_display_name"] = (
                (p.display_name or p.work_email) if p else None
            )
            out.append(brief)
        return out

    # ───────────────────────── Candidates ─────────────────────────

    async def list_candidates(
        self,
        process_id: UUID,
        *,
        q: str | None = None,
        limit: int = 50,
    ) -> list[dict[str, Any]]:
        """Reservable devices for one hire (in-stock computers, plus the
        leaver-held ones, minus anything earmarked for someone else)."""
        proc = await self._get_process(process_id)
        conditions = [
            AssetItem.deleted_at.is_(None),
            AssetItem.category.in_(COMPUTER_CATEGORIES),
            AssetItem.status.in_(RESERVABLE_STATUSES),
            or_(
                AssetItem.reserved_for_person_id.is_(None),
                AssetItem.reserved_for_person_id == proc.person_id,
            ),
        ]
        if q:
            kw = f"%{q.strip()}%"
            conditions.append(
                or_(
                    AssetItem.asset_tag.like(kw),
                    AssetItem.asset_name.like(kw),
                    AssetItem.model.like(kw),
                    AssetItem.brand.like(kw),
                    AssetItem.serial_number.like(kw),
                )
            )
        assets = list(
            (
                await self.db.execute(
                    select(AssetItem)
                    .where(and_(*conditions))
                    .order_by(AssetItem.status, AssetItem.updated_at.desc())
                    .limit(max(1, min(200, limit)))
                )
            )
            .scalars()
            .all()
        )
        return [_serialize_asset_brief(a) for a in assets]

    # ───────────────────────── Reserve / release ─────────────────────────

    async def reserve(
        self,
        process_id: UUID,
        asset_id: UUID,
        *,
        trace_id: str,
        actor_user_id: UUID,
    ) -> AssetItem:
        proc = await self._get_process(process_id)
        if proc.status not in PROC_OPEN_STATUSES:
            raise AppException(
                code="ONBOARDING_TERMINAL",
                message=f"onboarding process is {proc.status}",
                status_code=409,
            )
        # Serialize concurrent reserves for the SAME hire on the person
        # row. The "one device per hire" cleanup below reads the person's
        # other earmarks, but two concurrent reserves of *different*
        # assets A and B each lock only their own target row — under
        # MVCC neither sees the other's uncommitted earmark, so both
        # would commit and leave the hire double-reserved. Locking the
        # person first funnels them through one at a time.
        await self.db.execute(
            select(IdentityPerson.id)
            .where(IdentityPerson.id == proc.person_id)
            .with_for_update()
        )
        asset = (
            await self.db.execute(
                select(AssetItem)
                .where(
                    AssetItem.id == asset_id,
                    AssetItem.deleted_at.is_(None),
                )
                .with_for_update()
                # The router's region gate already loaded this row into
                # the identity map; without populate_existing the
                # post-lock object would keep the PRE-lock attribute
                # snapshot and the status/reservation checks below
                # could race a concurrent reserve (review-r1).
                .execution_options(populate_existing=True)
            )
        ).scalar_one_or_none()
        if asset is None:
            raise NotFoundError("Asset", str(asset_id))
        if (
            asset.status not in RESERVABLE_STATUSES
            # A device already earmarked for THIS hire stays reservable
            # for the idempotent short-circuit below even if a careless
            # status edit moved it out of the stock set.
            and asset.reserved_for_person_id != proc.person_id
        ):
            raise AppException(
                code="ASSET_NOT_RESERVABLE",
                message=(
                    f"Asset is {asset.status}; only in-stock or "
                    "leaver-held devices can be reserved"
                ),
                status_code=409,
            )
        if (
            asset.reserved_for_person_id is not None
            and asset.reserved_for_person_id != proc.person_id
        ):
            raise AppException(
                code="ASSET_RESERVED",
                message="Asset is already reserved for another hire",
                status_code=409,
            )
        if asset.reserved_for_person_id == proc.person_id:
            # Idempotent re-reserve of the same device: no duplicate
            # voucher / audit rows (MCP retries and double-submits hit
            # this; the FE re-entry guard alone isn't a contract).
            return asset
        # One device per hire: release any prior earmark for this
        # person first (re-reserving a different machine is the common
        # correction flow; two simultaneous earmarks would double-hold
        # stock). Each release gets its own voucher so the device's
        # operation trail explains why the earmark vanished.
        #
        # Region note (security review, 2026-06-12): the prior earmark
        # may live on an asset OUTSIDE the acting admin's region. This
        # cleanup deliberately crosses that scope — it only UN-sets an
        # earmark for the SAME hire the admin is legitimately
        # re-provisioning (no ownership change, fully vouchered below);
        # blocking it would strand the one-device-per-hire invariant
        # whenever a hire relocates offices mid-prep.
        prior = (
            await self.db.execute(
                select(AssetItem)
                .where(
                    AssetItem.reserved_for_person_id == proc.person_id,
                    AssetItem.id != asset_id,
                    AssetItem.deleted_at.is_(None),
                )
                .with_for_update()
            )
        ).scalars().all()
        for old in prior:
            old.reserved_for_person_id = None
            old.reserved_at = None
            self.db.add(
                AssetOperation(
                    operation_number=_generate_operation_number(),
                    asset_id=old.id,
                    operation_type="reserve_release",
                    status="confirmed",
                    from_user_id=proc.person_id,
                    operator_id=actor_user_id,
                    confirmed_at=datetime.now(UTC),
                    confirmed_by=actor_user_id,
                    notes=(
                        f"Released — hire re-reserved onto another "
                        f"device (process {proc.id})"
                    ),
                    detail={
                        "via": "one_device_per_hire_cleanup",
                        "replacement_asset_id": str(asset_id),
                        "process_id": str(proc.id),
                    },
                )
            )

        asset.reserved_for_person_id = proc.person_id
        asset.reserved_at = datetime.now(UTC)
        self.db.add(
            AssetOperation(
                operation_number=_generate_operation_number(),
                asset_id=asset.id,
                operation_type="reserve_for_hire",
                status="confirmed",
                to_user_id=proc.person_id,
                operator_id=actor_user_id,
                confirmed_at=datetime.now(UTC),
                confirmed_by=actor_user_id,
                notes=f"Reserved for hire (process {proc.id})",
            )
        )
        await self.db.flush()
        await self.audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=actor_user_id,
            action="asset.reserved_for_hire",
            status="success",
            resource_type="asset_item",
            resource_id=str(asset.id),
            detail={
                "process_id": str(proc.id),
                "person_id": str(proc.person_id),
                "released_prior": [str(o.id) for o in prior] or None,
            },
        )
        return asset

    async def sweep_stale_reservations(
        self, *, trace_id: str, dry_run: bool = False,
    ) -> dict[str, Any]:
        """Release earmarks whose hire no longer has an open process.

        Why a cron (2026-09-05): ``stale_reservations`` only ran when
        somebody opened the 待备机 tab, and nothing else in the platform
        looked at reservations at all — no scheduled job referenced them
        and ``delete_asset`` neither refused nor cleared a reserved row,
        so a soft delete made the earmark vanish silently. A rescinded
        offer therefore held a laptop out of stock indefinitely, and
        since 2026-09-05 an earmark ALSO satisfies the Day-1 hardware
        dimension — a zombie earmark is now a device that is both
        unavailable and quietly counted as somebody's Day-1 readiness.

        Releases through the same voucher path a human release takes
        (``reserve_release`` operation), tagged
        ``detail.via='stale_reservation_sweep'``. ``operator_id`` is the
        hire the earmark was held for — the semantic subject, per the
        offboarding-automation convention, since the column is NOT NULL
        and a cron has no acting admin.
        """
        stale = await self.stale_reservations()
        if not stale:
            return {"candidates": 0, "released": 0, "dry_run": dry_run}
        released = 0
        now = datetime.now(UTC)
        for brief in stale:
            asset_id = UUID(str(brief["id"]))
            asset = (
                await self.db.execute(
                    select(AssetItem)
                    .where(
                        AssetItem.id == asset_id,
                        AssetItem.deleted_at.is_(None),
                    )
                    .with_for_update()
                    .execution_options(populate_existing=True)
                )
            ).scalar_one_or_none()
            # Re-assert under the lock, against the SAME person the
            # stale list judged. The list read is lock-free, so between
            # it and here a human release or — the case that makes the
            # person check load-bearing — a re-reserve of this device
            # for a DIFFERENT hire may have landed. Checking only
            # "still reserved" would then release that new hire's
            # legitimate earmark: their process is open, so the device
            # was never stale for them.
            if (
                asset is None
                or asset.reserved_for_person_id is None
                or str(asset.reserved_for_person_id)
                != str(brief.get("reserved_for_person_id"))
            ):
                continue
            prior_person = asset.reserved_for_person_id
            if dry_run:
                released += 1
                continue
            asset.reserved_for_person_id = None
            asset.reserved_at = None
            self.db.add(
                AssetOperation(
                    operation_number=_generate_operation_number(),
                    asset_id=asset.id,
                    operation_type="reserve_release",
                    status="confirmed",
                    from_user_id=prior_person,
                    operator_id=prior_person,
                    confirmed_at=now,
                    confirmed_by=prior_person,
                    notes=(
                        "Auto-released: the hire this device was "
                        "earmarked for no longer has an open onboarding "
                        "process"
                    ),
                    detail={
                        "via": "stale_reservation_sweep",
                        "reserved_for_person_id": str(prior_person),
                    },
                )
            )
            await self.audit.log(
                trace_id=trace_id,
                actor_type="system",
                action="asset.reservation_auto_released",
                status="success",
                resource_type="asset_item",
                resource_id=str(asset.id),
                detail={
                    "via": "stale_reservation_sweep",
                    "reserved_for_person_id": str(prior_person),
                    "asset_tag": asset.asset_tag,
                },
            )
            released += 1
        if released and not dry_run:
            await self.db.flush()
        logger.info(
            "asset.stale_reservation_sweep: candidates=%d released=%d "
            "dry_run=%s",
            len(stale), released, dry_run,
        )
        return {
            "candidates": len(stale),
            "released": released,
            "dry_run": dry_run,
        }

    async def release(
        self,
        asset_id: UUID,
        *,
        trace_id: str,
        actor_user_id: UUID,
        reason: str | None = None,
    ) -> AssetItem:
        asset = (
            await self.db.execute(
                select(AssetItem)
                .where(
                    AssetItem.id == asset_id,
                    AssetItem.deleted_at.is_(None),
                )
                .with_for_update()
                # See reserve() — refresh past the router's region-gate
                # identity-map snapshot.
                .execution_options(populate_existing=True)
            )
        ).scalar_one_or_none()
        if asset is None:
            raise NotFoundError("Asset", str(asset_id))
        if asset.reserved_for_person_id is None:
            return asset
        prior_person = asset.reserved_for_person_id
        asset.reserved_for_person_id = None
        asset.reserved_at = None
        self.db.add(
            AssetOperation(
                operation_number=_generate_operation_number(),
                asset_id=asset.id,
                operation_type="reserve_release",
                status="confirmed",
                from_user_id=prior_person,
                operator_id=actor_user_id,
                confirmed_at=datetime.now(UTC),
                confirmed_by=actor_user_id,
                notes=(reason or "").strip()[:512] or "Reservation released",
            )
        )
        await self.db.flush()
        await self.audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=actor_user_id,
            action="asset.reservation_released",
            status="success",
            resource_type="asset_item",
            resource_id=str(asset.id),
            detail={
                "prior_person_id": str(prior_person),
                "reason": (reason or "").strip()[:512] or None,
            },
        )
        return asset

    # ───────────────────────── Helpers ─────────────────────────

    async def _get_process(self, process_id: UUID) -> OnboardingProcess:
        proc = await self.db.get(OnboardingProcess, process_id)
        if proc is None:
            raise NotFoundError("OnboardingProcess", str(process_id))
        return proc


def _serialize_asset_brief(a: AssetItem) -> dict[str, Any]:
    return {
        "id": str(a.id),
        "asset_tag": a.asset_tag,
        "asset_name": a.asset_name,
        "category": a.category,
        "brand": a.brand,
        "model": a.model,
        "serial_number": a.serial_number,
        "status": a.status,
        "location": a.location,
        "holder_office_id": (
            str(a.holder_office_id) if a.holder_office_id else None
        ),
        "assigned_to": str(a.assigned_to) if a.assigned_to else None,
        "reserved_for_person_id": (
            str(a.reserved_for_person_id)
            if a.reserved_for_person_id
            else None
        ),
        "reserved_at": (
            a.reserved_at.isoformat() if a.reserved_at else None
        ),
    }