"""我的 IT 足迹 — a permanent, personal "year-in-review"-style recap of one
employee's life on the platform.

This is the warm counterpart to the utilitarian ``/me`` IT-home: instead of
"here is everything you currently hold and must manage", it answers "what has
your story here been so far" — when you joined and where you sit in the line of
people who came before you, your very first device and the one that has been
with you longest, how many software seats you've ever carried, how many times IT
has had your back, and a curated timeline of milestones.

Read-only, self-scoped (no target param → no IDOR surface). Every query is a
tight indexed read over existing domain tables; nothing here mutates. The
backend stays language-agnostic — milestone copy is emitted as ``title_key`` +
``title_params`` so the frontend renders it in zh-CN / en-US / ja-JP.

By product decision (2026-06-21) the recap never names a device's *previous*
owner — a device's lineage is the gallery's concern, the footprint is about the
relationship between you and your kit.
"""
from __future__ import annotations

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

from pydantic import BaseModel
from sqlalchemy import func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.tenure import completed_years
from app.core.working_hours import resolve_timezone
from app.domains.asset.models import AssetHistory, AssetItem, AssetOperation
from app.domains.permission.models import UserPermission
from app.domains.support.models import (
    STATUS_CLOSED,
    STATUS_RESOLVED,
    SupportTicket,
)
from app.identity.person_models import IdentityPerson

logger = logging.getLogger(__name__)

# Operation / history action types that mean "this device came into your hands".
# These MUST match what the asset service actually writes:
#   * AssetOperation.operation_type ∈ {transfer, claim, borrow} carry a to_user_id.
#     (NB: the primary admin "assign device to employee" path writes NO operation —
#     only an AssetHistory row — so the history trail below is essential, not
#     redundant.)
#   * AssetHistory.action ∈ {assigned, transferred} (past tense) carry a to_user.
# Borrow is intentionally included — a loaned device still rode along with you —
# but the gallery is where the held-vs-borrowed distinction is drawn out.
_HELD_OPERATION_TYPES = ("transfer", "claim", "borrow")
_HELD_HISTORY_ACTIONS = ("assigned", "transferred")

# Anniversary marks worth calling out on the timeline. Every passed year is real,
# but surfacing all of them would bury the page; these are the ones that feel
# like a milestone.
_MILESTONE_ANNIVERSARY_YEARS = (1, 2, 3, 5, 7, 10, 15, 20)


class FootprintMilestone(BaseModel):
    """One dot on the personal timeline. Copy is rendered frontend-side from
    ``title_key`` + ``params`` so the recap speaks the viewer's language."""

    kind: str  # 'join' | 'anniversary' | 'first_device' | 'first_ticket'
    date: str  # ISO date
    title_key: str
    params: dict[str, Any] = {}
    icon: str  # emoji flourish for the timeline dot


class FootprintResponse(BaseModel):
    person: dict[str, Any]
    tenure: dict[str, Any]
    devices: dict[str, Any]
    seats: dict[str, Any]
    permissions: dict[str, Any]
    tickets: dict[str, Any]
    milestones: list[FootprintMilestone]
    generated_at: str


def _days_between(later: date, earlier: date) -> int:
    return max(0, (later - earlier).days)


async def build_my_footprint(
    db: AsyncSession, person: IdentityPerson,
) -> FootprintResponse:
    """Assemble the caller's personal IT recap. Self-scoped: ``person`` is
    always the authenticated caller."""
    tz = resolve_timezone(person)
    today = datetime.now(tz).date()
    pid = person.id

    tenure = await _tenure(db, person, today)
    devices = await _devices(db, pid, today)
    seats = await _seats(db, pid)
    permissions = await _permissions(db, pid)
    tickets = await _tickets(db, pid)
    milestones = _build_milestones(person, today, devices, tickets)

    return FootprintResponse(
        person={
            "id": str(person.id),
            "display_name": person.display_name or person.work_email,
            "employee_no": person.employee_no,
            "department": person.primary_department,
            "avatar_url": person.avatar_url,
            "employment_status": person.employment_status,
        },
        tenure=tenure,
        devices=devices,
        seats=seats,
        permissions=permissions,
        tickets=tickets,
        milestones=milestones,
        generated_at=datetime.now(tz).isoformat(),
    )


async def _tenure(
    db: AsyncSession, person: IdentityPerson, today: date,
) -> dict[str, Any]:
    """Where the caller sits in the line of people who joined before them, plus
    plain司龄 (days / completed years). ``join_rank`` is historical truth — it
    counts everyone who joined earlier regardless of whether they've since left,
    so an old-timer keeps their "你是第 N 位伙伴" badge even as early colleagues
    move on. The current-headcount stat deliberately excludes leavers (the
    headcount-display convention)."""
    hire_date = person.hire_date
    # ``created_at`` is a server-default column: reading it as a Python
    # attribute mid-statement-build lazy-loads and trips MissingGreenlet on a
    # session that hasn't refreshed the row. Compare via a scalar subquery, the
    # same way the year recap does.
    my_created = (
        select(IdentityPerson.created_at)
        .where(IdentityPerson.id == person.id)
        .scalar_subquery()
    )
    join_rank: int | None = None
    if hire_date is not None:
        # People who joined strictly before me (non-deleted rows only — a
        # hard-deleted person never existed for ranking purposes). Tie-break by
        # created_at then id so a same-day cohort gets a stable, distinct order.
        earlier = await db.scalar(
            select(func.count())
            .select_from(IdentityPerson)
            .where(
                IdentityPerson.deleted_at.is_(None),
                IdentityPerson.hire_date.is_not(None),
                or_(
                    IdentityPerson.hire_date < hire_date,
                    (IdentityPerson.hire_date == hire_date)
                    & (IdentityPerson.created_at < my_created),
                ),
            )
        )
        join_rank = int(earlier or 0) + 1

    current_headcount = await db.scalar(
        select(func.count())
        .select_from(IdentityPerson)
        .where(
            IdentityPerson.deleted_at.is_(None),
            IdentityPerson.employment_status != "left",
        )
    )

    # Colleagues STILL HERE who joined after this person. Its own query, not
    # ``current_headcount - join_rank``: the rank counts leavers and the
    # headcount excludes them, so once cumulative hires exceeded current
    # headcount the subtraction went negative and clamped to 0 — every recent
    # joiner read "第 N 位加入 · 早于 0 位同事". Same fix as the year recap.
    earlier_than_current = 0
    if hire_date is not None:
        earlier_than_current = int(await db.scalar(
            select(func.count())
            .select_from(IdentityPerson)
            .where(
                IdentityPerson.deleted_at.is_(None),
                IdentityPerson.employment_status != "left",
                IdentityPerson.hire_date.is_not(None),
                IdentityPerson.id != person.id,
                ~or_(
                    IdentityPerson.hire_date < hire_date,
                    (IdentityPerson.hire_date == hire_date)
                    & (IdentityPerson.created_at < my_created),
                ),
            )
        ) or 0)

    return {
        "hire_date": hire_date.isoformat() if hire_date else None,
        "days": _days_between(today, hire_date) if hire_date else None,
        "years": completed_years(hire_date, today) if hire_date else None,
        "join_rank": join_rank,
        "earlier_than_current": earlier_than_current,
        "current_headcount": int(current_headcount or 0),
    }


async def _devices(
    db: AsyncSession, pid: UUID, today: date,
) -> dict[str, Any]:
    """Every device that has ever been in the caller's hands (operations +
    legacy history), the first one, the current ones, and which current device
    has stayed longest. Never surfaces a previous owner."""
    # earliest "came to me" timestamp per asset, unioned across both trails.
    first_seen: dict[UUID, datetime] = {}

    op_rows = (await db.execute(
        select(AssetOperation.asset_id, AssetOperation.created_at).where(
            AssetOperation.to_user_id == pid,
            AssetOperation.status == "confirmed",
            AssetOperation.operation_type.in_(_HELD_OPERATION_TYPES),
        )
    )).all()
    for asset_id, created_at in op_rows:
        if asset_id is None or created_at is None:
            continue
        cur = first_seen.get(asset_id)
        if cur is None or created_at < cur:
            first_seen[asset_id] = created_at

    hist_rows = (await db.execute(
        select(AssetHistory.asset_id, AssetHistory.created_at).where(
            AssetHistory.to_user == pid,
            AssetHistory.action.in_(_HELD_HISTORY_ACTIONS),
        )
    )).all()
    for asset_id, created_at in hist_rows:
        if asset_id is None or created_at is None:
            continue
        cur = first_seen.get(asset_id)
        if cur is None or created_at < cur:
            first_seen[asset_id] = created_at

    total_ever = len(first_seen)

    # First device — earliest first-seen across history. Resolve its display
    # info (it may be soft-deleted / disposed by now; that's fine, it's a
    # memory).
    first_device: dict[str, Any] | None = None
    if first_seen:
        first_asset_id, first_at = min(first_seen.items(), key=lambda kv: kv[1])
        a = await db.get(AssetItem, first_asset_id)
        if a is not None:
            first_device = {
                "id": str(a.id),
                "asset_name": a.asset_name,
                "category": a.category,
                "brand": a.brand,
                "model": a.model,
                "received_at": first_at.isoformat(),
                "days": _days_between(today, first_at.date()),
            }

    # Current devices held by the caller (active rows only).
    current_rows = list((await db.execute(
        select(AssetItem).where(
            AssetItem.assigned_to == pid,
            AssetItem.deleted_at.is_(None),
        )
    )).scalars().all())
    current_count = len(current_rows)

    longest: dict[str, Any] | None = None
    for a in current_rows:
        if a.assigned_at is None:
            continue
        days = _days_between(today, a.assigned_at.date())
        if longest is None or days > longest["days"]:
            longest = {
                "id": str(a.id),
                "asset_name": a.asset_name,
                "category": a.category,
                "brand": a.brand,
                "model": a.model,
                "days": days,
                "assigned_at": a.assigned_at.isoformat(),
            }

    return {
        "total_ever": total_ever,
        "current_count": current_count,
        "first_device": first_device,
        "longest_companion": longest,
    }


async def _seats(db: AsyncSession, pid: UUID) -> dict[str, Any]:
    """Current vs. ever-held software seats — imported lazily to keep the
    asset-model import block focused (the seat models live in the asset
    domain's models module alongside the catalog)."""
    from app.domains.asset.models import SoftwareSeatAllocation

    current = await db.scalar(
        select(func.count()).select_from(SoftwareSeatAllocation).where(
            SoftwareSeatAllocation.user_id == pid,
            SoftwareSeatAllocation.status == "active",
        )
    )
    ever = await db.scalar(
        select(func.count(func.distinct(SoftwareSeatAllocation.plan_id)))
        .select_from(SoftwareSeatAllocation)
        .where(SoftwareSeatAllocation.user_id == pid)
    )
    return {"current": int(current or 0), "total_ever": int(ever or 0)}


async def _permissions(db: AsyncSession, pid: UUID) -> dict[str, Any]:
    active = await db.scalar(
        select(func.count()).select_from(UserPermission).where(
            UserPermission.user_id == pid,
            UserPermission.status == "active",
        )
    )
    return {"active": int(active or 0)}


async def _tickets(db: AsyncSession, pid: UUID) -> dict[str, Any]:
    """How many times the caller has reached out to IT, and how many were seen
    through to a resolution. ``first_at`` seeds the 'first time you asked IT for
    help' milestone."""
    total = await db.scalar(
        select(func.count()).select_from(SupportTicket).where(
            SupportTicket.requester_id == pid,
        )
    )
    resolved = await db.scalar(
        select(func.count()).select_from(SupportTicket).where(
            SupportTicket.requester_id == pid,
            SupportTicket.status.in_((STATUS_RESOLVED, STATUS_CLOSED)),
        )
    )
    first_at = await db.scalar(
        select(func.min(SupportTicket.created_at)).where(
            SupportTicket.requester_id == pid,
        )
    )
    return {
        "total": int(total or 0),
        "resolved": int(resolved or 0),
        "first_at": first_at.isoformat() if first_at else None,
    }


def _build_milestones(
    person: IdentityPerson,
    today: date,
    devices: dict[str, Any],
    tickets: dict[str, Any],
) -> list[FootprintMilestone]:
    """A curated, chronological timeline. Kept deliberately sparse — joining,
    the anniversaries that feel like milestones, the first device, and the first
    time IT had your back."""
    out: list[FootprintMilestone] = []

    hire_date = person.hire_date
    if hire_date is not None:
        out.append(FootprintMilestone(
            kind="join",
            date=hire_date.isoformat(),
            title_key="footprint.milestone.join",
            params={},
            icon="🌱",
        ))
        years = completed_years(hire_date, today)
        for y in _MILESTONE_ANNIVERSARY_YEARS:
            if y > years:
                break
            try:
                anniv = hire_date.replace(year=hire_date.year + y)
            except ValueError:
                # Feb-29 hire date in a non-leap anniversary year → Feb-28.
                anniv = hire_date.replace(year=hire_date.year + y, day=28)
            out.append(FootprintMilestone(
                kind="anniversary",
                date=anniv.isoformat(),
                title_key="footprint.milestone.anniversary",
                params={"years": y},
                icon="🎉",
            ))

    first_device = devices.get("first_device")
    if first_device and first_device.get("received_at"):
        out.append(FootprintMilestone(
            kind="first_device",
            date=first_device["received_at"][:10],
            title_key="footprint.milestone.firstDevice",
            params={
                "name": first_device.get("asset_name")
                or first_device.get("model")
                or first_device.get("category")
                or "",
            },
            icon="💻",
        ))

    first_ticket_at = tickets.get("first_at")
    if first_ticket_at:
        out.append(FootprintMilestone(
            kind="first_ticket",
            date=first_ticket_at[:10],
            title_key="footprint.milestone.firstTicket",
            params={},
            icon="🤝",
        ))

    out.sort(key=lambda m: m.date)
    return out


__all__ = ["build_my_footprint", "FootprintResponse", "FootprintMilestone"]