"""Alias-aware email → IdentityPerson lookup. External systems (Zoom, Slack, Jira, Feishu seat sync, …) key their users by email, but the email they use isn't always the canonical ``work_email`` — it may be one of the Google Workspace aliases mirrored into ``identity_persons.email_aliases`` (populated today by the ``email_group`` domain's group-mirror sync; previously by ``google.dept_sync`` which was retired in PR #identity-sync-center). This helper builds a single dict keyed by every known address (primary + aliases) so the caller can do a cheap ``lookup.get(external_email.lower())`` without needing to re-query or branch on "did we match the primary or an alias". All keys are lowercased; callers should lowercase the candidate before looking up. Designed for sync loops iterating thousands of external users — one **chunked** bulk SELECT → in-memory dict → O(1) lookups. **Chunking note (2026-05-02):** an earlier shape concatenated all candidate emails into a single OR-chain (``lower(work_email) IN (…)`` plus N ``json_contains(email_aliases, …)`` clauses). With small inputs (Zoom ~150 users, Slack ~100, Jira ~70) this performed fine. With Feishu's tenant-users sync (~480 users → ~960 placeholders + a giant query plan), MySQL 8.0 + asyncmy started returning ``OperationalError 1412 'Table definition has changed, please retry transaction'`` — a misleading error code that the server emits when its prepared-statement metadata cache is overwhelmed. The fix is plain query chunking: split the candidate set into batches of :data:`_CHUNK_SIZE`, run one query per batch, merge. The shape of the return value is unchanged. Smaller-N callers see the same single round-trip they always had. """ from __future__ import annotations import json from collections.abc import Iterable from sqlalchemy import func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.identity.leaver_notice import ( STATUS_LEAVER_PENDING, is_within_leaver_notice_period, ) from app.identity.person_models import IdentityPerson # Per-query candidate-email cap. MySQL accepts 65k placeholders in # theory, but a single query that mixes a 100-element ``IN (…)`` with # 100 ``json_contains`` clauses already pushes the prepared-statement # metadata cache enough to trigger spurious 1412 "table definition has # changed" errors on MySQL 8.0 + asyncmy. 100 is comfortably under that # threshold while keeping round-trip count low (~5 queries for a # 500-user tenant). Lower if a future driver hits the same wall. _CHUNK_SIZE = 100 async def build_person_by_email_index( db: AsyncSession, candidate_emails: Iterable[str], *, exclude_employment_statuses: tuple[str, ...] = (), leaver_notice_grace: bool = True, ) -> dict[str, IdentityPerson]: """Return a lowercase-email → IdentityPerson dict covering both ``work_email`` and ``email_aliases``. Only non-deleted persons are considered. A person with 3 aliases appears 4 times in the dict (primary + 3). If the same address somehow lives on two persons, the one with the matching ``work_email`` wins over the one holding it as an alias; among alias-only collisions the last-written in the query order wins (deterministic but not meaningful — this shouldn't happen in a clean dataset). ``exclude_employment_statuses``: pass a tuple of values (e.g. ``("left", "leaver_pending")``) to also filter out persons in those states. Seat-sync handlers pass this so the daily reconcile doesn't re-import an active seat for someone who is already gone on the Helix side but still lingering on the vendor side. Default empty tuple keeps the legacy "match anyone non-deleted" semantics for callers (org chart, search index) that need every person. ``leaver_notice_grace`` (default True) refines the ``"leaver_pending"`` exclusion with the notice-period invariant in :mod:`app.identity.leaver_notice`: a confirmed FUTURE leaver whose last working day has not ended locally still legitimately holds their seats, so they stay IN the index. Only once the LWD has passed (or the status flips to ``left``) are they excluded and their seats fall through to drift cleanup. A ``leaver_pending`` row with NO ``termination_date`` also stays — we can't prove the last day has passed, and acting early is the failure mode this guards. Why this is the default rather than opt-in (2026-08-11): every seat-sync handler that excluded ``leaver_pending`` outright drift-revoked a still-working employee's seats the night IT confirmed the termination, then re-imported them as BRAND-NEW ``software_seat_allocations`` rows on the next tick that re-matched the person — new row ids that defeated offboarding's per-seat revoke-task dedup and produced duplicate cancel events in ``/admin/software/cancellations`` (P0779 / P01361 cohort). Pass ``leaver_notice_grace=False`` only for a caller that genuinely wants a hard status exclusion. The query is chunked under the hood — see module docstring for the rationale. The dict-build pass is also chunk-local so a person who matches across two chunks (e.g. has both work_email and an alias in the candidate set) still ends up with primary > alias resolution. """ lowered = {e.strip().lower() for e in candidate_emails if e and e.strip()} if not lowered: return {} # ``leaver_pending`` can't be excluded in SQL when the notice-period # grace applies — whether the LWD has passed depends on the person's # own timezone (``resolve_timezone``), which SQL can't evaluate. Drop # it from the SQL predicate and re-apply the refined test in Python # over the matched rows. sql_excluded: tuple[str, ...] = tuple(exclude_employment_statuses) notice_grace_applies = ( leaver_notice_grace and STATUS_LEAVER_PENDING in sql_excluded ) if notice_grace_applies: sql_excluded = tuple( s for s in sql_excluded if s != STATUS_LEAVER_PENDING ) # Stable order so the alias-vs-primary collision rule below is # deterministic even when the caller passed an unordered set. candidates: list[str] = sorted(lowered) index: dict[str, IdentityPerson] = {} matched_persons: list[IdentityPerson] = [] for start in range(0, len(candidates), _CHUNK_SIZE): chunk = candidates[start : start + _CHUNK_SIZE] if not chunk: continue # Pull any person whose work_email OR any alias matches a # candidate from THIS chunk. MySQL's JSON_CONTAINS is the # portable check; TiDB mirrors it. # # ``json.dumps(e)`` (NOT a hand-rolled f-string) handles the # rare case where an address contains characters that need # JSON escaping ('"', '\\'). RFC 5321 allows quoted-local-part # emails; even though our HRIS never emits them, a malformed # value reaching this helper would otherwise produce invalid # SQL and crash the whole sync run. work_email_match = func.lower(IdentityPerson.work_email).in_(chunk) alias_clauses = [ func.json_contains( IdentityPerson.email_aliases, json.dumps(e), ) for e in chunk ] conditions = [work_email_match, *alias_clauses] where_clauses = [ IdentityPerson.deleted_at.is_(None), or_(*conditions), ] if sql_excluded: where_clauses.append( IdentityPerson.employment_status.notin_(sql_excluded), ) rows = (await db.execute( select(IdentityPerson).where(*where_clauses), )).scalars().all() matched_persons.extend(rows) # Build the index in two passes against the merged result set so # the "primary wins over alias" rule survives chunking — a primary # match in chunk N still beats an alias match for the same address # found in chunk M. seen_person_ids: set = set() deduped: list[IdentityPerson] = [] for p in matched_persons: if p.id in seen_person_ids: continue seen_person_ids.add(p.id) deduped.append(p) if notice_grace_applies: # Keep confirmed future leavers (still inside their notice # period); drop the ones whose last working day has ended. deduped = [ p for p in deduped if p.employment_status != STATUS_LEAVER_PENDING or is_within_leaver_notice_period(p) ] # Alias first, primary last — so primary overwrites alias on collision. for p in deduped: for alias in p.email_aliases or []: key = (alias or "").strip().lower() if key and key in lowered: index[key] = p for p in deduped: if p.work_email: key = p.work_email.strip().lower() if key in lowered: index[key] = p return index __all__ = ["build_person_by_email_index"]