"""Directory pull API — /api/v1/open/directory data endpoints.

Pins:

* keyset pagination: full drain across pages with no dup / no loss,
  including same-second ``updated_at`` ties (id tiebreaker)
* incremental: ``updated_since`` returns only later-updated rows;
  ``next_cursor`` round-trips
* population filter: out-of-filter persons are invisible in the list
  AND 404 on direct key lookup (not 403 — existence must not leak)
* leaver visibility: default filter hides ``left``; a filter whose
  ``employment_statuses`` includes ``left`` streams the transition
* PII tier: masked without ``directory.pii:read``, plaintext with it
* scope split: persons scope alone cannot read org endpoints
* memberships are restricted to in-filter persons
* addressing: /persons/{key} resolves UUID, employee_no and email
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta

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

from app.identity.consumers.models import IdentityConsumer
from app.identity.consumers.token_service import ConsumerTokenService
from app.identity.person_models import (
    IdentityOrgMembership,
    IdentityOrgNode,
    IdentityPerson,
)

pytestmark = pytest.mark.asyncio

PERSONS = "/api/v1/open/directory/persons"
ORG_NODES = "/api/v1/open/directory/org/nodes"
MEMBERSHIPS = "/api/v1/open/directory/org/memberships"


def _headers(token: str) -> dict:
    return {"Authorization": f"Bearer {token}"}


async def _mk_consumer(
    db: AsyncSession,
    *,
    scopes: list[str],
    filter_config: dict | None = None,
    granted_fields: list[str] | None = None,
) -> str:
    owner = IdentityPerson(
        id=uuid.uuid4(),
        work_email=f"owner-{uuid.uuid4().hex[:8]}@pingcap.com",
        display_name="Owner",
        employment_status="active",
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:10]}",
    )
    db.add(owner)
    await db.flush()
    from app.identity.consumers.field_catalog import (
        default_fields_for_scopes,
        validate_granted_fields,
    )
    fields = (
        validate_granted_fields(granted_fields, scopes)
        if granted_fields is not None
        else default_fields_for_scopes(scopes)
    )
    consumer = IdentityConsumer(
        id=uuid.uuid4(),
        name="Puller",
        slug=f"puller-{uuid.uuid4().hex[:6]}",
        owner_person_id=owner.id,
        scopes=scopes,
        granted_fields=fields,
        filter_config=filter_config or {},
    )
    db.add(consumer)
    await db.flush()
    await db.refresh(consumer)
    _row, plaintext = await ConsumerTokenService(db).mint(
        consumer=consumer, name="t", expires_in_days=None, created_by=None,
    )
    await db.commit()
    return plaintext


async def _mk_person(
    db: AsyncSession,
    *,
    email: str,
    status: str = "active",
    employee_no: str | None = None,
    dept: tuple[str, str] | None = None,
    mobile: str | None = None,
    personal_email: str | None = None,
    updated_at: datetime | None = None,
    github_id: str | None = None,
) -> uuid.UUID:
    pid = uuid.uuid4()
    db.add(IdentityPerson(
        id=pid,
        work_email=email,
        display_name=email.split("@")[0],
        employment_status=status,
        source_system="successfactors",
        hris_person_id=f"sf-{email}",
        employee_no=employee_no,
        mobile_phone=mobile,
        personal_email=personal_email,
        github_id=github_id,
    ))
    await db.flush()
    if dept is not None:
        source_id, name = dept
        db.add(IdentityOrgMembership(
            id=uuid.uuid4(),
            person_id=pid,
            org_node_source_id=source_id,
            org_node_name=name,
            org_path=f"PingCAP / {name}",
            is_primary=True,
        ))
        await db.flush()
    if updated_at is not None:
        # Bypass onupdate so the keyset position is deterministic.
        await db.execute(
            text("UPDATE identity_persons SET updated_at = :u WHERE id = :i"),
            {"u": updated_at.replace(tzinfo=None), "i": str(pid)},
        )
    return pid


async def _drain(client, token: str, *, page_size: int = 2, params: dict | None = None):
    """Walk the has_more chain to exhaustion; return all items."""
    items: list[dict] = []
    cursor = None
    for _ in range(50):  # hard stop against accidental infinite loop
        q = {"page_size": page_size, **(params or {})}
        if cursor:
            q["cursor"] = cursor
        r = await client.get(PERSONS, headers=_headers(token), params=q)
        assert r.status_code == 200, r.text
        data = r.json()["data"]
        items.extend(data["items"])
        cursor = data["next_cursor"]
        if not data["has_more"]:
            break
    return items


async def test_keyset_pagination_no_dup_no_loss(client, db_session):
    base = datetime(2026, 6, 1, 12, 0, 0, tzinfo=UTC)
    emails = []
    for i in range(5):
        # Two pairs share the same updated_at second → id tiebreaker path.
        ts = base + timedelta(seconds=i // 2)
        email = f"p{i}-{uuid.uuid4().hex[:6]}@pingcap.com"
        emails.append(email)
        await _mk_person(db_session, email=email, updated_at=ts)
    token = await _mk_consumer(db_session, scopes=["directory.persons:read"])

    items = await _drain(client, token, page_size=2)
    got = [i["work_email"] for i in items]
    assert sorted(got) == sorted([*emails, *[
        e for e in got if e not in emails
    ]])  # all created emails present, nothing duplicated
    assert len(got) == len(set(got))
    for e in emails:
        assert e in got


async def test_updated_since_incremental(client, db_session):
    old = datetime(2026, 5, 1, 0, 0, 0, tzinfo=UTC)
    new = datetime(2026, 6, 10, 0, 0, 0, tzinfo=UTC)
    await _mk_person(db_session, email="old@pingcap.com", updated_at=old)
    await _mk_person(db_session, email="new@pingcap.com", updated_at=new)
    token = await _mk_consumer(db_session, scopes=["directory.persons:read"])

    r = await client.get(PERSONS, headers=_headers(token), params={
        "updated_since": "2026-06-01T00:00:00",
    })
    assert r.status_code == 200
    emails = [i["work_email"] for i in r.json()["data"]["items"]]
    assert "new@pingcap.com" in emails
    assert "old@pingcap.com" not in emails

    # cursor + updated_since together is a 422
    r = await client.get(PERSONS, headers=_headers(token), params={
        "updated_since": "2026-06-01T00:00:00", "cursor": "abc",
    })
    assert r.status_code == 422
    # garbage cursor is a clean 422
    r = await client.get(PERSONS, headers=_headers(token), params={
        "cursor": "not-base64!!!",
    })
    assert r.status_code == 422


async def test_population_filter_list_and_404(client, db_session):
    await _mk_person(
        db_session, email="eng@pingcap.com",
        dept=("D100", "Engineering"), employee_no="P00001",
    )
    outsider = await _mk_person(
        db_session, email="sales@pingcap.com",
        dept=("D200", "Sales"), employee_no="P00002",
    )
    token = await _mk_consumer(
        db_session,
        scopes=["directory.persons:read"],
        filter_config={"departments": ["D100"], "include_subtree": False},
    )

    items = await _drain(client, token, page_size=50)
    emails = [i["work_email"] for i in items]
    assert "eng@pingcap.com" in emails
    assert "sales@pingcap.com" not in emails

    # direct lookup of out-of-filter person → 404, not 403
    for key in (str(outsider), "P00002", "sales@pingcap.com"):
        r = await client.get(f"{PERSONS}/{key}", headers=_headers(token))
        assert r.status_code == 404, key


async def test_leaver_visibility_via_filter(client, db_session):
    await _mk_person(db_session, email="leaver@pingcap.com", status="left")
    await _mk_person(db_session, email="alive@pingcap.com", status="active")

    default_token = await _mk_consumer(
        db_session, scopes=["directory.persons:read"],
    )
    items = await _drain(client, default_token, page_size=50)
    emails = [i["work_email"] for i in items]
    assert "alive@pingcap.com" in emails
    assert "leaver@pingcap.com" not in emails  # default filter = active only

    wide_token = await _mk_consumer(
        db_session,
        scopes=["directory.persons:read"],
        filter_config={"employment_statuses": ["active", "leaver_pending", "left"]},
    )
    items = await _drain(client, wide_token, page_size=50)
    by_email = {i["work_email"]: i for i in items}
    assert by_email["leaver@pingcap.com"]["employment_status"] == "left"

    # employment_status narrow-down param works within the filter
    r = await client.get(PERSONS, headers=_headers(wide_token), params={
        "employment_status": "left", "page_size": 50,
    })
    emails = [i["work_email"] for i in r.json()["data"]["items"]]
    assert emails and all(
        e == "leaver@pingcap.com" for e in emails if "@pingcap.com" in e
    )
    # invalid status value → 422
    r = await client.get(PERSONS, headers=_headers(wide_token), params={
        "employment_status": "offboarded,active",
    })
    assert r.status_code == 422


async def test_pii_omitted_vs_plaintext(client, db_session):
    """Field-level grant: PII is omitted entirely without the pii bundle
    (no masked middle ground), plaintext when its field is granted."""
    await _mk_person(
        db_session, email="pii@pingcap.com",
        mobile="+8613812345678", personal_email="someone@gmail.com",
    )
    no_pii_token = await _mk_consumer(
        db_session, scopes=["directory.persons:read"],
    )
    r = await client.get(
        f"{PERSONS}/pii@pingcap.com", headers=_headers(no_pii_token),
    )
    assert r.status_code == 200
    data = r.json()["data"]
    # Omitted, not masked — the key is simply absent.
    assert "mobile_phone" not in data
    assert "personal_email" not in data
    # Non-PII bundle fields are present.
    assert data["work_email"] == "pii@pingcap.com"

    pii_token = await _mk_consumer(
        db_session,
        scopes=["directory.persons:read", "directory.pii:read"],
    )
    r = await client.get(
        f"{PERSONS}/pii@pingcap.com", headers=_headers(pii_token),
    )
    data = r.json()["data"]
    assert data["mobile_phone"] == "+8613812345678"
    assert data["personal_email"] == "someone@gmail.com"


async def test_field_level_grant_narrows_projection(client, db_session):
    """A consumer granted a subset of person fields gets exactly that
    subset (plus the always-on id/updated_at), nothing else."""
    await _mk_person(
        db_session, email="narrow@pingcap.com", employee_no="P12321",
        mobile="+8613800000000", personal_email="x@gmail.com",
    )
    token = await _mk_consumer(
        db_session,
        scopes=["directory.persons:read", "directory.pii:read"],
        granted_fields=["work_email", "display_name", "mobile_phone"],
    )
    r = await client.get(
        f"{PERSONS}/narrow@pingcap.com", headers=_headers(token),
    )
    assert r.status_code == 200
    data = r.json()["data"]
    # Granted + always-on only.
    assert set(data.keys()) == {
        "id", "updated_at", "work_email", "display_name", "mobile_phone",
    }
    assert data["mobile_phone"] == "+8613800000000"
    # employee_no / personal_email were NOT granted → absent.
    assert "employee_no" not in data
    assert "personal_email" not in data


async def test_github_id_projection_and_grant_gate(client, db_session):
    """``github_id`` rides the persons bundle: present (value or explicit
    null) when granted, absent entirely when the grant is narrowed past
    it. The explicit-null case matters — a consumer reconciling GitHub org
    membership must be able to tell "we have no handle on file" apart from
    "this field isn't in my grant", and only a present-but-null key says
    the former."""
    await _mk_person(
        db_session, email="gh-has@pingcap.com", employee_no="P30001",
        github_id="octo-han",
    )
    await _mk_person(
        db_session, email="gh-none@pingcap.com", employee_no="P30002",
    )

    # Bundle default (no narrowing) → field is in the grant.
    token = await _mk_consumer(db_session, scopes=["directory.persons:read"])
    r = await client.get(
        f"{PERSONS}/gh-has@pingcap.com", headers=_headers(token),
    )
    assert r.status_code == 200, r.text
    assert r.json()["data"]["github_id"] == "octo-han"

    r = await client.get(
        f"{PERSONS}/gh-none@pingcap.com", headers=_headers(token),
    )
    data = r.json()["data"]
    assert "github_id" in data  # present…
    assert data["github_id"] is None  # …and explicitly null, not omitted

    # It also flows through the keyset list feed (that's the sync path).
    listed = {
        i["work_email"]: i.get("github_id")
        for i in await _drain(client, token, page_size=50)
    }
    assert listed["gh-has@pingcap.com"] == "octo-han"

    # Narrowed grant that excludes it → key absent entirely.
    narrow = await _mk_consumer(
        db_session, scopes=["directory.persons:read"],
        granted_fields=["work_email", "display_name"],
    )
    r = await client.get(
        f"{PERSONS}/gh-has@pingcap.com", headers=_headers(narrow),
    )
    assert r.status_code == 200
    assert "github_id" not in r.json()["data"]


async def test_catalog_and_serializer_stay_in_lockstep(db_session):
    """The field catalog IS the authorization surface for the person
    serializer: a key the serializer emits but the catalog doesn't list
    can never be granted (dead field, silently invisible), and a catalog
    key the serializer never emits is a grantable field that returns
    nothing (a lie in the admin UI + the /scopes catalog). The
    field_catalog docstring demands the two move together; this pins it
    so the next field addition can't half-land."""
    from app.identity.consumers.directory_service import serialize_person
    from app.identity.consumers.field_catalog import person_field_keys

    catalog_keys = set(person_field_keys())
    person = IdentityPerson(
        id=uuid.uuid4(),
        work_email="lockstep@pingcap.com",
        display_name="Lockstep",
        employment_status="active",
        source_system="successfactors",
        hris_person_id="sf-lockstep",
    )
    # granted_fields=every catalog key → the serializer's full key set.
    emitted = set(serialize_person(
        person, membership=None, manager=None, hrbp=None,
        granted_fields=catalog_keys,
    ))
    assert emitted == catalog_keys, (
        "field catalog and serialize_person drifted: "
        f"catalog-only={sorted(catalog_keys - emitted)}, "
        f"serializer-only={sorted(emitted - catalog_keys)}"
    )


async def test_person_addressing_by_uuid_empno_email(client, db_session):
    pid = await _mk_person(
        db_session, email="addr@pingcap.com", employee_no="P01572",
    )
    token = await _mk_consumer(db_session, scopes=["directory.persons:read"])
    for key in (str(pid), "P01572", "addr@pingcap.com"):
        r = await client.get(f"{PERSONS}/{key}", headers=_headers(token))
        assert r.status_code == 200, key
        assert r.json()["data"]["work_email"] == "addr@pingcap.com"


async def test_org_scope_split_and_nodes(client, db_session):
    db_session.add(IdentityOrgNode(
        id=uuid.uuid4(),
        org_node_source_id="D100",
        org_node_name="Engineering",
        org_path="PingCAP / Engineering",
    ))
    await db_session.commit()

    persons_only = await _mk_consumer(
        db_session, scopes=["directory.persons:read"],
    )
    r = await client.get(ORG_NODES, headers=_headers(persons_only))
    assert r.status_code == 403

    org_token = await _mk_consumer(db_session, scopes=["directory.org:read"])
    r = await client.get(ORG_NODES, headers=_headers(org_token))
    assert r.status_code == 200
    names = [n["name"] for n in r.json()["data"]["items"]]
    assert "Engineering" in names
    # org scope alone cannot read persons
    r = await client.get(PERSONS, headers=_headers(org_token))
    assert r.status_code == 403


async def test_memberships_respect_person_filter(client, db_session):
    await _mk_person(
        db_session, email="m-eng@pingcap.com", dept=("D100", "Engineering"),
    )
    await _mk_person(
        db_session, email="m-sales@pingcap.com", dept=("D200", "Sales"),
    )
    token = await _mk_consumer(
        db_session,
        scopes=["directory.persons:read", "directory.org:read"],
        filter_config={"departments": ["D100"], "include_subtree": False},
    )
    r = await client.get(MEMBERSHIPS, headers=_headers(token), params={
        "page_size": 50,
    })
    assert r.status_code == 200
    rows = r.json()["data"]["items"]
    emails = [m["work_email"] for m in rows]
    assert "m-eng@pingcap.com" in emails
    assert "m-sales@pingcap.com" not in emails


async def test_manager_ref_respects_population_filter(client, db_session):
    """An out-of-filter manager resolves to null — even identifier-only
    leakage (employee_no/work_email) crosses the population boundary
    (security review fix 2026-06-11). An in-filter manager resolves to
    the identifier pair."""
    mgr_out = await _mk_person(
        db_session, email="boss@pingcap.com", employee_no="P00100",
        dept=("D200", "Sales"),
    )
    mgr_in = await _mk_person(
        db_session, email="lead@pingcap.com", employee_no="P00101",
        dept=("D100", "Engineering"),
    )
    pid = await _mk_person(
        db_session, email="report@pingcap.com", dept=("D100", "Engineering"),
    )
    row = await db_session.get(IdentityPerson, pid)
    row.manager_person_id = mgr_out
    await db_session.commit()

    token = await _mk_consumer(
        db_session,
        scopes=["directory.persons:read"],
        filter_config={"departments": ["D100"], "include_subtree": False},
    )
    r = await client.get(
        f"{PERSONS}/report@pingcap.com", headers=_headers(token),
    )
    assert r.status_code == 200
    assert r.json()["data"]["manager"] is None

    row = await db_session.get(IdentityPerson, pid)
    row.manager_person_id = mgr_in
    await db_session.commit()
    r = await client.get(
        f"{PERSONS}/report@pingcap.com", headers=_headers(token),
    )
    assert r.json()["data"]["manager"] == {
        "employee_no": "P00101", "work_email": "lead@pingcap.com",
    }


async def test_drained_stream_keeps_high_water_cursor(client, db_session):
    """has_more=false still returns a resumable cursor: a row updated
    AFTER the drain must appear when polling the saved cursor."""
    await _mk_person(
        db_session, email="hw1@pingcap.com",
        updated_at=datetime(2026, 6, 1, 0, 0, 0, tzinfo=UTC),
    )
    token = await _mk_consumer(db_session, scopes=["directory.persons:read"])

    r = await client.get(PERSONS, headers=_headers(token), params={
        "page_size": 50,
    })
    data = r.json()["data"]
    assert data["has_more"] is False
    cursor = data["next_cursor"]
    assert cursor, "drained stream must still hand back the high-water cursor"

    # nothing new yet → empty page, same protocol
    r = await client.get(PERSONS, headers=_headers(token), params={
        "cursor": cursor, "page_size": 50,
    })
    data = r.json()["data"]
    assert data["items"] == []
    assert data["has_more"] is False

    # a later update reappears after the saved cursor. The cursor sits
    # at the consumer-owner row's NOW()-ish updated_at (the newest row
    # in the drain), so the "later" row must be stamped after that.
    await _mk_person(
        db_session, email="hw2@pingcap.com",
        updated_at=datetime.now(UTC) + timedelta(minutes=5),
    )
    await db_session.commit()
    r = await client.get(PERSONS, headers=_headers(token), params={
        "cursor": cursor, "page_size": 50,
    })
    emails = [i["work_email"] for i in r.json()["data"]["items"]]
    assert "hw2@pingcap.com" in emails
    assert "hw1@pingcap.com" not in emails


async def test_feed_safety_horizon_hides_fresh_rows(client, db_session):
    """With the lag enabled, a just-written row is held back; a row
    older than the lag is served."""
    from app.core.config import settings

    await _mk_person(
        db_session, email="old-row@pingcap.com",
        updated_at=datetime(2026, 6, 1, 0, 0, 0, tzinfo=UTC),
    )
    await _mk_person(db_session, email="fresh-row@pingcap.com")  # now()
    token = await _mk_consumer(db_session, scopes=["directory.persons:read"])

    settings.DIRECTORY_API_FEED_SAFETY_LAG_SECONDS = 30
    try:
        r = await client.get(PERSONS, headers=_headers(token), params={
            "page_size": 50,
        })
        emails = [i["work_email"] for i in r.json()["data"]["items"]]
        assert "old-row@pingcap.com" in emails
        assert "fresh-row@pingcap.com" not in emails
    finally:
        settings.DIRECTORY_API_FEED_SAFETY_LAG_SECONDS = 0


async def test_pii_read_is_audited(client, db_session):
    from app.audit.models import AuditLog

    await _mk_person(
        db_session, email="audited@pingcap.com", mobile="+8613812345678",
    )
    token = await _mk_consumer(
        db_session,
        scopes=["directory.persons:read", "directory.pii:read"],
    )
    r = await client.get(
        f"{PERSONS}/audited@pingcap.com", headers=_headers(token),
    )
    assert r.status_code == 200

    await db_session.commit()  # end REPEATABLE READ snapshot
    rows = (await db_session.execute(
        select(AuditLog).where(AuditLog.action == "directory_api.pii_read")
    )).scalars().all()
    assert rows, "plaintext PII reads must land in audit_logs"
    assert rows[-1].detail["persons_served"] == 1

async def test_cursor_is_opaque_and_signed(client, db_session):
    """The cursor must not be a client-decodable record of an
    out-of-filter row's id/timestamp, and a tampered cursor is a clean
    422 (security review 2026-06-11)."""
    import base64 as _b64

    # in-filter person + out-of-filter person with a LATER updated_at,
    # so a single-row page ends on the in-filter row but the scan has
    # to step over... actually we just assert the cursor is signed +
    # not plain-base64-decodable to a JSON position.
    await _mk_person(
        db_session, email="cur1@pingcap.com",
        updated_at=datetime(2026, 6, 1, 0, 0, 0, tzinfo=UTC),
    )
    await _mk_person(
        db_session, email="cur2@pingcap.com",
        updated_at=datetime(2026, 6, 2, 0, 0, 0, tzinfo=UTC),
    )
    token = await _mk_consumer(db_session, scopes=["directory.persons:read"])
    r = await client.get(PERSONS, headers=_headers(token), params={"page_size": 1})
    cursor = r.json()["data"]["next_cursor"]
    assert cursor and "." in cursor, "cursor must carry an HMAC signature"

    # the payload half must NOT be a naked decodable position the client
    # can read — tampering the signature is rejected
    payload_b64 = cursor.rsplit(".", 1)[0]
    tampered = f"{payload_b64}.{'0' * 32}"
    r = await client.get(PERSONS, headers=_headers(token), params={"cursor": tampered})
    assert r.status_code == 422

    # a totally forged base64 cursor is also 422, not 500
    forged = _b64.urlsafe_b64encode(b'{"u":"2026-01-01T00:00:00","i":"x"}').decode() + ".deadbeef"
    r = await client.get(PERSONS, headers=_headers(token), params={"cursor": forged})
    assert r.status_code == 422

    # the legitimate cursor still round-trips
    r = await client.get(PERSONS, headers=_headers(token), params={
        "cursor": cursor, "page_size": 1,
    })
    assert r.status_code == 200


async def test_directory_excludes_purged_nopush_noncorporate(client, db_session):
    """Egress-lifecycle gate parity with the push candidate loader
    (security review 2026-07-22).

    The Directory PULL surface must not leak persons the push side would
    never even consider: HRIS-purged rows (``hris_record_state='purged'``,
    whose stale ``employment_status`` may still read 'active'), SF no-push
    opt-outs (``sf_downstream_push_code='2061'``), and non-corporate-email
    junk records. Before this fix ``resolve_in_scope_person_ids`` applied
    only ``matches()``, which deliberately omits these data-quality gates.
    """
    await _mk_person(db_session, email="normal-egress@pingcap.com")
    purged = await _mk_person(db_session, email="purged-egress@pingcap.com")
    nopush = await _mk_person(db_session, email="nopush-egress@pingcap.com")
    await _mk_person(db_session, email="junk-egress@gmail.com")
    await db_session.execute(
        text(
            "UPDATE identity_persons SET hris_record_state='purged' "
            "WHERE id=:i"
        ),
        {"i": str(purged)},
    )
    await db_session.execute(
        text(
            "UPDATE identity_persons SET sf_downstream_push_code='2061' "
            "WHERE id=:i"
        ),
        {"i": str(nopush)},
    )
    await db_session.commit()

    token = await _mk_consumer(db_session, scopes=["directory.persons:read"])
    items = await _drain(client, token, page_size=50)
    emails = {i["work_email"] for i in items}

    assert "normal-egress@pingcap.com" in emails
    assert "purged-egress@pingcap.com" not in emails
    assert "nopush-egress@pingcap.com" not in emails
    assert "junk-egress@gmail.com" not in emails