"""Devices still out with departed employees — with a clock on them.

Why (2026-09-05). ``held_by_leaver`` is a real, well-populated state (10
assets in production) and it had **no clock and no owner**: the weekly
digest was ``SELECT count(*)`` plus five sample rows mailed to every
asset admin, with no ageing buckets, nobody accountable, and no path to
closure. A device that leaves with someone is either coming back or
being written off, and neither happens because a number in an email
never names a next action.

The clock this uses is **the leaver's own termination date**, not when IT
got round to marking the asset. "How long has the company been without
this laptop" is the question worth asking; "how long since we noticed"
is not. ``identity_persons.termination_date`` is the authoritative field
(HRIS-owned); the ``mark_held_by_leaver`` operation timestamp is the
fallback for a device marked against a person whose termination date
never landed, and rows fall back to the asset row's own ``updated_at``
only as a last resort — each row reports WHICH clock it used, because a
bucket built on a fallback deserves less confidence than one built on
the HRIS date.

Ageing buckets are 30 / 60 / 90+ days. Not cosmetic: a device two weeks
out is normal offboarding latency, one 90 days out is either lost or
quietly written off, and lumping them into one count is what let the
production 10 sit unexamined.

**Accountability**: each row names the office admin(s) for the device's
location — derived, never stored — because "who should chase this" was
the other half of what the digest never said. Where the location does
not resolve, the row says so instead of silently blaming nobody.
"""
from __future__ import annotations

import logging
from datetime import UTC, date, datetime
from typing import Any, Literal
from uuid import UUID

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

from app.domains.asset.location_admin_sync import (
    get_location_ids_managed_by,
    resolve_asset_location_ids_bulk,
)
from app.domains.asset.models import (
    ASSET_STATUS_HELD_BY_LEAVER,
    AssetItem,
    AssetLocation,
    AssetLocationAdmin,
    AssetOperation,
)
from app.identity.person_models import IdentityPerson

logger = logging.getLogger(__name__)

ClockSource = Literal["termination_date", "marked_at", "row_updated_at"]

# 30 / 60 / 90+. A device two weeks out is normal offboarding latency;
# one 90 days out is lost or quietly written off. One count for both is
# what let production's 10 rows sit unexamined.
BUCKET_EDGES: tuple[int, ...] = (30, 60, 90)


def _bucket(days: int) -> str:
    if days >= BUCKET_EDGES[2]:
        return "d90_plus"
    if days >= BUCKET_EDGES[1]:
        return "d60_89"
    if days >= BUCKET_EDGES[0]:
        return "d30_59"
    return "under_30"


def _as_date(value: Any) -> date | None:
    if value is None:
        return None
    if isinstance(value, datetime):
        return value.astimezone(UTC).date() if value.tzinfo else value.date()
    if isinstance(value, date):
        return value
    return None


class LeaverAssetRecoveryService:
    """Read + reconcile the devices departed employees still hold."""

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

    async def build(
        self,
        *,
        actor_person_id: UUID,
        role_names: list[str],
        bucket: str | None = None,
    ) -> dict[str, Any]:
        from app.core.rbac import has_global_asset_scope

        today = datetime.now(UTC).date()
        assets = (
            await self.db.execute(
                select(AssetItem).where(
                    AssetItem.deleted_at.is_(None),
                    AssetItem.status == ASSET_STATUS_HELD_BY_LEAVER,
                )
            )
        ).scalars().all()
        if not assets:
            return _empty()

        # Region scope BEFORE counting, same rule as every asset surface:
        # a regional admin's numbers must be about their own estate, and
        # an asset whose location resolves to nothing belongs to no region.
        global_tier = has_global_asset_scope(role_names)
        location_ids = await resolve_asset_location_ids_bulk(
            self.db, list(assets),
        )
        if not global_tier:
            managed = await get_location_ids_managed_by(
                self.db, actor_person_id,
            )
            if not managed:
                return _empty()
            assets = [
                a for a in assets if location_ids.get(a.id) in managed
            ]
            if not assets:
                return _empty()

        holder_ids = {a.assigned_to for a in assets if a.assigned_to}
        holders: dict[UUID, IdentityPerson] = {}
        if holder_ids:
            holders = {
                p.id: p
                for p in (
                    await self.db.execute(
                        select(IdentityPerson).where(
                            IdentityPerson.id.in_(holder_ids),
                        )
                    )
                ).scalars()
            }
        marked_at = await self._marked_at(list(assets))
        admins = await self._admins_by_location(
            {lid for lid in location_ids.values() if lid is not None}
        )

        rows: list[dict[str, Any]] = []
        buckets: dict[str, int] = {
            "under_30": 0, "d30_59": 0, "d60_89": 0, "d90_plus": 0,
        }
        clock_sources: dict[str, int] = {}
        for asset in assets:
            holder = holders.get(asset.assigned_to) if asset.assigned_to else None
            held_since, source = self._resolve_clock(
                asset, holder, marked_at.get(asset.id),
            )
            days = max(0, (today - held_since).days)
            b = _bucket(days)
            buckets[b] += 1
            clock_sources[source] = clock_sources.get(source, 0) + 1
            loc_id = location_ids.get(asset.id)
            rows.append(
                {
                    "asset_id": str(asset.id),
                    "asset_tag": asset.asset_tag,
                    "category": asset.category,
                    "brand": asset.brand,
                    "model": asset.model,
                    "serial_number": asset.serial_number,
                    "location": asset.location,
                    "purchase_cost": (
                        str(asset.purchase_cost)
                        if asset.purchase_cost is not None
                        else None
                    ),
                    "currency": asset.currency,
                    # Provenance travels with the amount. Three other
                    # surfaces label an estimated figure; one that
                    # showed the bare number would teach the reader to
                    # trust every number on every page.
                    "purchase_cost_source": asset.purchase_cost_source,
                    "holder_person_id": (
                        str(asset.assigned_to) if asset.assigned_to else None
                    ),
                    "holder_name": (
                        holder.display_name or holder.work_email
                        if holder
                        else None
                    ),
                    "holder_employee_no": holder.employee_no if holder else None,
                    "holder_employment_status": (
                        holder.employment_status if holder else None
                    ),
                    "held_since": held_since.isoformat(),
                    # Which clock produced ``held_since``. A bucket built
                    # on a fallback deserves less confidence than one
                    # built on the HRIS termination date, and the reader
                    # has to be able to tell.
                    "clock_source": source,
                    "days_held": days,
                    "bucket": b,
                    # Derived, never stored: who should chase this.
                    "responsible_admin_ids": [
                        str(pid) for pid in admins.get(loc_id, ())
                    ] if loc_id is not None else [],
                    "location_resolved": loc_id is not None,
                }
            )

        rows.sort(key=lambda r: (-r["days_held"], r["asset_tag"] or ""))
        filtered = (
            [r for r in rows if r["bucket"] == bucket] if bucket else rows
        )
        return {
            "rows": filtered,
            "summary": {
                "total": len(rows),
                "buckets": buckets,
                "clock_sources": clock_sources,
                # Devices whose location does not resolve have no office
                # admin, so nobody is accountable unless a global admin
                # picks them up. Counted so that is visible, not silent.
                "unassignable": sum(
                    1 for r in rows if not r["responsible_admin_ids"]
                ),
                "filtered_count": len(filtered),
                "regional_scope": not global_tier,
                "bucket_edges": list(BUCKET_EDGES),
            },
        }

    def _resolve_clock(
        self,
        asset: AssetItem,
        holder: IdentityPerson | None,
        marked: datetime | None,
    ) -> tuple[date, ClockSource]:
        """Prefer the HRIS termination date over IT's bookkeeping.

        "How long has the company been without this laptop" is the
        question worth asking; "how long since we noticed" is not.
        """
        termination = _as_date(holder.termination_date) if holder else None
        if termination is not None:
            return termination, "termination_date"
        marked_date = _as_date(marked)
        if marked_date is not None:
            return marked_date, "marked_at"
        return (
            _as_date(asset.updated_at) or datetime.now(UTC).date(),
            "row_updated_at",
        )

    async def _marked_at(
        self, assets: list[AssetItem],
    ) -> dict[UUID, datetime | None]:
        """When each asset was last marked held-by-leaver.

        MAX, not MIN: a device recovered and later marked again is held
        since the LATEST mark — the earlier episode is closed history.
        """
        ids = [a.id for a in assets]
        if not ids:
            return {}
        out: dict[UUID, datetime | None] = {}
        for asset_id, stamp in (
            await self.db.execute(
                select(
                    AssetOperation.asset_id,
                    func.max(AssetOperation.created_at),
                )
                .where(
                    AssetOperation.asset_id.in_(ids),
                    AssetOperation.operation_type == "mark_held_by_leaver",
                )
                .group_by(AssetOperation.asset_id)
            )
        ).all():
            out[asset_id] = stamp
        return out

    async def _admins_by_location(
        self, location_ids: set[UUID],
    ) -> dict[UUID, tuple[UUID, ...]]:
        if not location_ids:
            return {}
        rows = (
            await self.db.execute(
                select(
                    AssetLocationAdmin.location_id,
                    AssetLocationAdmin.person_id,
                )
                .join(
                    AssetLocation,
                    AssetLocation.id == AssetLocationAdmin.location_id,
                )
                .where(AssetLocationAdmin.location_id.in_(location_ids))
            )
        ).all()
        grouped: dict[UUID, list[UUID]] = {}
        for location_id, person_id in rows:
            grouped.setdefault(location_id, []).append(person_id)
        return {k: tuple(v) for k, v in grouped.items()}


def _empty() -> dict[str, Any]:
    return {
        "rows": [],
        "summary": {
            "total": 0,
            "buckets": {
                "under_30": 0, "d30_59": 0, "d60_89": 0, "d90_plus": 0,
            },
            "clock_sources": {},
            "unassignable": 0,
            "filtered_count": 0,
            "regional_scope": False,
            "bucket_edges": list(BUCKET_EDGES),
        },
    }