"""Directory pull API — the REST data surface for identity consumers.

    GET /api/v1/open/directory/persons            directory.persons:read
    GET /api/v1/open/directory/persons/{key}      directory.persons:read
    GET /api/v1/open/directory/org/nodes          directory.org:read
    GET /api/v1/open/directory/org/memberships    directory.org:read

Design contract (documented for integrators in docs/integrations/
identity-pull-api.md and the /help manual):

* **Keyset incremental** — list endpoints order by ``(updated_at, id)``
  and return ``{items, next_cursor, has_more}``. ``next_cursor`` is the
  scan position to resume from — persist it after every call and pass
  it back verbatim; ``has_more=false`` means the stream is drained for
  now (keep polling the same cursor later — changed rows reappear after
  it, which IS the incremental feed). A bare timestamp comparison would
  lose rows updated in the same second by a bulk HRIS sync; the id
  tiebreaker makes the scan loss-free. Rows younger than a small
  safety lag (``DIRECTORY_API_FEED_SAFETY_LAG_SECONDS``, default 5s)
  are held back so an in-flight transaction committing late can't be
  skipped over by a cursor that already advanced past its timestamp.
* **No tombstone protocol** — persons are never hard-deleted. An
  offboarded person shows up in the stream with
  ``employment_status='left'`` *iff* the consumer's admin-configured
  filter includes leavers (``employment_statuses`` containing
  ``left`` / ``leaver_pending``). Default filters are active-only, so
  ask the Helix admin to widen the filter if you need leave events.
* **Population filter** — every response row is inside the consumer's
  admin-configured person filter. ``/persons/{key}`` answers 404 (not
  403) for an out-of-filter person so existence doesn't leak.
* **PII tier** — ``mobile_phone`` / ``personal_email`` come masked
  unless the consumer holds ``directory.pii:read``.
"""
from __future__ import annotations

import uuid as uuid_mod
from typing import Any

from fastapi import APIRouter, Depends, Query, Request
from sqlalchemy import select, tuple_
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.database import get_db
from app.core.exceptions import NotFoundError, ValidationError
from app.core.response import success_response
from app.identity.consumers.auth import (
    ConsumerContext,
    audit_pii_read,
    require_consumer,
)
from app.identity.consumers.directory_service import (
    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,
)
from app.identity.consumers.models import SCOPE_ORG_READ, SCOPE_PERSONS_READ
from app.identity.person_models import (
    IdentityOrgMembership,
    IdentityOrgNode,
    IdentityPerson,
)

router = APIRouter(
    prefix="/api/v1/open/directory",
    tags=["open-directory"],
)

#: Rows fetched per keyset scan round. Larger than the page cap so a
#: sparse filter (few matches in a dense id range) still fills a page
#: in one or two scans instead of N.
_SCAN_BATCH = 1000

_VALID_STATUSES = {"active", "leaver_pending", "left", "pre_hire"}


def _parse_pagination(
    cursor: str | None, updated_since: str | None, page_size: int,
):
    """Resolve the (after_updated_at, after_id) keyset start position."""
    if cursor and updated_since:
        raise ValidationError("Pass either cursor or updated_since, not both.")
    if cursor:
        u, i = decode_cursor(cursor)
        return naive_utc(u), i
    if updated_since:
        from datetime import datetime

        try:
            u = datetime.fromisoformat(updated_since)
        except ValueError as e:
            raise ValidationError(
                "updated_since must be an ISO-8601 datetime",
            ) from e
        # Start strictly *at* the timestamp with an empty id so rows
        # whose updated_at == updated_since are included (id > '').
        return naive_utc(u), ""
    return None, None


def _parse_status_param(raw: str | None) -> set[str] | None:
    if not raw:
        return None
    values = {s.strip() for s in raw.split(",") if s.strip()}
    unknown = values - _VALID_STATUSES
    if unknown:
        raise ValidationError(
            f"unknown employment_status value(s): {', '.join(sorted(unknown))}",
        )
    return values


async def _load_person_refs(
    db: AsyncSession,
    persons: list[IdentityPerson],
    *,
    scope_ids: set,
) -> tuple[dict, dict, dict]:
    """Batch-load primary memberships + manager/HRBP reference rows.

    References are restricted to the consumer's population filter: a
    manager/HRBP outside ``scope_ids`` resolves to ``None`` (the field
    is emitted as ``null``). Even identifier-only leakage (employee_no,
    work_email — and on SCIM the display name) of out-of-filter people
    crosses the boundary the filter exists to draw; a consumer that
    needs the full chain gets its filter widened by the admin instead
    (security review fix 2026-06-11).
    """
    ids = [p.id for p in persons]
    memberships: dict = {}
    if ids:
        rows = (await db.execute(
            select(IdentityOrgMembership).where(
                IdentityOrgMembership.person_id.in_(ids),
                IdentityOrgMembership.is_primary.is_(True),
                IdentityOrgMembership.deleted_at.is_(None),
            )
        )).scalars().all()
        memberships = {m.person_id: m for m in rows}

    ref_ids = {
        rid
        for p in persons
        for rid in (p.manager_person_id, p.hrbp_person_id)
        if rid is not None and rid in scope_ids
    }
    refs: dict = {}
    if ref_ids:
        rows = (await db.execute(
            select(IdentityPerson).where(IdentityPerson.id.in_(ref_ids))
        )).scalars().all()
        refs = {r.id: r for r in rows}

    managers = {p.id: refs.get(p.manager_person_id) for p in persons}
    hrbps = {p.id: refs.get(p.hrbp_person_id) for p in persons}
    return memberships, managers, hrbps


@router.get("/persons", summary="List persons (keyset incremental)")
async def list_persons(
    request: Request,
    cursor: str | None = Query(default=None),
    updated_since: str | None = Query(default=None),
    employment_status: str | None = Query(
        default=None,
        description="CSV narrow-down within the consumer's filter, e.g. 'active,left'",
    ),
    page_size: int = Query(default=DEFAULT_PAGE_SIZE, ge=1, le=MAX_PAGE_SIZE),
    db: AsyncSession = Depends(get_db),
    ctx: ConsumerContext = Depends(require_consumer(SCOPE_PERSONS_READ)),
):
    after_u, after_i = _parse_pagination(cursor, updated_since, page_size)
    status_narrow = _parse_status_param(employment_status)
    scope_ids = await resolve_in_scope_person_ids(db, ctx.filter_config)

    out_rows: list[IdentityPerson] = []
    scan_u, scan_i = after_u, after_i
    exhausted = False
    horizon = horizon_clause(IdentityPerson.updated_at)
    while len(out_rows) < page_size:
        stmt = (
            select(IdentityPerson)
            .where(IdentityPerson.deleted_at.is_(None))
            .order_by(IdentityPerson.updated_at.asc(), IdentityPerson.id.asc())
            .limit(_SCAN_BATCH)
        )
        if horizon is not None:
            stmt = stmt.where(horizon)
        if scan_u is not None:
            # Keyset: strictly after the (updated_at, id) position.
            # ``id`` compares as its CHAR(36) string form on MySQL/TiDB.
            stmt = stmt.where(
                tuple_(IdentityPerson.updated_at, IdentityPerson.id)
                > tuple_(scan_u, scan_i)
            )
        batch = list((await db.execute(stmt)).scalars().all())
        if not batch:
            exhausted = True
            break
        for p in batch:
            scan_u, scan_i = p.updated_at, str(p.id)
            if p.id not in scope_ids:
                continue
            if status_narrow and p.employment_status not in status_narrow:
                continue
            out_rows.append(p)
            if len(out_rows) >= page_size:
                break
        if len(batch) < _SCAN_BATCH and len(out_rows) < page_size:
            exhausted = True
            break

    # next_cursor = the furthest scan position. Persist it after every
    # call; when has_more=false, polling the SAME cursor later is the
    # incremental feed (rows whose updated_at advances reappear after
    # it). With no input cursor and an empty table it stays null.
    next_cursor = (
        encode_cursor(scan_u, scan_i) if scan_u is not None else None
    )
    has_more = not exhausted

    memberships, managers, hrbps = await _load_person_refs(
        db, out_rows, scope_ids=scope_ids,
    )
    granted = set(ctx.granted_fields)
    items = [
        serialize_person(
            p,
            membership=memberships.get(p.id),
            manager=managers.get(p.id),
            hrbp=hrbps.get(p.id),
            granted_fields=granted,
        )
        for p in out_rows
    ]
    if ctx.reads_pii():
        await audit_pii_read(
            request=request, ctx=ctx, persons_served=len(out_rows),
        )
    return success_response(
        data={"items": items, "next_cursor": next_cursor, "has_more": has_more},
        request=request,
    )


@router.get("/persons/{key}", summary="Get one person by id / employee_no / email")
async def get_person(
    key: str,
    request: Request,
    db: AsyncSession = Depends(get_db),
    ctx: ConsumerContext = Depends(require_consumer(SCOPE_PERSONS_READ)),
):
    conds: list[Any] = []
    try:
        conds.append(IdentityPerson.id == str(uuid_mod.UUID(key)))
    except ValueError:
        lowered = key.strip()
        if "@" in lowered:
            conds.append(IdentityPerson.work_email == lowered.lower())
        else:
            conds.append(IdentityPerson.employee_no == lowered)

    # Scope is resolved BEFORE the row lookup so the miss path and the
    # out-of-filter path do the same work — otherwise the response-time
    # difference is a (weak) oracle for "this email/employee_no is a
    # real employee outside my filter" (security review 2026-06-11).
    scope_ids = await resolve_in_scope_person_ids(db, ctx.filter_config)
    row = (await db.execute(
        select(IdentityPerson).where(
            conds[0], IdentityPerson.deleted_at.is_(None),
        )
    )).scalars().first()

    # Out-of-filter == not found: a 403 here would leak that the key
    # resolves to a real employee outside the consumer's population.
    if row is not None and row.id not in scope_ids:
        row = None
    if row is None:
        raise NotFoundError("Person", key)

    memberships, managers, hrbps = await _load_person_refs(
        db, [row], scope_ids=scope_ids,
    )
    if ctx.reads_pii():
        await audit_pii_read(request=request, ctx=ctx, persons_served=1)
    return success_response(
        data=serialize_person(
            row,
            membership=memberships.get(row.id),
            manager=managers.get(row.id),
            hrbp=hrbps.get(row.id),
            granted_fields=set(ctx.granted_fields),
        ),
        request=request,
    )


@router.get("/org/nodes", summary="List org nodes (keyset incremental)")
async def list_org_nodes(
    request: Request,
    cursor: str | None = Query(default=None),
    updated_since: str | None = Query(default=None),
    page_size: int = Query(default=DEFAULT_PAGE_SIZE, ge=1, le=MAX_PAGE_SIZE),
    db: AsyncSession = Depends(get_db),
    ctx: ConsumerContext = Depends(require_consumer(SCOPE_ORG_READ)),
):
    after_u, after_i = _parse_pagination(cursor, updated_since, page_size)
    stmt = (
        select(IdentityOrgNode)
        .where(IdentityOrgNode.deleted_at.is_(None))
        .order_by(IdentityOrgNode.updated_at.asc(), IdentityOrgNode.id.asc())
        .limit(page_size)
    )
    horizon = horizon_clause(IdentityOrgNode.updated_at)
    if horizon is not None:
        stmt = stmt.where(horizon)
    if after_u is not None:
        stmt = stmt.where(
            tuple_(IdentityOrgNode.updated_at, IdentityOrgNode.id)
            > tuple_(after_u, after_i)
        )
    rows = list((await db.execute(stmt)).scalars().all())
    if rows:
        next_cursor = encode_cursor(rows[-1].updated_at, rows[-1].id)
    elif after_u is not None:
        next_cursor = encode_cursor(after_u, after_i)
    else:
        next_cursor = None
    has_more = len(rows) == page_size
    return success_response(
        data={
            "items": [serialize_org_node(n) for n in rows],
            "next_cursor": next_cursor,
            "has_more": has_more,
        },
        request=request,
    )


@router.get("/org/memberships", summary="List person-org memberships (keyset incremental)")
async def list_org_memberships(
    request: Request,
    cursor: str | None = Query(default=None),
    updated_since: str | None = Query(default=None),
    page_size: int = Query(default=DEFAULT_PAGE_SIZE, ge=1, le=MAX_PAGE_SIZE),
    db: AsyncSession = Depends(get_db),
    ctx: ConsumerContext = Depends(require_consumer(SCOPE_ORG_READ)),
):
    """Memberships are person-linked data, so rows are restricted to the
    consumer's person filter — org:read alone must not become a side
    channel for enumerating people the filter excludes."""
    after_u, after_i = _parse_pagination(cursor, updated_since, page_size)
    scope_ids = await resolve_in_scope_person_ids(db, ctx.filter_config)

    out_rows: list[IdentityOrgMembership] = []
    scan_u, scan_i = after_u, after_i
    exhausted = False
    horizon = horizon_clause(IdentityOrgMembership.updated_at)
    while len(out_rows) < page_size:
        stmt = (
            select(IdentityOrgMembership)
            .where(IdentityOrgMembership.deleted_at.is_(None))
            .order_by(
                IdentityOrgMembership.updated_at.asc(),
                IdentityOrgMembership.id.asc(),
            )
            .limit(_SCAN_BATCH)
        )
        if horizon is not None:
            stmt = stmt.where(horizon)
        if scan_u is not None:
            stmt = stmt.where(
                tuple_(IdentityOrgMembership.updated_at, IdentityOrgMembership.id)
                > tuple_(scan_u, scan_i)
            )
        batch = list((await db.execute(stmt)).scalars().all())
        if not batch:
            exhausted = True
            break
        for m in batch:
            scan_u, scan_i = m.updated_at, str(m.id)
            if m.person_id not in scope_ids:
                continue
            out_rows.append(m)
            if len(out_rows) >= page_size:
                break
        if len(batch) < _SCAN_BATCH and len(out_rows) < page_size:
            exhausted = True
            break

    next_cursor = (
        encode_cursor(scan_u, scan_i) if scan_u is not None else None
    )
    has_more = not exhausted

    person_ids = {m.person_id for m in out_rows}
    persons: dict = {}
    if person_ids:
        rows = (await db.execute(
            select(IdentityPerson).where(IdentityPerson.id.in_(person_ids))
        )).scalars().all()
        persons = {p.id: p for p in rows}

    return success_response(
        data={
            "items": [
                serialize_membership(m, persons.get(m.person_id))
                for m in out_rows
            ],
            "next_cursor": next_cursor,
            "has_more": has_more,
        },
        request=request,
    )


__all__ = ["router"]