"""Identity golden-record aggregation — the spine of the Identity Hub.

One person is the atom of the identity platform. This service assembles
the *golden record* for a single :class:`IdentityPerson` by joining the
data that today lives scattered across four subsystems:

* **Login identities** — ``identity_external_identities`` rows (Google /
  Feishu bindings, their kill-switch + suspend state, last sync).
* **Downstream projections** — ``identity_sync_state`` rows across every
  sync target the person is (or was) projected into, each annotated with
  a *drift* flag computed by re-rendering the live mapping payload and
  comparing its hash to the last-pushed hash. This reuses the SAME
  rendering path (``IdentitySyncService.preview_payload``) the mapping
  preview drawer uses — golden record and config builder render through
  one engine, never two.
* **Attribute provenance** — derived from existing signals only
  (``source_system`` / ``last_synced_at`` / ``source_updated_at`` + the
  known set of HR-overridable fields). No new per-field history table —
  see the 2026-06-09 design decision (derive-from-signals over a
  field-history migration whose marginal value at 500 people is low).
* **Change timeline** — ``audit_logs`` rows for this person emitted by
  the identity audit listener (insert / update / delete / contact
  reveal), newest first.

Every nested entity links back out (target → /admin/identity-sync,
finding → this record, run → its target) so the hub is navigable in
both directions; the service returns the ids the frontend needs to wire
those links.

Read-only except for :meth:`set_external_identity_login`, the admin
kill-switch / suspend mutation that the golden record exposes inline.
"""
from __future__ import annotations

import uuid
from datetime import datetime
from typing import Any

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

from app.audit.models import AuditLog
from app.audit.service import AuditService
from app.core.exceptions import NotFoundError, ValidationError
from app.core.masking import mask_phone, mask_tail
from app.domains.identity_sync.models import (
    IdentitySyncState,
    IdentitySyncTarget,
)
from app.identity.person_models import (
    IdentityExternalIdentity,
    IdentityPerson,
)

# ── Attribute provenance spec ─────────────────────────────────────────
# Honest, signal-derived provenance: we declare which *category* each
# golden-record attribute comes from. We do NOT claim a per-field
# "overridden on date X" history (that would need a history table we
# deliberately did not build). ``overridable`` marks fields HR can edit
# locally after the first HRIS write, so the UI can show "HR-managed"
# rather than "from HRIS" for them.
#
# source categories:
#   ``hris``      — pulled from the upstream HRIS (SuccessFactors)
#   ``hr_managed``— first-written from HRIS, thereafter HR-editable
#   ``lifecycle`` — driven by the JML state machine + HRIS events
#   ``sso``       — set by the authentication path, not the HRIS
#   ``system``    — granted/derived inside Helix (roles)
_PROVENANCE_SPEC: tuple[tuple[str, str, bool], ...] = (
    ("work_email", "hris", False),
    ("display_name", "hris", False),
    ("employee_no", "hris", False),
    ("job_title", "hris", False),
    ("department", "hris", False),
    ("manager", "hris", False),
    ("country", "hris", False),
    ("city", "hris", False),
    ("company_name", "hris", False),
    ("cost_center", "hris", False),
    ("hire_date", "hris", False),
    ("termination_date", "hris", False),
    ("mobile_phone", "hris", False),
    ("personal_email", "hris", False),
    # GitHub handle — SF ``cust_Other.cust_id`` (migration 369). Sparse:
    # only staff with a handle on file in SF carry a value.
    ("github_id", "hris", False),
    ("employment_status", "lifecycle", False),
    ("birthday", "hr_managed", True),
    ("annual_leave_carry_legal_days", "hr_managed", True),
    ("annual_leave_carry_welfare_days", "hr_managed", True),
    ("avatar_url", "sso", False),
    ("last_login_at", "sso", False),
    ("roles", "system", False),
)

# Sensitive fields whose provenance VALUE is masked at this surface.
# Unmasked values come only via the audited
# ``POST /users/{id}/reveal-contact`` endpoint — the golden record never
# leaks raw PII, even to admins, without an audit row.
_MASKED_PROVENANCE_VALUE = {
    "mobile_phone": mask_phone,
    "personal_email": lambda v: mask_tail(v, visible=4),
}

# Dept-phase targets do not render a person-shaped payload through the
# mapping engine (Phase A builds dept bodies directly), so a per-person
# drift render is meaningless for the dept phase. The user phase of
# ``feishu_contact`` IS person-shaped, so it still renders fine; we only
# skip drift when a render genuinely can't apply. Kept as an explicit
# set so a future dept-only kind is excluded deliberately, not by luck.
_NO_PERSON_PAYLOAD_KINDS: frozenset[str] = frozenset()


def _iso(value: datetime | None) -> str | None:
    return value.isoformat() if value is not None else None


class IdentityProfileService:
    """Assemble + mutate the identity golden record for one person."""

    def __init__(self, db: AsyncSession):
        self.db = db

    # ── Read: full golden record ─────────────────────────────────────
    async def get_identity_profile(
        self, person_id: uuid.UUID, *, include_drift: bool = True,
    ) -> dict[str, Any]:
        person = await self._load_person(person_id)
        return {
            "person": self._person_summary(person),
            "source": self._source_block(person),
            "login_identities": self._login_identities(person),
            "projections": await self._projections(
                person, include_drift=include_drift,
            ),
            "provenance": self._provenance(person),
            "timeline": await self._timeline(person_id),
        }

    async def _load_person(self, person_id: uuid.UUID) -> IdentityPerson:
        person = (await self.db.execute(
            select(IdentityPerson)
            .where(IdentityPerson.id == person_id)
            .options(
                selectinload(IdentityPerson.external_identities),
                selectinload(IdentityPerson.org_memberships),
                selectinload(IdentityPerson.user_roles),
                # ``manager`` is lazy='select' by default — eager-load it
                # here so ``_person_summary`` / ``_provenance_value`` can
                # read ``person.manager`` without tripping MissingGreenlet
                # in the async session.
                selectinload(IdentityPerson.manager),
            )
        )).scalars().first()
        if person is None or person.deleted_at is not None:
            raise NotFoundError("IdentityPerson", str(person_id))
        return person

    def _person_summary(self, person: IdentityPerson) -> dict[str, Any]:
        mgr = person.manager
        return {
            "id": str(person.id),
            "work_email": person.work_email,
            "display_name": person.display_name,
            "avatar_url": person.avatar_url,
            "employee_no": person.employee_no,
            "department": person.department,
            "job_title": person.job_title,
            "employment_status": person.employment_status,
            # Real granted roles, not a requester's X-View-Role narrowing
            # (the current user's own row shares the session identity-map
            # instance that carries ``_role_names_override``). Symmetric with
            # /auth/me, /auth/me/profile and the RBAC workbench list.
            "roles": person.actual_role_names,
            "manager": (
                {
                    "id": str(mgr.id),
                    "display_name": mgr.display_name,
                    "work_email": mgr.work_email,
                }
                if mgr is not None
                else None
            ),
        }

    def _source_block(self, person: IdentityPerson) -> dict[str, Any]:
        """The upstream-source identity card — who owns this record."""
        return {
            "source_system": person.source_system,
            "hris_instance_id": (
                str(person.hris_instance_id)
                if person.hris_instance_id is not None
                else None
            ),
            "hris_person_id": person.hris_person_id,
            "hris_record_state": person.hris_record_state,
            "source_updated_at": _iso(person.source_updated_at),
            "last_synced_at": _iso(person.last_synced_at),
        }

    # ── Login identities (auth bindings) ─────────────────────────────
    def _login_identities(
        self, person: IdentityPerson,
    ) -> list[dict[str, Any]]:
        rows: list[dict[str, Any]] = []
        for ext in person.external_identities:
            rows.append({
                "id": str(ext.id),
                "provider": ext.provider,
                # The provider subject is a durable opaque id (Google
                # ``sub`` / Feishu ``union_id``). Not a secret, but tail-
                # masked here — admins rarely need the full value and a
                # full dump invites copy-paste into the wrong system.
                "provider_subject_masked": mask_tail(
                    ext.provider_subject, visible=6,
                ),
                "provider_email": ext.provider_email,
                "provider_username": ext.provider_username,
                "is_login_identity": ext.is_login_identity,
                "identity_status": ext.identity_status,
                "last_synced_at": _iso(ext.last_synced_at),
                "created_at": _iso(ext.created_at),
                # A tombstoned binding survives (provider, subject) unique
                # so login revival is collision-safe; surface it so the
                # admin sees "previously bound, now removed" distinctly.
                "deleted_at": _iso(ext.deleted_at),
            })
        # Stable order: live bindings first, then by provider name.
        rows.sort(key=lambda r: (r["deleted_at"] is not None, r["provider"]))
        return rows

    # ── Downstream projection matrix ─────────────────────────────────
    async def _projections(
        self, person: IdentityPerson, *, include_drift: bool,
    ) -> list[dict[str, Any]]:
        rows = (await self.db.execute(
            select(IdentitySyncState, IdentitySyncTarget)
            .join(
                IdentitySyncTarget,
                IdentitySyncTarget.id == IdentitySyncState.target_id,
            )
            .where(
                IdentitySyncState.person_id == person.id,
                IdentitySyncTarget.deleted_at.is_(None),
            )
        )).all()

        # Lazily build the sync service only if we need drift renders.
        sync_svc = None
        if include_drift and rows:
            from app.domains.identity_sync.service import IdentitySyncService
            sync_svc = IdentitySyncService(self.db)

        out: list[dict[str, Any]] = []
        for state, target in rows:
            entry: dict[str, Any] = {
                "target_id": str(target.id),
                "target_name": target.name,
                "target_slug": target.slug,
                "target_kind": target.kind,
                "target_enabled": target.enabled,
                "external_id": state.external_id,
                "last_status": state.last_status,
                "last_synced_at": _iso(state.last_synced_at),
                "last_error": state.last_error,
                "retry_count": state.retry_count,
                "auto_included_reason": state.auto_included_reason,
                # drift = the live-rendered payload no longer matches what
                # we last pushed. None = not computed (drift render off, or
                # render not applicable / failed for this target).
                "drift": None,
            }
            if (
                sync_svc is not None
                and target.kind not in _NO_PERSON_PAYLOAD_KINDS
            ):
                entry["drift"] = await self._compute_drift(
                    sync_svc, person.id, state, target,
                )
            out.append(entry)
        out.sort(key=lambda r: r["target_name"].lower())
        return out

    async def _compute_drift(
        self,
        sync_svc: Any,
        person_id: uuid.UUID,
        state: IdentitySyncState,
        target: IdentitySyncTarget,
    ) -> bool | None:
        """True when the live payload hash differs from the last pushed
        hash (only meaningful for a person we believe is in-sync). Render
        failures degrade to ``None`` — never crash the golden record over
        one target's mapping problem.
        """
        # Only states that represent "we pushed something" carry a
        # comparable hash. deactivated / deleted / excluded / pending have
        # no current-payload meaning.
        if state.last_status not in ("success", "skipped"):
            return None
        if not state.last_payload_hash:
            return None
        try:
            preview = await sync_svc.preview_payload(
                person_id=person_id,
                mapping_config=target.mapping_config or {},
                runner_options=target.runner_options or {},
            )
        except Exception:
            return None
        current = preview.get("rendered_payload_hash") or ""
        if not current:
            return None
        return current != state.last_payload_hash

    # ── Attribute provenance (signal-derived) ────────────────────────
    def _provenance(self, person: IdentityPerson) -> list[dict[str, Any]]:
        hris_synced = _iso(person.last_synced_at)
        out: list[dict[str, Any]] = []
        for field, source, overridable in _PROVENANCE_SPEC:
            value = self._provenance_value(person, field)
            out.append({
                "field": field,
                "value": value,
                "source": source,
                "overridable": overridable,
                # Last-synced only carries meaning for HRIS-owned fields;
                # hr_managed / sso / system fields are not stamped by the
                # HRIS sweep, so we don't imply a sync time for them.
                "last_synced_at": hris_synced if source == "hris" else None,
            })
        return out

    def _provenance_value(
        self, person: IdentityPerson, field: str,
    ) -> str | None:
        if field == "manager":
            return person.manager.display_name if person.manager else None
        if field == "department":
            return person.department
        if field == "roles":
            # Real grants, not a requester's narrowed X-View-Role set (see
            # _person_summary): provenance must report what the person holds.
            return (
                ", ".join(person.actual_role_names)
                if person.actual_role_names
                else None
            )
        if field in _MASKED_PROVENANCE_VALUE:
            return _MASKED_PROVENANCE_VALUE[field](getattr(person, field))
        raw = getattr(person, field, None)
        if raw is None:
            return None
        if isinstance(raw, datetime):
            return raw.isoformat()
        # date / Decimal / str → string
        return str(raw)

    # ── Change timeline ──────────────────────────────────────────────
    async def _timeline(
        self, person_id: uuid.UUID, *, limit: int = 50,
    ) -> list[dict[str, Any]]:
        logs = (await self.db.execute(
            select(AuditLog)
            .where(
                AuditLog.resource_type == "identity_person",
                AuditLog.resource_id == str(person_id),
            )
            .order_by(AuditLog.created_at.desc())
            .limit(limit)
        )).scalars().all()

        actor_ids = {
            log.actor_user_id for log in logs if log.actor_user_id is not None
        }
        actor_names = await self._resolve_actor_names(actor_ids)

        out: list[dict[str, Any]] = []
        for log in logs:
            detail = log.detail or {}
            out.append({
                "id": str(log.id),
                "created_at": _iso(log.created_at),
                "action": log.action,
                "status": log.status,
                "actor_user_id": (
                    str(log.actor_user_id)
                    if log.actor_user_id is not None
                    else None
                ),
                "actor_name": actor_names.get(log.actor_user_id),
                "actor_type": log.actor_type,
                "changed_fields": detail.get("changed_fields"),
                "via": detail.get("via"),
            })
        return out

    async def _resolve_actor_names(
        self, actor_ids: set[uuid.UUID],
    ) -> dict[uuid.UUID, str]:
        if not actor_ids:
            return {}
        rows = (await self.db.execute(
            select(IdentityPerson.id, IdentityPerson.display_name)
            .where(IdentityPerson.id.in_(actor_ids))
        )).all()
        return {pid: name for pid, name in rows}

    # ── Mutation: login kill-switch / suspend ────────────────────────
    async def set_external_identity_login(
        self,
        *,
        person_id: uuid.UUID,
        ext_identity_id: uuid.UUID,
        is_login_identity: bool | None,
        identity_status: str | None,
        actor_user_id: uuid.UUID,
        trace_id: str,
        ip_address: str | None,
        user_agent: str | None,
    ) -> dict[str, Any]:
        """Toggle a login binding's kill-switch / suspend state.

        ``is_login_identity=False`` is the real admin kill-switch — the
        subject lookup in the auth path filters on it, so the binding can
        no longer be used to log in (see the 2026-05-29 SSO-resolution
        ADR). ``identity_status='suspended'`` is the softer admin freeze.
        Both are audited.
        """
        if is_login_identity is None and identity_status is None:
            raise ValidationError(
                "at least one of is_login_identity / identity_status "
                "must be provided",
            )
        if identity_status is not None and identity_status not in (
            "active", "suspended",
        ):
            raise ValidationError(
                f"identity_status must be 'active' or 'suspended', "
                f"got {identity_status!r}",
            )

        ext = await self.db.get(IdentityExternalIdentity, ext_identity_id)
        if (
            ext is None
            or ext.deleted_at is not None
            or ext.person_id != person_id
        ):
            raise NotFoundError("IdentityExternalIdentity", str(ext_identity_id))

        before = {
            "is_login_identity": ext.is_login_identity,
            "identity_status": ext.identity_status,
        }
        if is_login_identity is not None:
            ext.is_login_identity = is_login_identity
        if identity_status is not None:
            ext.identity_status = identity_status

        audit = AuditService(self.db)
        await audit.log(
            trace_id=trace_id,
            actor_user_id=actor_user_id,
            actor_type="user",
            action="identity.external_identity.update_login",
            resource_type="identity_external_identity",
            resource_id=str(ext_identity_id),
            status="success",
            detail={
                "person_id": str(person_id),
                "provider": ext.provider,
                "before": before,
                "after": {
                    "is_login_identity": ext.is_login_identity,
                    "identity_status": ext.identity_status,
                },
            },
            ip_address=ip_address,
            user_agent=user_agent,
        )
        await self.db.commit()
        await self.db.refresh(ext)
        return {
            "id": str(ext.id),
            "provider": ext.provider,
            "is_login_identity": ext.is_login_identity,
            "identity_status": ext.identity_status,
        }