"""Read-only SCIM 2.0 server (RFC 7643/7644) for identity consumers.

    GET /scim/v2/ServiceProviderConfig      any valid service token
    GET /scim/v2/ResourceTypes[/{id}]       any valid service token
    GET /scim/v2/Schemas[/{id}]             any valid service token
    GET /scim/v2/Users[/{id}]               directory.persons:read
    GET /scim/v2/Groups[/{id}]              directory.org:read
    POST/PUT/PATCH/DELETE anywhere          405 (Helix is the source of
                                            truth; SCIM writes are not
                                            accepted on this surface)

Protocol notes (ADR 2026-06-11 — envelope exemption):

* Response bodies are RFC 7644 SCIM JSON (``application/scim+json``),
  NOT the Helix ``{success,data,meta,errors}`` envelope — the protocol
  prescribes the shape and standard SCIM clients parse it literally.
  Endpoint-level errors (400/404/405) use the SCIM Error schema; auth
  failures (401/403/429) bubble through the platform handler — SCIM
  clients act on the status code there, and keeping the shared auth
  dependency untouched means every defence (rate limits, audit) stays
  identical across the REST and SCIM surfaces.
* Filter support is deliberately minimal and honestly advertised:
  ``userName eq``, ``externalId eq``, ``meta.lastModified gt/ge`` for
  Users (single clause or two ``and``-joined), ``displayName eq`` /
  ``externalId eq`` for Groups. Anything else → 400 invalidFilter.
  INCREMENTAL-SYNC CAVEAT: SCIM paginates by ``startIndex`` and orders
  by ``(created_at, id)``, so ``meta.lastModified`` is a coarse
  timestamp filter, NOT a loss-free cursor — a watermark-and-``gt``
  poll can step over a row whose ``updated_at`` shares the watermark
  second (a bulk HRIS sync writes hundreds in one second). Consumers
  needing loss-free incremental sync MUST either use the REST
  ``/persons`` keyset feed (which resumes on the full ``(updated_at,id)``
  tuple), or poll with ``ge`` and rely on idempotent upsert on their
  side (re-fetch the boundary second, never skip it). The horizon clamp
  below only narrows the freshest-edge race, it does not fix
  same-second boundary loss.
* The consumer's person filter applies everywhere: out-of-filter Users
  404, Group ``members`` only list in-filter persons.
* ``phoneNumbers`` is emitted ONLY when the consumer holds
  ``directory.pii:read`` — a masked phone string inside a SCIM
  attribute breaks downstream parsers, so the attribute is omitted
  entirely rather than masked.
"""
from __future__ import annotations

import re
import uuid as uuid_mod
from datetime import datetime
from typing import Any

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

from app.core.config import settings
from app.core.database import get_db
from app.identity.consumers.auth import (
    ConsumerContext,
    audit_pii_read,
    require_consumer,
)
from app.identity.consumers.directory_service import (
    horizon_clause,
    naive_utc,
    resolve_in_scope_person_ids,
)
from app.identity.consumers.models import (
    SCOPE_ORG_READ,
    SCOPE_PERSONS_READ,
)
from app.identity.person_models import (
    IdentityOrgMembership,
    IdentityOrgNode,
    IdentityPerson,
)

router = APIRouter(prefix="/scim/v2", tags=["scim"])

SCIM_CONTENT_TYPE = "application/scim+json"
_LIST_SCHEMA = "urn:ietf:params:scim:api:messages:2.0:ListResponse"
_ERROR_SCHEMA = "urn:ietf:params:scim:api:messages:2.0:Error"
_USER_SCHEMA = "urn:ietf:params:scim:schemas:core:2.0:User"
_ENTERPRISE_SCHEMA = "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User"
_GROUP_SCHEMA = "urn:ietf:params:scim:schemas:core:2.0:Group"

_MAX_COUNT = 500
_DEFAULT_COUNT = 100


def _scim_json(payload: dict, status: int = 200) -> JSONResponse:
    return JSONResponse(payload, status_code=status, media_type=SCIM_CONTENT_TYPE)


def _scim_error(status: int, detail: str, scim_type: str | None = None) -> JSONResponse:
    body: dict[str, Any] = {
        "schemas": [_ERROR_SCHEMA],
        "status": str(status),
        "detail": detail,
    }
    if scim_type:
        body["scimType"] = scim_type
    return _scim_json(body, status)


def _iso(dt: datetime | None) -> str | None:
    if dt is None:
        return None
    # SCIM wants RFC3339 with timezone; Helix stores UTC.
    return naive_utc(dt).isoformat() + "Z"


# ── Filter parsing ────────────────────────────────────────────────────

#: One comparison clause: ``attr op "value"``. The value group is
#: quote-delimited, so it may legitimately contain `` and `` — which is
#: exactly why the parser below matches whole clauses instead of
#: splitting the raw string on the bare ``and`` token (a department
#: named "Research and Development" must parse, review fix 2026-06-11).
_CLAUSE_PATTERN = r'(?P<attr{n}>[A-Za-z][A-Za-z0-9_.]*)\s+(?P<op{n}>eq|gt|ge)\s+"(?P<value{n}>[^"]*)"'
_SINGLE_RE = re.compile(r"^\s*" + _CLAUSE_PATTERN.format(n=1) + r"\s*$", re.IGNORECASE)
_DOUBLE_RE = re.compile(
    r"^\s*" + _CLAUSE_PATTERN.format(n=1)
    + r"\s+and\s+" + _CLAUSE_PATTERN.format(n=2) + r"\s*$",
    re.IGNORECASE,
)


def parse_scim_filter(
    raw: str | None, *, allowed: dict[str, set[str]],
) -> list[tuple[str, str, str]] | JSONResponse:
    """Parse the supported SCIM filter subset.

    ``allowed`` maps lowercase attribute path → allowed operators.
    Returns a list of (attr, op, value) clauses, or a 400 SCIM error
    response for anything outside the advertised subset. Quote-aware:
    the whole filter is matched as one or two complete clauses, so a
    quoted value containing `` and `` parses correctly.
    """
    if not raw:
        return []
    m = _SINGLE_RE.match(raw)
    groups: list[tuple[str, str, str]] = []
    if m:
        groups = [(m.group("attr1"), m.group("op1"), m.group("value1"))]
    else:
        m = _DOUBLE_RE.match(raw)
        if m:
            groups = [
                (m.group("attr1"), m.group("op1"), m.group("value1")),
                (m.group("attr2"), m.group("op2"), m.group("value2")),
            ]
    if not groups:
        return _scim_error(
            400,
            'Unsupported filter syntax. Supported: <attr> eq|gt|ge "value", '
            "optionally two clauses joined by 'and'.",
            scim_type="invalidFilter",
        )
    clauses: list[tuple[str, str, str]] = []
    for raw_attr, raw_op, value in groups:
        attr = raw_attr.lower()
        op = raw_op.lower()
        if attr not in allowed or op not in allowed[attr]:
            return _scim_error(
                400, f"Filtering on '{raw_attr}' with '{op}' is not supported.",
                scim_type="invalidFilter",
            )
        clauses.append((attr, op, value))
    return clauses


def _parse_dt(value: str) -> datetime | None:
    try:
        return naive_utc(datetime.fromisoformat(value.replace("Z", "+00:00")))
    except ValueError:
        return None


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


def _user_resource(
    p: IdentityPerson,
    *,
    membership: IdentityOrgMembership | None,
    manager: IdentityPerson | None,
    granted_fields: frozenset[str] | set[str],
    base_url: str,
) -> dict[str, Any]:
    # Project against the consumer's field-level grant, mirroring the REST
    # serializer (``directory_service._serialize_person`` → ``if k in
    # granted_fields``). SECURITY: previously SCIM emitted every attribute
    # unconditionally (only phoneNumbers was gated), so a consumer whose
    # grant was narrowed below its bundle (e.g. display_name + work_email
    # only, excluding employee_no / cost_center / department / manager /
    # location) could read all the withheld internal-tier fields simply by
    # switching from the REST endpoint to /scim/v2/Users. ``id`` +
    # ``updated_at`` are catalog ``always`` fields, so ``meta`` + ``id``
    # stay unconditional (they anchor the keyset incremental pull).
    gf = granted_fields
    enterprise: dict[str, Any] = {}
    if "employee_no" in gf:
        enterprise["employeeNumber"] = p.employee_no
    if "cost_center" in gf:
        enterprise["costCenter"] = p.cost_center
    if "company" in gf:
        enterprise["organization"] = p.company_name
    if "department" in gf:
        enterprise["department"] = membership.org_node_name if membership else None
    if "manager" in gf and manager is not None:
        enterprise["manager"] = {
            "value": str(manager.id),
            "displayName": manager.display_name,
        }
    resource: dict[str, Any] = {
        "schemas": [_USER_SCHEMA, _ENTERPRISE_SCHEMA],
        "id": str(p.id),
        "meta": {
            "resourceType": "User",
            "created": _iso(p.created_at),
            "lastModified": _iso(p.updated_at),
            "location": f"{base_url}/scim/v2/Users/{p.id}",
        },
    }
    if "employee_no" in gf:
        resource["externalId"] = p.employee_no
    if "work_email" in gf:
        resource["userName"] = p.work_email
        resource["emails"] = [
            {"value": p.work_email, "type": "work", "primary": True},
        ]
    if "display_name" in gf:
        resource["displayName"] = p.display_name
    name: dict[str, Any] = {}
    if "first_name" in gf:
        name["givenName"] = p.first_name
    if "last_name" in gf:
        name["familyName"] = p.last_name
    if name:
        resource["name"] = name
    if "employment_status" in gf:
        resource["active"] = p.employment_status == "active"
    if "job_title" in gf:
        resource["title"] = p.job_title
    if "preferred_language" in gf:
        resource["preferredLanguage"] = p.preferred_language
    if "work_location" in gf:
        resource["addresses"] = [{
            "type": "work",
            "locality": p.city,
            "country": p.country,
        }]
    if enterprise:
        resource[_ENTERPRISE_SCHEMA] = enterprise
    if "mobile_phone" in gf and p.mobile_phone:
        resource["phoneNumbers"] = [{"value": p.mobile_phone, "type": "mobile"}]
    return resource


def _group_resource(
    n: IdentityOrgNode,
    *,
    members: list[IdentityPerson],
    base_url: str,
) -> dict[str, Any]:
    return {
        "schemas": [_GROUP_SCHEMA],
        "id": str(n.id),
        "externalId": n.org_node_source_id,
        "displayName": n.org_node_name,
        "members": [
            {
                "value": str(m.id),
                "display": m.display_name,
                "$ref": f"{base_url}/scim/v2/Users/{m.id}",
            }
            for m in members
        ],
        "meta": {
            "resourceType": "Group",
            "created": _iso(n.created_at),
            "lastModified": _iso(n.updated_at),
            "location": f"{base_url}/scim/v2/Groups/{n.id}",
        },
    }


def _list_response(
    resources: list[dict], *, total: int, start_index: int, count: int,
) -> dict[str, Any]:
    return {
        "schemas": [_LIST_SCHEMA],
        "totalResults": total,
        "startIndex": start_index,
        "itemsPerPage": len(resources),
        "Resources": resources,
    }


def _base_url(request: Request) -> str:
    """Absolute base URL for SCIM ``meta.location`` / ``$ref``.

    Derived from the configured public origin (first CORS_ORIGINS
    entry), NOT the request ``Host`` header. nginx forwards Host
    verbatim, so trusting it would let a client with a valid token
    poison its own response URLs (e.g. ``https://evil/scim/...``) and,
    if its SCIM client follows ``$ref`` with the token attached, leak
    the credential to an attacker-controlled host. SCIM requires
    absolute URLs, so a fixed deployment origin is the correct source.
    Falls back to the request origin only when no public origin is
    configured (local dev / tests). (Security review 2026-06-11.)
    """
    origins = settings.CORS_ORIGINS or []
    if origins:
        return str(origins[0]).rstrip("/")
    proto = request.url.scheme
    host = request.url.netloc
    return f"{proto}://{host}"


def _page_params(start_index: int, count: int) -> tuple[int, int]:
    # RFC 7644 §3.4.2.4: startIndex < 1 → 1; count < 0 → 0.
    return max(start_index, 1), min(max(count, 0), _MAX_COUNT)


# ── Discovery endpoints ───────────────────────────────────────────────


@router.get("/ServiceProviderConfig")
async def service_provider_config(
    request: Request,
    _ctx: ConsumerContext = Depends(require_consumer()),
):
    return _scim_json({
        "schemas": ["urn:ietf:params:scim:schemas:core:2.0:ServiceProviderConfig"],
        # In-product API docs (People / org directory tab) — discoverable
        # by any signed-in colleague, unlike a repo-internal markdown file.
        "documentationUri": f"{str(request.base_url).rstrip('/')}/api-docs",
        "patch": {"supported": False},
        "bulk": {"supported": False, "maxOperations": 0, "maxPayloadSize": 0},
        # Honest subset advertisement — see module docstring.
        "filter": {"supported": True, "maxResults": _MAX_COUNT},
        "changePassword": {"supported": False},
        "sort": {"supported": False},
        "etag": {"supported": False},
        "authenticationSchemes": [{
            "type": "oauthbearertoken",
            "name": "Helix service token",
            "description": (
                "Authorization: Bearer helix_svc_… issued by a Helix "
                "global admin on the Identity Consumers console."
            ),
        }],
        "meta": {
            "resourceType": "ServiceProviderConfig",
            "location": f"{_base_url(request)}/scim/v2/ServiceProviderConfig",
        },
    })


_RESOURCE_TYPES = [
    {
        "schemas": ["urn:ietf:params:scim:schemas:core:2.0:ResourceType"],
        "id": "User",
        "name": "User",
        "endpoint": "/Users",
        "schema": _USER_SCHEMA,
        "schemaExtensions": [
            {"schema": _ENTERPRISE_SCHEMA, "required": False},
        ],
    },
    {
        "schemas": ["urn:ietf:params:scim:schemas:core:2.0:ResourceType"],
        "id": "Group",
        "name": "Group",
        "endpoint": "/Groups",
        "schema": _GROUP_SCHEMA,
    },
]


@router.get("/ResourceTypes")
async def resource_types(
    _ctx: ConsumerContext = Depends(require_consumer()),
):
    return _scim_json(_list_response(
        list(_RESOURCE_TYPES), total=len(_RESOURCE_TYPES), start_index=1,
        count=len(_RESOURCE_TYPES),
    ))


@router.get("/ResourceTypes/{type_id}")
async def resource_type(
    type_id: str,
    _ctx: ConsumerContext = Depends(require_consumer()),
):
    for rt in _RESOURCE_TYPES:
        if rt["id"] == type_id:
            return _scim_json(dict(rt))
    return _scim_error(404, f"ResourceType {type_id} not found.")


_SCHEMAS = [
    {"id": _USER_SCHEMA, "name": "User",
     "description": "Helix person golden record projected as a SCIM User."},
    {"id": _ENTERPRISE_SCHEMA, "name": "EnterpriseUser",
     "description": "employeeNumber / department / costCenter / organization / manager."},
    {"id": _GROUP_SCHEMA, "name": "Group",
     "description": "Helix org node with in-filter members."},
]


@router.get("/Schemas")
async def schemas(
    _ctx: ConsumerContext = Depends(require_consumer()),
):
    return _scim_json(_list_response(
        [dict(s) for s in _SCHEMAS], total=len(_SCHEMAS), start_index=1,
        count=len(_SCHEMAS),
    ))


@router.get("/Schemas/{schema_id}")
async def schema(
    schema_id: str,
    _ctx: ConsumerContext = Depends(require_consumer()),
):
    for s in _SCHEMAS:
        if s["id"] == schema_id:
            return _scim_json(dict(s))
    return _scim_error(404, f"Schema {schema_id} not found.")


# ── Users ─────────────────────────────────────────────────────────────


async def _load_scoped_persons_from_ids(
    db: AsyncSession, scope_ids: set,
) -> list[IdentityPerson]:
    if not scope_ids:
        return []
    stmt = (
        select(IdentityPerson)
        .where(
            IdentityPerson.id.in_(scope_ids),
            IdentityPerson.deleted_at.is_(None),
        )
        # Stable order for startIndex paging.
        .order_by(IdentityPerson.created_at.asc(), IdentityPerson.id.asc())
    )
    # Same visibility horizon as the REST keyset feed: hold back the
    # freshest rows so a ``meta.lastModified gt`` incremental poll can't
    # step over a row whose updated_at is stamped but whose transaction
    # hasn't committed yet (logic review round 2, 2026-06-11).
    horizon = horizon_clause(IdentityPerson.updated_at)
    if horizon is not None:
        stmt = stmt.where(horizon)
    rows = (await db.execute(stmt)).scalars().all()
    return list(rows)


async def _user_context(
    db: AsyncSession, persons: list[IdentityPerson], *, scope_ids: set,
) -> tuple[dict, dict]:
    """Memberships + manager rows for serialization.

    The manager reference is restricted to the consumer's population
    filter: a SCIM manager block carries the manager's displayName
    (full name) — emitting it for an out-of-filter person walks the
    org chain straight across the filter boundary, so the block is
    omitted 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}
    mgr_ids = {
        p.manager_person_id for p in persons
        if p.manager_person_id and p.manager_person_id in scope_ids
    }
    managers: dict = {}
    if mgr_ids:
        rows = (await db.execute(
            select(IdentityPerson).where(IdentityPerson.id.in_(mgr_ids))
        )).scalars().all()
        by_id = {r.id: r for r in rows}
        managers = {
            p.id: by_id.get(p.manager_person_id) for p in persons
        }
    return memberships, managers


def _apply_user_clauses(
    persons: list[IdentityPerson], clauses: list[tuple[str, str, str]],
) -> list[IdentityPerson] | JSONResponse:
    out = persons
    for attr, op, value in clauses:
        if attr == "username":
            out = [p for p in out if (p.work_email or "").lower() == value.lower()]
        elif attr == "externalid":
            out = [p for p in out if (p.employee_no or "") == value]
        elif attr == "meta.lastmodified":
            cut = _parse_dt(value)
            if cut is None:
                return _scim_error(
                    400, "meta.lastModified value must be an ISO-8601 datetime.",
                    scim_type="invalidFilter",
                )
            if op == "gt":
                out = [p for p in out if p.updated_at and naive_utc(p.updated_at) > cut]
            else:  # ge
                out = [p for p in out if p.updated_at and naive_utc(p.updated_at) >= cut]
    return out


@router.get("/Users")
async def list_users(
    request: Request,
    filter: str | None = Query(default=None),  # noqa: A002 — SCIM param name
    startIndex: int = Query(default=1),  # noqa: N803 — SCIM param name
    count: int = Query(default=_DEFAULT_COUNT),
    db: AsyncSession = Depends(get_db),
    ctx: ConsumerContext = Depends(require_consumer(SCOPE_PERSONS_READ)),
):
    clauses = parse_scim_filter(filter, allowed={
        "username": {"eq"},
        "externalid": {"eq"},
        "meta.lastmodified": {"gt", "ge"},
    })
    if isinstance(clauses, JSONResponse):
        return clauses

    scope_ids = await resolve_in_scope_person_ids(db, ctx.filter_config)
    persons = await _load_scoped_persons_from_ids(db, scope_ids)
    filtered = _apply_user_clauses(persons, clauses)
    if isinstance(filtered, JSONResponse):
        return filtered

    start, page = _page_params(startIndex, count)
    window = filtered[start - 1 : start - 1 + page]
    memberships, managers = await _user_context(db, window, scope_ids=scope_ids)
    base = _base_url(request)
    # SCIM's only PII attribute is phoneNumbers — gate it on the granted
    # mobile_phone field (field-level grant), not merely the pii scope.
    include_phone = "mobile_phone" in ctx.granted_fields
    resources = [
        _user_resource(
            p,
            membership=memberships.get(p.id),
            manager=managers.get(p.id),
            granted_fields=ctx.granted_fields,
            base_url=base,
        )
        for p in window
    ]
    if include_phone:
        await audit_pii_read(
            request=request, ctx=ctx, persons_served=len(window),
        )
    return _scim_json(_list_response(
        resources, total=len(filtered), start_index=start, count=page,
    ))


@router.get("/Users/{user_id}")
async def get_user(
    user_id: str,
    request: Request,
    db: AsyncSession = Depends(get_db),
    ctx: ConsumerContext = Depends(require_consumer(SCOPE_PERSONS_READ)),
):
    try:
        pid = uuid_mod.UUID(user_id)
    except ValueError:
        return _scim_error(404, f"User {user_id} not found.")
    # Resolve scope BEFORE the row lookup so the not-found path and the
    # out-of-filter path do equal work — otherwise the response-time gap
    # is an oracle for "this UUID is a real employee outside my filter".
    # Mirrors the REST get_person fix (security review 2026-06-11).
    scope_ids = await resolve_in_scope_person_ids(db, ctx.filter_config)
    row = await db.get(IdentityPerson, pid)
    if row is None or row.deleted_at is not None or row.id not in scope_ids:
        # Out-of-filter == not found: a distinguishable response would
        # leak existence of people outside the consumer's population.
        return _scim_error(404, f"User {user_id} not found.")
    memberships, managers = await _user_context(db, [row], scope_ids=scope_ids)
    include_phone = "mobile_phone" in ctx.granted_fields
    if include_phone:
        await audit_pii_read(request=request, ctx=ctx, persons_served=1)
    return _scim_json(_user_resource(
        row,
        membership=memberships.get(row.id),
        manager=managers.get(row.id),
        granted_fields=ctx.granted_fields,
        base_url=_base_url(request),
    ))


# ── Groups ────────────────────────────────────────────────────────────


@router.get("/Groups")
async def list_groups(
    request: Request,
    filter: str | None = Query(default=None),  # noqa: A002
    startIndex: int = Query(default=1),  # noqa: N803
    count: int = Query(default=_DEFAULT_COUNT),
    db: AsyncSession = Depends(get_db),
    ctx: ConsumerContext = Depends(require_consumer(SCOPE_ORG_READ)),
):
    clauses = parse_scim_filter(filter, allowed={
        "displayname": {"eq"},
        "externalid": {"eq"},
    })
    if isinstance(clauses, JSONResponse):
        return clauses

    nodes = list((await db.execute(
        select(IdentityOrgNode)
        .where(IdentityOrgNode.deleted_at.is_(None))
        .order_by(IdentityOrgNode.created_at.asc(), IdentityOrgNode.id.asc())
    )).scalars().all())
    for attr, _op, value in clauses:
        if attr == "displayname":
            nodes = [n for n in nodes if n.org_node_name == value]
        elif attr == "externalid":
            nodes = [n for n in nodes if n.org_node_source_id == value]

    start, page = _page_params(startIndex, count)
    window = nodes[start - 1 : start - 1 + page]
    members_by_node = await _load_group_members(
        db, ctx, [n.org_node_source_id for n in window],
    )
    base = _base_url(request)
    resources = [
        _group_resource(
            n, members=members_by_node.get(n.org_node_source_id, []),
            base_url=base,
        )
        for n in window
    ]
    return _scim_json(_list_response(
        resources, total=len(nodes), start_index=start, count=page,
    ))


@router.get("/Groups/{group_id}")
async def get_group(
    group_id: str,
    request: Request,
    db: AsyncSession = Depends(get_db),
    ctx: ConsumerContext = Depends(require_consumer(SCOPE_ORG_READ)),
):
    try:
        gid = uuid_mod.UUID(group_id)
    except ValueError:
        return _scim_error(404, f"Group {group_id} not found.")
    row = await db.get(IdentityOrgNode, gid)
    if row is None or row.deleted_at is not None:
        return _scim_error(404, f"Group {group_id} not found.")
    members = await _load_group_members(db, ctx, [row.org_node_source_id])
    return _scim_json(_group_resource(
        row, members=members.get(row.org_node_source_id, []),
        base_url=_base_url(request),
    ))


async def _load_group_members(
    db: AsyncSession, ctx: ConsumerContext, node_source_ids: list[str],
) -> dict[str, list[IdentityPerson]]:
    """In-filter persons whose primary membership sits on each node."""
    if not node_source_ids:
        return {}
    scope_ids = await resolve_in_scope_person_ids(db, ctx.filter_config)
    if not scope_ids:
        return {}
    memberships = (await db.execute(
        select(IdentityOrgMembership).where(
            IdentityOrgMembership.org_node_source_id.in_(node_source_ids),
            IdentityOrgMembership.is_primary.is_(True),
            IdentityOrgMembership.deleted_at.is_(None),
        )
    )).scalars().all()
    wanted = [m for m in memberships if m.person_id in scope_ids]
    person_ids = {m.person_id for m in wanted}
    persons: dict = {}
    if person_ids:
        rows = (await db.execute(
            select(IdentityPerson).where(
                IdentityPerson.id.in_(person_ids),
                IdentityPerson.deleted_at.is_(None),
            )
        )).scalars().all()
        persons = {p.id: p for p in rows}
    out: dict[str, list[IdentityPerson]] = {}
    for m in wanted:
        p = persons.get(m.person_id)
        if p is not None:
            out.setdefault(m.org_node_source_id, []).append(p)
    for members in out.values():
        members.sort(key=lambda p: (p.work_email or ""))
    return out


# ── Write rejection (read-only server) ────────────────────────────────


@router.api_route(
    "/{rest_of_path:path}",
    methods=["GET", "POST", "PUT", "PATCH", "DELETE"],
    include_in_schema=False,
)
async def scim_fallback(
    rest_of_path: str,
    request: Request,
    _ctx: ConsumerContext = Depends(require_consumer()),
):
    """Catch-all for everything the explicit GET routes don't serve.

    * Writes (POST/PUT/PATCH/DELETE) anywhere → 405 SCIM Error with
      ``Allow: GET`` — Helix is the authoritative identity source and
      does not accept SCIM writes.
    * GET on an unknown path (``/Bulk``, ``/Me``, typos) → 404 SCIM
      Error. Without the GET registration here, Starlette's partial
      route match answered 405 with a *reversed* ``Allow`` header
      (claiming the write methods were the allowed ones) in plain-JSON
      shape, and skipped auth entirely (review fix 2026-06-11).

    Auth runs first in both branches so unauthenticated probes get the
    same 401/429 treatment as reads.
    """
    if request.method == "GET":
        return _scim_error(404, f"Resource /{rest_of_path} not found.")
    resp = _scim_error(
        405,
        "This SCIM endpoint is read-only: Helix is the authoritative "
        "identity source and does not accept writes.",
    )
    resp.headers["Allow"] = "GET"
    return resp


__all__ = ["router"]