"""Read-model helpers for the Directory pull API.

Two concerns:

* **Scope resolution** — evaluate a consumer's ``filter_config``
  (``app.core.person_filters``, the same engine the Identity Sync
  Center uses for push targets) over the whole person table and return
  the in-scope person-id set. ~500 employees ⇒ an in-memory pass per
  request is cheap and keeps pull/push semantics provably identical.
* **Serialization** — the wire shape of a person / org node /
  membership, with PII masked unless the consumer holds
  ``directory.pii:read``. Cross-person references (manager, HRBP) are
  emitted as identifiers only (employee_no + work_email), never
  embedded objects — a reference target outside the consumer's filter
  must not leak its profile through the back door.
"""
from __future__ import annotations

import base64
import binascii
import hashlib
import hmac
import json
from datetime import datetime
from typing import Any
from uuid import UUID

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

from app.core.config import settings
from app.core.exceptions import ValidationError
from app.core.person_filters import (
    CandidatePerson,
    FilterConfig,
    FilterError,
    expand_department_subtree,
    is_non_corporate_email,
    matches,
)

#: SF ``EmpJob.customString24`` value that means "不推送到下游系统" — the
#: HR/SF-owned per-person downstream-provisioning opt-out. Mirrors
#: ``app.domains.identity_sync.service._SF_NO_PUSH_CODE`` (source of
#: truth); duplicated here (not imported) to keep this egress surface
#: free of a cross-domain import. HR-confirmed 2026-07-07.
_SF_NO_PUSH_CODE = "2061"
from app.identity.person_models import (
    IdentityOrgMembership,
    IdentityOrgNode,
    IdentityPerson,
)

#: Server-side cap on ``page_size`` — bounds response size and the
#: per-request DB scan, mirroring the FE↔BE pagination contract rule.
MAX_PAGE_SIZE = 500
DEFAULT_PAGE_SIZE = 100


# ── Scope resolution ──────────────────────────────────────────────────


async def resolve_in_scope_person_ids(
    db: AsyncSession, filter_config: dict[str, Any],
) -> set[UUID]:
    """Evaluate the consumer's person filter and return matching ids.

    A malformed filter_config (admin saved an older shape, or a future
    field was removed) degrades to "match nobody" rather than "match
    everybody" — fail-closed is the only defensible posture on a data
    egress surface. Governance validation makes this near-impossible,
    but the data plane must not rely on the control plane's hygiene.
    """
    try:
        cfg = FilterConfig.from_dict(filter_config or {})
    except FilterError:
        return set()

    # HRIS-lifecycle egress gates — mirror the push candidate loader
    # (``identity_sync.service._load_candidates``) so the pull surface
    # (Directory / SCIM) and the push surface agree on exactly WHO may
    # egress. ``matches()`` is the shared per-consumer authorization
    # boundary, but it deliberately does NOT fold in these data-quality
    # exclusions (they would override an admin's configurable dimensions),
    # so — exactly like the push loader — they are applied HERE, before
    # ``matches()``:
    #   • ``hris_record_state='purged'`` — upstream HRIS dropped the row
    #     entirely (migration 222); a stale ``employment_status='active'``
    #     must NOT keep leaking their directory record to consumers, in
    #     parity with ``is_effectively_departed`` on the login surface.
    #   • ``sf_downstream_push_code='2061'`` (不推送) — HR/SF per-person
    #     opt-out from ALL downstream provisioning.
    #   • non-corporate ``work_email`` — HRIS junk / non-people that were
    #     never a real directory identity.
    # ``coalesce(...)`` treats NULL/unbackfilled as the safe non-excluded
    # value so a legacy row is never mistaken for purged / no-push.
    persons = (await db.execute(
        select(
            IdentityPerson.id,
            IdentityPerson.work_email,
            IdentityPerson.employment_status,
            IdentityPerson.country,
            IdentityPerson.job_title,
            IdentityPerson.employee_no,
            IdentityPerson.company_code,
            IdentityPerson.employee_type_code,
        ).where(
            IdentityPerson.deleted_at.is_(None),
            func.coalesce(
                IdentityPerson.hris_record_state, "present",
            ) != "purged",
            func.coalesce(
                IdentityPerson.sf_downstream_push_code, "",
            ) != _SF_NO_PUSH_CODE,
        )
    )).all()

    memberships = (await db.execute(
        select(
            IdentityOrgMembership.person_id,
            IdentityOrgMembership.org_path,
            IdentityOrgMembership.org_node_source_id,
        ).where(
            IdentityOrgMembership.is_primary.is_(True),
            IdentityOrgMembership.deleted_at.is_(None),
        )
    )).all()
    membership_by_person = {m.person_id: m for m in memberships}

    dept_ids: set[str] = set()
    if cfg.departments:
        nodes = (await db.execute(
            select(
                IdentityOrgNode.org_node_source_id,
                IdentityOrgNode.parent_org_node_source_id,
            ).where(IdentityOrgNode.deleted_at.is_(None))
        )).all()
        children_index: dict[str, list[str]] = {}
        for n in nodes:
            if n.parent_org_node_source_id:
                children_index.setdefault(
                    n.parent_org_node_source_id, [],
                ).append(n.org_node_source_id)
        dept_ids = expand_department_subtree(
            cfg.departments,
            children_index=children_index,
            include_subtree=cfg.include_subtree,
        )

    out: set[UUID] = set()
    for p in persons:
        if not p.work_email:
            continue
        # Non-corporate work_email exclusion (HRIS junk / non-people) —
        # applied in Python via the shared primitive, exactly like the
        # push loader, so the rule lives in one place.
        if is_non_corporate_email(p.work_email):
            continue
        m = membership_by_person.get(p.id)
        candidate = CandidatePerson(
            person_id=p.id,
            work_email=p.work_email,
            employment_status=p.employment_status,
            country=p.country,
            job_title=p.job_title,
            employee_no=p.employee_no,
            company_code=p.company_code,
            employee_type_code=p.employee_type_code,
            org_path=m.org_path if m else None,
            org_node_source_id=m.org_node_source_id if m else None,
        )
        if matches(candidate, cfg, department_source_ids=dept_ids):
            out.add(p.id)
    return out


# ── Keyset cursor ─────────────────────────────────────────────────────


def _cursor_sig(payload_b64: str) -> str:
    """HMAC-SHA256 tag over the cursor payload, keyed by JWT_SECRET_KEY.

    The cursor encodes the (updated_at, id) scan position — and the id
    at a page boundary can be an OUT-OF-FILTER row's UUID (the scan
    advances over filtered rows). Without a signature a consumer could
    base64-decode the cursor and read the existence + change-time of
    people outside its population filter — the same dimension
    ``/persons/{key}`` hides behind a 404. Signing makes the cursor a
    genuinely opaque server token: it round-trips for pagination but
    decodes to nothing for the client (security review 2026-06-11).
    """
    return hmac.new(
        settings.JWT_SECRET_KEY.encode("utf-8"),
        payload_b64.encode("ascii"),
        hashlib.sha256,
    ).hexdigest()[:32]


def encode_cursor(updated_at: datetime, row_id: UUID | str) -> str:
    payload = json.dumps(
        {"u": updated_at.isoformat(), "i": str(row_id)},
        separators=(",", ":"),
    )
    payload_b64 = base64.urlsafe_b64encode(payload.encode("utf-8")).decode("ascii")
    return f"{payload_b64}.{_cursor_sig(payload_b64)}"


def decode_cursor(raw: str) -> tuple[datetime, str]:
    """Parse + verify an opaque cursor; raises ValidationError on garbage
    or a bad signature.

    The cursor only drives pagination, but its payload may name an
    out-of-filter row id, so it is HMAC-signed (see ``_cursor_sig``) and
    treated as opaque. A tampered / unsigned value is a clean 422.
    """
    try:
        payload_b64, sig = raw.rsplit(".", 1)
        if not hmac.compare_digest(sig, _cursor_sig(payload_b64)):
            raise ValidationError("cursor is not valid")
        payload = json.loads(
            base64.urlsafe_b64decode(payload_b64.encode("ascii")).decode("utf-8"),
        )
        u = datetime.fromisoformat(payload["u"])
        i = str(payload["i"])
    except ValidationError:
        raise
    except (ValueError, KeyError, TypeError, binascii.Error) as e:
        raise ValidationError("cursor is not valid") from e
    return u, i


def horizon_clause(col):
    """``col <= NOW() - lag`` visibility horizon for the directory feeds.

    ``updated_at`` is stamped at statement time but a row only becomes
    visible at COMMIT — a consumer polling between the two can advance
    its cursor (REST) or its ``meta.lastModified`` watermark (SCIM) past
    the row's timestamp and never see it again. Holding back the
    freshest ``lag`` seconds closes that race on the DB's own clock.
    Shared by the REST keyset endpoints and the SCIM Users surface so
    both honour the same guarantee. ``lag=0`` disables it (tests).
    """
    lag = int(settings.DIRECTORY_API_FEED_SAFETY_LAG_SECONDS)
    if lag <= 0:
        return None
    return col <= func.date_sub(func.now(), text(f"INTERVAL {lag} SECOND"))


def naive_utc(dt: datetime) -> datetime:
    """Normalise to naive-UTC for comparison against MySQL DATETIME
    keyset columns (the driver hands rows back naive; Helix writes UTC)."""
    if dt.tzinfo is not None:
        from datetime import UTC

        return dt.astimezone(UTC).replace(tzinfo=None)
    return dt


# ── Serializers ───────────────────────────────────────────────────────


def _iso(dt: datetime | None) -> str | None:
    """RFC3339 with explicit UTC marker.

    asyncmy hands DATETIME back naive (UTC by Helix convention); a bare
    ``isoformat()`` shipped ``2026-06-11T03:00:00`` with no zone, and a
    consumer parsing that as LOCAL time in a western timezone computes
    an ``updated_since`` in the future → silently skipped incremental
    window. Always suffix ``Z`` — same as the SCIM surface (review fix
    2026-06-11).
    """
    return naive_utc(dt).isoformat() + "Z" if dt else None


def _person_ref(p: IdentityPerson | None) -> dict[str, Any] | None:
    """Identifier-only cross-person reference (no profile embedding)."""
    if p is None:
        return None
    return {"employee_no": p.employee_no, "work_email": p.work_email}


def serialize_person(
    p: IdentityPerson,
    *,
    membership: IdentityOrgMembership | None,
    manager: IdentityPerson | None,
    hrbp: IdentityPerson | None,
    granted_fields: set[str],
) -> dict[str, Any]:
    """Project a person to the consumer's granted field set.

    ``granted_fields`` is the least-privilege grant from
    ``field_catalog`` (see ``ConsumerContext.granted_fields``). Fields
    outside it — including PII fields the consumer wasn't granted — are
    *omitted entirely* rather than masked: on a field-level grant a
    field is either authorized (returned plaintext, and PII reads are
    audited upstream) or it is not present at all. ``id`` + ``updated_at``
    are always in the grant (catalog ``always`` flag) so the keyset feed
    stays intact.
    """
    department = None
    if membership is not None:
        department = {
            "source_id": membership.org_node_source_id,
            "name": membership.org_node_name,
            "path": membership.org_path,
        }
    full: dict[str, Any] = {
        "id": str(p.id),
        "employee_no": p.employee_no,
        "work_email": p.work_email,
        "display_name": p.display_name,
        "first_name": p.first_name,
        "last_name": p.last_name,
        "avatar_url": p.avatar_url,
        "employment_status": p.employment_status,
        "job_title": p.job_title,
        "is_china_employee": p.is_china_employee,
        "department": department,
        "manager": _person_ref(manager),
        "hrbp": _person_ref(hrbp),
        "work_location": {
            "code": p.work_location_code,
            "name": p.work_location_name,
            "country": p.country,
            "city": p.city,
        },
        "company": {"code": p.company_code, "name": p.company_name},
        "cost_center": p.cost_center,
        "hire_date": p.hire_date.isoformat() if p.hire_date else None,
        "termination_date": (
            p.termination_date.isoformat() if p.termination_date else None
        ),
        "preferred_language": p.preferred_language,
        "email_aliases": list(p.email_aliases or []),
        "github_id": p.github_id,
        "mobile_phone": p.mobile_phone,
        "personal_email": p.personal_email,
        "source_updated_at": _iso(p.source_updated_at),
        "updated_at": _iso(p.updated_at),
    }
    return {k: v for k, v in full.items() if k in granted_fields}


def serialize_org_node(n: IdentityOrgNode) -> dict[str, Any]:
    return {
        "source_id": n.org_node_source_id,
        "type": n.org_node_type,
        "code": n.org_node_code,
        "name": n.org_node_name,
        "parent_source_id": n.parent_org_node_source_id,
        "path": n.org_path,
        "level": n.org_level,
        "status": n.node_status,
        "updated_at": _iso(n.updated_at),
    }


def serialize_membership(
    m: IdentityOrgMembership, person: IdentityPerson | None,
) -> dict[str, Any]:
    return {
        "person_id": str(m.person_id),
        "employee_no": person.employee_no if person else None,
        "work_email": person.work_email if person else None,
        "org_node_source_id": m.org_node_source_id,
        "org_node_name": m.org_node_name,
        "is_primary": m.is_primary,
        "membership_status": m.membership_status,
        "updated_at": _iso(m.updated_at),
    }


__all__ = [
    "DEFAULT_PAGE_SIZE",
    "MAX_PAGE_SIZE",
    "decode_cursor",
    "encode_cursor",
    "horizon_clause",
    "naive_utc",
    "resolve_in_scope_person_ids",
    "serialize_membership",
    "serialize_org_node",
    "serialize_person",
]