"""Write the corporate email Helix minted for a new hire back to HRIS. When Helix auto-creates a Google Workspace account for a joiner it picks the final ``primary_email`` (the HRIS-suggested 邮箱名, or a collision-resolved ``…01`` … ``…09`` variant). HR's identity team needs that final address back in SuccessFactors so the HRIS record reflects the real mailbox. 2026-06-11: HR handed over the real callback — the SF OData ``upsert`` endpoint writing a ``PerEmail`` row (sample they provided):: POST https://api…sapsf.cn/odata/v2/upsert?$format=json [{"__metadata": {"uri": "PerEmail"}, "emailType": "1584", # tenant picklist id, 办公邮箱 "personIdExternal": "00000939", # SF person key (≠ userId) "emailAddress": "6222@xx.com", "isPrimary": true}] This replaced the earlier "generic HTTP POST to a URL HR will provide" placeholder. The upsert reuses the existing SuccessFactors connector credentials (same integration row the read sync uses) — no new secrets. ``personIdExternal`` is resolved per call via EmpEmployment because Helix only persists ``userId`` (``identity_persons.hris_person_id``). Config (``platform_settings.hris.email_writeback``, all optional):: { "enabled": true, # default true — connector creds are the opt-in "email_type": "1046", # PerEmail.emailType picklist id for 办公邮箱. # 1046 = business email on PingCAP's CN/EU # tenants (metadata probe 2026-04); HR's vendor # sample showed 1584 — override here if our # tenant ever disagrees. "personal_email_type": "1047", # 个人邮箱 type — demoted from primary # when is_primary makes the corporate email # the new primary. "is_primary": true # make the corporate email the SF PRIMARY. # SF allows one primary, so the write-back # atomically demotes the personal (1047) # primary in the same upsert batch. } Two-step primary switch (2026-07-15) ------------------------------------ The write-back fires at TWO moments in a hire's lifecycle and they carry different ``isPrimary`` intent: 1. **Onboard event arrives** (weeks/days ahead) — ``reserve_work_email_for`` REGISTERS the planned corporate address with ``isPrimary=false``. The mailbox doesn't exist yet, so promoting it to SF primary (and demoting the personal email, SF's only live pre-boarding contact channel) would point SF's primary at a dead mailbox. Address-only is the safe write. 2. **Hire-date account creation** (``create_account_for``, ~24h ahead) — once the live Google mailbox exists, the write-back runs with the configured default (``isPrimary=true`` + personal demotion), flipping the same PerEmail row to primary. Idempotent: same ``(personIdExternal, emailType)`` key; a collision-corrected address at creation self-heals because step 2 re-sends the final address. The function NEVER raises — a HRIS write-back failure must not roll back a Google account that already exists. Callers get a structured dict they can fold into their audit row. """ from __future__ import annotations from typing import Any from sqlalchemy.ext.asyncio import AsyncSession from app.core.exceptions import AppException from app.core.safe_logging import get_safe_logger from app.identity.hris_client_routing import ( # Re-exported under the historical private name: the multi-instance # routing test patches ``hris_email_writeback._sf_client_for_person``, # and the helper moved out only because a second write-back needed it. sf_client_for_person as _sf_client_for_person, ) from app.identity.person_models import IdentityPerson from app.integrations.service import get_platform_setting logger = get_safe_logger(__name__) _SETTING_KEY = "hris.email_writeback" # PerEmail.emailType picklist id for 办公邮箱 on PingCAP's SF tenants. _DEFAULT_EMAIL_TYPE = "1046" # PerEmail.emailType for 个人邮箱 (see successfactors_client.list_personal_emails) # — demoted from primary when the corporate email is promoted to primary. _DEFAULT_PERSONAL_EMAIL_TYPE = "1047" _MSG_CAP = 500 async def notify_hris_email_assigned( db: AsyncSession, *, person: IdentityPerson, work_email: str, trace_id: str, is_primary_override: bool | None = None, ) -> dict[str, Any]: """Upsert the assigned corporate email onto SF's ``PerEmail``. ``is_primary_override`` (2026-07-15): when not ``None`` it WINS over the configured ``hris.email_writeback.is_primary``. The onboard-event reserve path (:func:`reserve_work_email_for`) passes ``False`` so the address is merely REGISTERED on the record weeks ahead of day-1 — without promoting a not-yet-existing mailbox to SF primary and without demoting the personal email (SF's only live pre-boarding contact channel) from primary. The actual hire-date account creation (:func:`create_account_for`) leaves it ``None`` so the configured default (primary=true + personal demotion) applies at the moment the mailbox goes live. See the two-step design note in the module docstring. Returns a structured result; never raises. Possible ``status``: * ``skipped`` — switch off, person has no ``hris_person_id``, or the SuccessFactors connector isn't configured. ``reason`` explains. * ``sent`` — SF acknowledged the upsert (record-level OK). * ``failed`` — SF reachable but rejected the write (record-level ERROR in the 200 envelope, or no EmpEmployment row to resolve ``personIdExternal`` from). * ``error`` — transport failure or HTTP >= 400 talking to SF. """ setting = await get_platform_setting(_SETTING_KEY, default={}, session=db) if not isinstance(setting, dict): setting = {} if setting.get("enabled") is False: return {"status": "skipped", "reason": "hris.email_writeback disabled"} hris_user_id = (person.hris_person_id or "").strip() if not hris_user_id: return { "status": "skipped", "reason": "person has no hris_person_id (not an HRIS-sourced row)", } email_type = str(setting.get("email_type") or _DEFAULT_EMAIL_TYPE).strip() personal_email_type = str( setting.get("personal_email_type") or _DEFAULT_PERSONAL_EMAIL_TYPE ).strip() if isinstance(is_primary_override, bool): is_primary = is_primary_override else: is_primary = setting.get("is_primary") if not isinstance(is_primary, bool): is_primary = True try: client = await _sf_client_for_person(db, person) except AppException as exc: # Connector not configured — same steady-state semantics the old # "no url yet" branch had: log at INFO, skip, never block. logger.info( "hris.email_writeback: SuccessFactors connector not configured " "— skipping write-back of %s for person_id=%s (%s)", work_email, person.id, exc.message, ) return {"status": "skipped", "reason": exc.message} # The whole body is SF HTTP I/O — ``db`` is threaded through only for # the connector lookup above and is never written to here (the unit # tests pass ``None`` for it). Nothing to isolate. # savepoint-exempt: no DB writes; body is external HTTP only try: person_id_external = await client.get_person_id_external(hris_user_id) if not person_id_external: logger.warning( "hris.email_writeback: no EmpEmployment row for userId=%s " "— cannot resolve personIdExternal; NOT writing %s back " "(person_id=%s)", hris_user_id, work_email, person.id, ) return { "status": "failed", "reason": ( f"no EmpEmployment row for userId={hris_user_id}; " "personIdExternal unresolved" ), } # Build the upsert batch. When making the corporate email the # PRIMARY, SF rejects a second simultaneous primary — so in the # SAME atomic batch we also DEMOTE the person's current personal # primary (PerEmail emailType 1047) to isPrimary=false. SF keys # PerEmail by (personIdExternal, emailType), so this targets the # existing personal row; the personal address itself is preserved # (just no longer the primary). Without a personal email on file, # or when not setting primary, we send the single corporate row. records: list[dict[str, Any]] = [{ "emailType": email_type, "personIdExternal": person_id_external, "emailAddress": work_email, "isPrimary": is_primary, }] demoted_personal: str | None = None personal_email = (person.personal_email or "").strip().lower() if ( is_primary and personal_email and personal_email != work_email.strip().lower() ): records.append({ "emailType": personal_email_type, "personIdExternal": person_id_external, "emailAddress": personal_email, "isPrimary": False, }) demoted_personal = personal_email resp = await client.upsert_per_emails(records) # The corporate row is records[0] → resp[0]; report on it. record = resp[0] if resp else {} except Exception as exc: # noqa: BLE001 — contract: never raise logger.warning( "hris.email_writeback: upsert error for person_id=%s (%s) — " "%s: %s", person.id, work_email, type(exc).__name__, exc, ) return { "status": "error", "reason": f"{type(exc).__name__}: {exc}"[:_MSG_CAP], } # SF's upsert returns HTTP 200 even when a record fails — the # per-record envelope carries the verdict. EVERY record in the batch # (corporate promote + personal demote) must be OK, else the primary # switch didn't take. def _rec_ok(rec: dict[str, Any]) -> bool: st = str(rec.get("status") or "").upper() hc = rec.get("httpCode") return st == "OK" or (not st and isinstance(hc, int) and hc < 300) record_status = str(record.get("status") or "").upper() http_code = record.get("httpCode") all_ok = bool(resp) and all(_rec_ok(r) for r in resp) if all_ok: logger.info( "hris.email_writeback: upserted PerEmail %s for " "personIdExternal=%s (person_id=%s, emailType=%s, isPrimary=%s%s)", work_email, person_id_external, person.id, email_type, is_primary, f", demoted personal {demoted_personal}" if demoted_personal else "", ) return { "status": "sent", "person_id_external": person_id_external, "email_type": email_type, "is_primary": is_primary, "demoted_personal": demoted_personal, "sf_status": record_status or None, "sf_http_code": http_code, } # Surface the first failing record's message for the audit row. failed = next((r for r in resp if not _rec_ok(r)), record) f_status = str(failed.get("status") or "").upper() f_http = failed.get("httpCode") f_msg = str(failed.get("message") or "")[:_MSG_CAP] logger.warning( "hris.email_writeback: SF rejected PerEmail upsert of %s for " "personIdExternal=%s (person_id=%s) — status=%s httpCode=%s %s", work_email, person_id_external, person.id, f_status, f_http, f_msg, ) return { "status": "failed", "person_id_external": person_id_external, "sf_status": f_status or None, "sf_http_code": f_http, "reason": f_msg or f"SF record status {f_status or f_http}", }