from __future__ import annotations

import re
from collections.abc import Iterable
from dataclasses import dataclass
from datetime import UTC, date, datetime
from uuid import UUID

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.corporate_domains import (
    email_is_corporate,
    get_corporate_email_domains,
)
from app.core.pinyin_utils import to_pinyin, to_pinyin_abbr
from app.core.safe_logging import get_safe_logger
from app.core.txn import best_effort
from app.identity.person_models import (
    IdentityExternalIdentity,
    IdentityOrgMembership,
    IdentityOrgNode,
    IdentityPerson,
)
from app.identity.successfactors_client import SuccessFactorsClient

logger = get_safe_logger(__name__)

# SF ``EmpJobRelationships`` relationshipType option codes for PingCAP's
# tenant (工作关系), verified live against sf-cn-prod:
# shuai.feng P0989 → 2448=HRBP (朱赟 P01115), 2447=调薪负责人 (倪捷 P0843)
# [2026-06-16], 2450=HR Ops (肖静 Jenny Xiao P0067) [2026-07-06; cross-
# checked: 46 people point to P0067 via 2450]. Tenant-specific — if SF
# re-keys the Job Relationship Type picklist, update these constants.
SF_REL_HRBP = "2448"
SF_REL_BUDGET_OWNER = "2447"  # 调薪负责人 → surfaced as 汇联易 "Budget Owner"
SF_REL_HR_OPS = "2450"  # HR Ops → second HR contact; new-hire card recipient

_ODATA_DATE_RE = re.compile(r"/Date\((?P<ms>-?\d+)(?P<offset>[+-]\d{4})?\)/")


@dataclass(slots=True)
class SuccessFactorsSyncSummary:
    fetched_jobs: int = 0
    fetched_users: int = 0
    upserted_org_nodes: int = 0
    upserted_people: int = 0
    upserted_memberships: int = 0


class SuccessFactorsSyncService:
    def __init__(self, db: AsyncSession, sf_client: SuccessFactorsClient | None = None):
        self.db = db
        self._sf_client = sf_client

    async def _get_client(self) -> SuccessFactorsClient:
        if self._sf_client is None:
            self._sf_client = await SuccessFactorsClient.from_config(session=self.db)
        return self._sf_client

    @property
    def sf_client(self) -> SuccessFactorsClient:
        if self._sf_client is None:
            raise RuntimeError(
                "SuccessFactorsClient has not been initialized; "
                "await _get_client() first"
            )
        return self._sf_client

    async def sync_users(self, limit: int | None = None) -> SuccessFactorsSyncSummary:
        from app.core.distributed_lock import LockBusy, distributed_lock
        try:
            async with distributed_lock("hris_full_roster_sync", ttl=600):
                return await self._sync_users_body(limit=limit)
        except LockBusy:
            raise RuntimeError(
                "sync_users: another full-roster sync is already running; "
                "skipping to prevent concurrent UNIQUE-constraint collision"
            )

    async def _sync_users_body(self, limit: int | None = None) -> SuccessFactorsSyncSummary:
        client = await self._get_client()
        companies = {
            company["externalCode"]: company
            for company in await client.list_active_companies()
            if company.get("externalCode")
        }
        departments = {
            department["externalCode"]: department
            for department in await client.list_active_departments()
            if department.get("externalCode")
        }
        locations = {
            location["externalCode"]: location
            for location in await client.list_active_locations()
            if location.get("externalCode")
        }
        # Position master-data — parity with the active hot path
        # (``hr/service.py`` ``_apply_event``): EmpJob's ``position``
        # code is opaque (``PO00001``) without the Position row's
        # ``externalName_defaultValue`` to project it into a readable
        # job title. Without this cache new persons synced via the
        # dormant full-roster path would land with ``job_title=None``
        # (the ``jobTitle``/``positionName(EN)`` fields were removed
        # from ``list_current_jobs`` on 2026-05-23 because they aren't
        # provisioned on PingCAP's SF tenant — see the comment in
        # ``successfactors_client.list_current_jobs``).
        #
        # Swallow tenant-side failures (Position entity not provisioned,
        # scope downgrade, transient 5xx) the same way the active path's
        # ``_safe`` wrapper does — better to land a person with a raw
        # ``position`` code than to abort the whole roster sync.
        try:
            positions_list = await client.list_active_positions()
        except Exception as exc:  # noqa: BLE001
            logger.warning(
                "successfactors-sync: list_active_positions failed (%s); "
                "job_title will fall back to raw position code",
                exc,
            )
            positions_list = []
        positions = {
            position["code"]: position
            for position in positions_list
            if position.get("code")
        }

        jobs = await client.list_current_jobs(limit=limit or None)
        summary = SuccessFactorsSyncSummary(fetched_jobs=len(jobs))
        summary.upserted_org_nodes = await self._upsert_org_nodes(departments)

        user_ids = sorted({job["userId"] for job in jobs if job.get("userId")})
        users = await client.get_users(user_ids)
        summary.fetched_users = len(users)

        jobs_by_user_id = {job["userId"]: job for job in jobs if job.get("userId")}
        users_by_user_id = {user["userId"]: user for user in users if user.get("userId")}
        corporate_domains = await get_corporate_email_domains(self.db)

        existing_people_result = await self.db.execute(
            select(IdentityPerson).where(
                IdentityPerson.source_system == "successfactors",
                IdentityPerson.hris_person_id.in_(user_ids),
            )
        )
        existing_people = {
            person.hris_person_id: person
            for person in existing_people_result.scalars().all()
        }

        synced_people: dict[str, IdentityPerson] = {}
        for user_id in user_ids:
            user = users_by_user_id.get(user_id)
            if not user:
                continue

            job = jobs_by_user_id[user_id]
            company = companies.get(job.get("company"))
            location = locations.get(job.get("location"))

            person = existing_people.get(user_id)
            # Lowercase work_email at every write site so the
            # case-insensitive UNIQUE index on identity_persons.work_email
            # never collides on "Foo@x" vs "foo@x". Matches the
            # normalization done in HrSyncService._resolve_unique_work_email.
            sf_email = (user.get("email") or "").strip().lower()
            # Corporate-mailbox protection (mirrors the active path's guard
            # in HrSyncService._resolve_unique_work_email): SF ``User.email``
            # stays the person's PERSONAL primary address (个人邮箱) even
            # after Helix mints the real corporate mailbox. An incoming
            # non-corporate email must never OVERWRITE an existing corporate
            # work_email (it feeds SSO / welcome email / write-back / the
            # UNIQUE index). When there is no corporate value to protect we
            # keep the historical behaviour; the Google provisioner derives
            # the real mailbox from the name regardless.
            if (
                sf_email
                and not email_is_corporate(sf_email, corporate_domains)
                and person is not None
                and email_is_corporate(person.work_email, corporate_domains)
            ):
                logger.warning(
                    "successfactors-sync: SF email %r for user_id=%s is "
                    "personal/non-corporate — keeping existing corporate "
                    "work_email %r (not overwriting)",
                    sf_email, user_id, person.work_email,
                )
                sf_email = ""
            placeholder = f"{user_id}@successfactors.local".lower()
            if person is None:
                if not sf_email:
                    # Mirror HrSyncService._resolve_unique_work_email's
                    # noisy-fallback behaviour so SRE / on-call sees a
                    # signal whenever a placeholder lands. The dormant
                    # full-roster path (this file) doesn't run in prod
                    # today, but if it gets re-wired we want every
                    # ``@successfactors.local`` row to have a paper trail.
                    logger.warning(
                        "successfactors-sync: SF returned no real email for "
                        "user_id=%s — creating row with placeholder %r",
                        user_id, placeholder,
                    )
                # Guard against two SF userId rows sharing the same work_email
                # (data quality issue in some SF configurations). Without this
                # check both rows INSERT, the second hits the UNIQUE constraint
                # on work_email and aborts the entire batch — all intermediate
                # state rolls back and the sync writes zero people. Detect
                # in-process duplicates (same email already processed this tick)
                # AND pre-existing DB duplicates (email belongs to a different
                # hris_person_id) and log+skip the second occurrence.
                effective_email = sf_email or placeholder
                if effective_email in (p.work_email for p in synced_people.values()):
                    logger.warning(
                        "successfactors-sync: duplicate work_email %r across "
                        "SF userId rows — first wins, skipping user_id=%s",
                        effective_email, user_id,
                    )
                    continue
                person = IdentityPerson(
                    source_system="successfactors",
                    hris_person_id=user_id,
                    work_email=effective_email,
                )
                self.db.add(person)

            # Backfill-on-every-sync: when SF now returns a real email and
            # the row sits on a stale placeholder, promote it. Conversely,
            # never downgrade an existing real email to a placeholder just
            # because SF returned an empty value on this tick (matches the
            # `current_email` guard in the active hr/service.py path).
            if sf_email and sf_email != person.work_email:
                logger.warning(
                    "successfactors-sync: work_email changed "
                    "hris_person_id=%s old=%r new=%r — "
                    "Google SSO / Feishu delivery will use new address from "
                    "next login. Verify SF data is correct.",
                    user.get("userId") or user.get("empId"),
                    person.work_email,
                    sf_email,
                )
                person.work_email = sf_email
            elif sf_email and not person.work_email:
                person.work_email = sf_email
            person.display_name = user.get("displayName") or person.display_name
            person.name_pinyin = to_pinyin(person.display_name)
            person.name_pinyin_abbr = to_pinyin_abbr(person.display_name)
            person.employee_no = user.get("empId") or person.employee_no
            person.employment_status = _normalize_employment_status(user, job)
            person.company_code = job.get("company")
            person.company_name = (company or {}).get("name")
            person.work_location_code = job.get("location")
            person.work_location_name = (location or {}).get("name")
            # Cost center comes from EmpJob.costCenter (added to the
            # SF client select-fields in the same PR that introduced
            # the Identity Sync Center). NULL preserved when the SF
            # record has no cost-center set so we don't overwrite a
            # downstream finance correction with an empty source.
            cc = job.get("costCenter")
            if cc:
                person.cost_center = cc
            # 区域邮件组/办公室 + 员工类型 — SF EmpJob.customString7 /
            # customString25 (migration 344). Stored RAW; translated by
            # the feishu_contact identity-sync target. NULL preserved on
            # absent values. Mirrors hr/service.py::_apply_event. NOTE:
            # ``gender`` is intentionally absent here — it lives on the
            # PerPersonal entity, which this dormant bootstrap path does
            # not fetch (the live hr/service path sets it from the
            # PerPersonal master cache).
            office_code = job.get("customString7")
            if office_code:
                person.office_code = office_code
            employee_type_code = job.get("customString25")
            if employee_type_code:
                person.employee_type_code = employee_type_code
            # 实际工作国家 — EmpJob.customString16 (HR-confirmed 2026-07-02),
            # precedence over FOCompany.country. Mirrors hr/service.py
            # (the live path). NULL preserved (company-country fallback
            # lives in the company-enrichment step of this path).
            work_country = (job.get("customString16") or "").strip().upper()
            if work_country:
                person.country = work_country
            # 是否推送到下游系统 — EmpJob.customString24 (migration 390).
            # Stored RAW for review; not yet gating the sync. Mirrors
            # hr/service.py. NULL preserved on absent.
            downstream_push = job.get("customString24")
            if downstream_push:
                person.sf_downstream_push_code = downstream_push
            person.is_china_employee = _is_china_employee(company, job.get("location"))
            # job_title resolution mirrors hr/service.py:_apply_event —
            # PingCAP's SF tenant doesn't expose ``jobTitle`` /
            # ``positionName(EN)`` on EmpJob (removed from select on
            # 2026-05-23 to stop the absent-check 400 storm), so the
            # canonical source is the Position master-data row keyed by
            # ``EmpJob.position`` code. The ``or`` chain keeps the SF
            # fields as defensive fallbacks in case a tenant migration
            # later provisions them.
            position_code = job.get("position")
            position_name = None
            if position_code:
                position_name = (positions.get(position_code) or {}).get(
                    "externalName_defaultValue"
                )
            person.job_title = (
                job.get("jobTitle")
                or job.get("positionNameEN")
                or job.get("positionName")
                or position_name
                or position_code
                or person.job_title
            )
            person.hire_date = _parse_date(user.get("hireDate"))
            person.source_updated_at = _parse_datetime(user.get("lastModifiedDateTime"))
            person.last_synced_at = datetime.now(UTC)

            synced_people[user_id] = person
            summary.upserted_people += 1

        await self.db.flush()

        for user_id, person in synced_people.items():
            job = jobs_by_user_id[user_id]
            await self._upsert_successfactors_identity(person, users_by_user_id[user_id])

            department_code = job.get("department")
            if not department_code:
                continue

            department = departments.get(department_code)
            await self._upsert_primary_membership(person.id, department_code, department, departments)
            summary.upserted_memberships += 1

        await self.db.flush()
        await self._resolve_managers(synced_people, jobs_by_user_id)
        await self.db.flush()
        # HRBP + Budget-Owner (调薪负责人) from SF 工作关系 — best-effort
        # enrichment, never breaks the roster sync. The method already
        # swallows the EmpJobRelationships pull failure; this outer guard
        # additionally insulates the core sync from any unexpected error in
        # the resolve/flush (this is a critical live path — an optional role
        # enrichment must never abort the roster commit).
        # SAVEPOINT-isolate (2026-08-06): without it the promise one comment
        # up — "an optional role enrichment must never abort the roster
        # commit" — is false for the failures that matter. A DB-level error
        # (deadlock, FK, dup-key) inside the resolve/flush deactivates the
        # transaction, so the swallow succeeds but ``backfill_manager_links``
        # + the flush right below raise, and the whole ~500-person roster
        # sync dies. The savepoint rolls back just the enrichment.
        async with best_effort(
            self.db, label="successfactors-sync: resolve HRIS roles", log=logger,
        ) as role_step:
            await self._resolve_hris_roles(synced_people, client)
            await self.db.flush()
        if not role_step.ok:
            logger.warning(
                "successfactors-sync: HRIS role resolve failed (%s); roster "
                "synced, HRBP/Budget-Owner not refreshed this run",
                role_step.reason(),
            )
        # Safety net for persons NOT in this roster slice whose declared
        # manager only now resolves (idempotent — a full roster already
        # had every manager re-resolved by ``_resolve_managers`` above,
        # so this typically links zero and is just defence-in-depth).
        from app.identity.identity_reconcile import backfill_manager_links
        await backfill_manager_links(self.db)
        await self.db.flush()

        return summary

    async def _upsert_org_nodes(self, departments: dict[str, dict]) -> int:
        department_codes = list(departments.keys())
        existing_nodes: dict[str, IdentityOrgNode] = {}
        if department_codes:
            result = await self.db.execute(
                select(IdentityOrgNode).where(
                    IdentityOrgNode.source_system == "successfactors",
                    IdentityOrgNode.org_node_type == "department",
                    IdentityOrgNode.org_node_source_id.in_(department_codes),
                )
            )
            existing_nodes = {
                node.org_node_source_id: node
                for node in result.scalars().all()
            }

        now = datetime.now(UTC)
        count = 0
        for department_code, department in departments.items():
            node = existing_nodes.get(department_code)
            if node is None:
                node = IdentityOrgNode(
                    source_system="successfactors",
                    org_node_type="department",
                    org_node_source_id=department_code,
                )
                self.db.add(node)

            node.org_node_code = department_code
            node.org_node_name = _department_name(department, department_code)
            node.parent_org_node_source_id = department.get("parent")
            node.org_path = _build_org_path(department, departments)
            node.org_level = _parse_org_level(department)
            node.node_status = "active"
            node.last_synced_at = now
            node.deleted_at = None
            count += 1

        await self.db.flush()

        # Deactivation reconcile — mirror ``hr.service`` hot path: SF's
        # active-only pull (``status eq 'A'``) means a deactivated dept
        # silently drops out of ``departments``; without this its node
        # keeps a stale ``active`` status. No-op on an empty pull, so a
        # transient failure can't tombstone the whole tree.
        from app.identity.identity_reconcile import deactivate_absent_org_nodes
        await deactivate_absent_org_nodes(self.db, departments.keys())

        return count

    async def _upsert_primary_membership(
        self,
        person_id: UUID,
        department_code: str,
        department: dict | None,
        departments: dict[str, dict],
    ) -> None:
        result = await self.db.execute(
            select(IdentityOrgMembership).where(
                IdentityOrgMembership.person_id == person_id,
                IdentityOrgMembership.is_primary.is_(True),
                IdentityOrgMembership.deleted_at.is_(None),
            )
        )
        membership = result.scalar_one_or_none()
        if membership is None:
            membership = IdentityOrgMembership(person_id=person_id)
            self.db.add(membership)

        department_name = _department_name(department, department_code)
        membership.source_system = "successfactors"
        membership.org_node_source_id = department_code
        membership.org_node_code = department_code
        membership.org_node_name = department_name
        membership.parent_org_node_source_id = (department or {}).get("parent")
        membership.org_path = _build_org_path(department, departments)
        membership.is_primary = True
        membership.membership_status = "active"
        membership.last_synced_at = datetime.now(UTC)

        # Department → platform-role auto-derivation (2026-06-17):
        # immediately (re)align the person's derived finance/hr grants to
        # their just-synced department so a Finance/HR hire or transfer
        # gains/loses the role on this sync instead of waiting for the
        # daily ``identity.department_role_reconcile`` backstop. Only
        # ``grant_source='derived'`` rows are touched — manual grants are
        # untouched. The pending membership is autoflushed by the SELECT
        # inside the sync helper, so the fresh org_path is visible.
        from app.identity.department_role_sync import (  # noqa: PLC0415
            sync_department_roles_for_person,
        )
        await sync_department_roles_for_person(self.db, person_id)

    async def _upsert_successfactors_identity(
        self,
        person: IdentityPerson,
        user: dict,
    ) -> None:
        # Look up WITHOUT deleted_at filter — see comment on the twin
        # path ``app/domains/hr/service.py::_upsert_external_identity``
        # for why. UNIQUE (person_id, provider) spans soft-deletes, so
        # an INSERT path that ignores the tombstoned row blows up on
        # flush. Revive the row instead.
        result = await self.db.execute(
            select(IdentityExternalIdentity).where(
                IdentityExternalIdentity.person_id == person.id,
                IdentityExternalIdentity.provider == "successfactors",
            )
        )
        identity = result.scalar_one_or_none()
        if identity is None:
            identity = IdentityExternalIdentity(
                person_id=person.id,
                provider="successfactors",
                provider_subject=user["userId"],
            )
            self.db.add(identity)
        elif identity.deleted_at is not None:
            identity.deleted_at = None
            identity.provider_subject = user["userId"]

        identity.provider_username = user.get("userId")
        identity.provider_email = user.get("email")
        identity.identity_status = "active"
        identity.is_login_identity = False
        identity.source_updated_at = _parse_datetime(user.get("lastModifiedDateTime"))
        identity.last_synced_at = datetime.now(UTC)

    async def _resolve_managers(
        self,
        synced_people: dict[str, IdentityPerson],
        jobs_by_user_id: dict[str, dict],
    ) -> None:
        manager_ids = sorted(
            {
                job["managerId"]
                for job in jobs_by_user_id.values()
                if job.get("managerId")
            }
        )
        existing_managers: dict[str, IdentityPerson] = {}
        if manager_ids:
            # Round-49: refuse leaver/soft-deleted candidates as
            # managers. HRIS update lag can leave the subordinate's
            # ``managerId`` pointing at a manager who just flipped to
            # ``left``; without these filters Helix would wire the
            # subordinate to a tombstoned mailbox and downstream
            # notifications would vanish silently. The clear-on-miss
            # branch below handles the resulting "no match" case
            # exactly the way it handles a literally-missing manager.
            #
            # Alias-aware since 2026-08-18 (``resolve_persons_by_hris_user_ids``
            # applies exactly those live filters, then falls back to
            # ``identity_hris_user_aliases`` for the leftovers): a manager
            # with two concurrent employments is referenced by whichever
            # userId owns the team, which may not be the userId Helix keyed
            # their person row on.
            from app.identity.identity_reconcile import (
                resolve_persons_by_hris_user_ids,
            )
            existing_managers = await resolve_persons_by_hris_user_ids(
                self.db, manager_ids,
            )

        from app.identity.identity_reconcile import normalize_manager_external_id
        for user_id, person in synced_people.items():
            manager_id = normalize_manager_external_id(
                jobs_by_user_id[user_id].get("managerId")
            )
            # Persist the declared id (migration 270) regardless of
            # whether it resolves this run — lets a late-arriving manager
            # be re-linked by ``backfill_manager_links`` and lets the
            # data-quality scanner separate a genuine HRIS "no manager"
            # (NULL — incl. SF's ``NO_MANAGER`` sentinel) from a Helix
            # link failure (set, but unresolved).
            person.manager_external_id = manager_id
            if not manager_id:
                person.manager_person_id = None
                continue

            manager = synced_people.get(manager_id) or existing_managers.get(manager_id)
            if manager and manager.id != person.id:
                person.manager_person_id = manager.id
            else:
                # Manager declared by SF but no IdentityPerson row exists
                # for that hris_person_id (the existing_managers query
                # already covers every previously-synced manager). Clear
                # rather than preserve: a stale pointer to a previous
                # manager is worse than a brief gap that the next sync
                # closes once the new manager onboards.
                if person.manager_person_id is not None:
                    logger.info(
                        "successfactors-sync: clearing manager_person_id "
                        "for user %s — declared manager %s has no IdentityPerson row",
                        user_id, manager_id,
                    )
                person.manager_person_id = None

    async def _resolve_hris_roles(
        self,
        synced_people: dict[str, IdentityPerson],
        client: SuccessFactorsClient,
    ) -> None:
        """Resolve HRBP / Budget-Owner over this run's batch.

        Thin wrapper around the shared :func:`apply_hris_role_relationships`
        so the dormant full-roster importer and the live ``hr``-domain
        confirm path share one resolver and can't drift.
        """
        await apply_hris_role_relationships(
            self.db, client, synced_people.values(),
        )


async def apply_hris_role_relationships(
    db: AsyncSession,
    client: SuccessFactorsClient,
    people: Iterable[IdentityPerson],
    *,
    scoped_pull: bool = False,
    effective_as_of: date | None = None,
) -> int:
    """Populate ``hris_hrbp_person_id`` + ``hris_budget_owner_person_id`` +
    ``hris_hrops_person_id`` on the people in ``people`` from SF
    ``EmpJobRelationships`` (工作关系).

    **Apply-only, never clear-on-absence (2026-06-16 fix).** Only writes a
    relationship type that the pull ACTUALLY returned for a person; a person
    (or a relationship type) absent from the pull is LEFT UNTOUCHED. This is
    deliberate: the caller (``confirm_batch``) runs this over the WHOLE active
    roster, and SF's ``EmpJobRelationships`` OData feed has been observed to
    return partial / intermittently-empty result sets. The previous version
    cleared every column not present in the pull, so a single partial pull on
    a whole-roster pass wiped HRBP / Budget-Owner for hundreds of people
    (prod incident: 382 → 2 populated over a few syncs). Apply-only converges
    to fully-populated as good pulls land and can never wipe on a bad one. The
    trade-off — a genuinely *removed* relationship lingers until replaced —
    is far cheaper than mass data loss, and a real replacement (the common
    case) still updates because the new relUserId IS in the pull.

    A relationship whose ``relUserId`` no longer resolves to a live Helix
    person (HRBP left the company) DOES clear that one person's pointer —
    that's a present-in-pull resolution miss, not an absence.

    ``scoped_pull=True`` restricts the SF ``EmpJobRelationships`` query to
    just the ``people``'s userIds (a ~3-row pull) instead of the tenant-wide
    feed. The live onboard-webhook path uses this to resolve ONE hire's HRBP
    / HR Ops *before* the account-creation card fires, so the card reaches
    them at build time rather than waiting for the next full-roster sync.
    (The apply-only / never-clear-on-absence semantics are unchanged — a
    scoped pull only ever touches the one person it's about.)

    ``effective_as_of`` (future-hire enrichment): the 24h-lead build fires
    ~1 day before a hire's start, when their 2448/2450 relationship rows are
    still future-dated and absent from SF's default as-of-today slice. Pass
    the hire's ``hire_date`` so SF returns the slice effective on the start
    date — populating ``hris_hrops_person_id`` / ``hris_hrbp_person_id`` at
    build so the welcome email can CC them. Best-effort like everything here:
    if HR hasn't entered the future-dated relationship yet, the pull is empty
    and the columns are left for the hire-day CC catchup to fill.

    Shared by BOTH inbound SF paths so they can't diverge: the dormant
    full-roster importer (:class:`SuccessFactorsSyncService`) and the live
    admin/cron confirm path (``hr`` domain ``confirm_batch``).

    The 汇联易 department sync + the Feishu contact HRBP push read these
    columns. **Best-effort** — a pull failure / empty pull leaves all columns
    untouched and returns ``0``; never aborts the caller's roster sync.
    Returns the number of people whose columns were (re)assigned.
    """
    people_by_hris: dict[str, IdentityPerson] = {
        p.hris_person_id: p for p in people if p.hris_person_id
    }
    if not people_by_hris:
        return 0

    try:
        _types = [SF_REL_HRBP, SF_REL_BUDGET_OWNER, SF_REL_HR_OPS]
        # Only pass ``user_ids`` on the scoped (per-hire webhook) path so the
        # tenant-wide call signature stays byte-identical to the pre-existing
        # one — keeps every current caller / test fake working unchanged.
        # Pass ``effective_as_of`` ONLY when set (future-hire enrichment), so
        # the default call signature stays byte-identical for every existing
        # caller + test fake (mirrors the ``scoped_pull`` rationale above) —
        # the fakes don't accept the kwarg. Explicit branches (not **kwargs)
        # keep this mypy-clean: ``effective_as_of`` is a plain ``date`` in the
        # enrichment branch, and the None path uses the pre-existing signature.
        uids = list(people_by_hris)
        if effective_as_of is not None:
            rels = (
                await client.list_job_relationships(
                    _types, user_ids=uids, effective_as_of=effective_as_of,
                )
                if scoped_pull
                else await client.list_job_relationships(
                    _types, effective_as_of=effective_as_of,
                )
            )
        elif scoped_pull:
            rels = await client.list_job_relationships(_types, user_ids=uids)
        else:
            rels = await client.list_job_relationships(_types)
    except Exception as exc:  # noqa: BLE001
        logger.warning(
            "successfactors-sync: EmpJobRelationships pull failed (%s); "
            "HRBP / Budget-Owner roles not refreshed this run", exc,
        )
        return 0

    # An empty (but non-erroring) pull is treated as "no data this run" —
    # NOT "every relationship was removed". On a whole-roster pass an empty
    # result must never wipe the columns; skip entirely.
    if not rels:
        logger.warning(
            "successfactors-sync: EmpJobRelationships pull returned 0 rows; "
            "leaving HRBP / Budget-Owner columns untouched (no clear-on-empty)",
        )
        return 0

    # userId → {relationshipType: relUserId}. The OData query carries no
    # date filter; SF returns the CURRENT-effective slice (verified live
    # 2026-06-16: P0989 → exactly one row per relationshipType), so the
    # per-(user,type) last-write here operates on a no-duplicate set. Even
    # if a tenant ever returned dated duplicates, the dept-side derivation
    # takes the member-shared MOST-COMMON value, so a single anomalous
    # member row can't flip a department's role.
    by_user: dict[str, dict[str, str]] = {}
    for r in rels:
        uid = r.get("userId")
        rtype = r.get("relationshipType")
        rel = r.get("relUserId")
        if uid and rtype and rel:
            by_user.setdefault(uid, {})[str(rtype)] = rel

    # Resolve every referenced relUserId (= hris_person_id) to a person.
    # Start from the caller's batch, then DB-load any target not in it.
    targets: dict[str, IdentityPerson] = dict(people_by_hris)
    referenced = {rel for m in by_user.values() for rel in m.values()}
    missing = sorted(referenced - set(targets))
    if missing:
        # Alias-aware (2026-08-18): ``relUserId`` is an EMPLOYMENT key, so
        # an HRBP / HR-Ops / budget owner who holds two concurrent
        # employments can be named by a userId Helix keeps no person row
        # for — the same blind spot that left ``managerId`` unresolvable.
        # Clearing the column on a miss (below) is the designed behaviour,
        # which is exactly what made this silent.
        from app.identity.identity_reconcile import (
            resolve_persons_by_hris_user_ids,
        )
        targets.update(await resolve_persons_by_hris_user_ids(
            db, missing, include_departed=True,
        ))

    # Iterate the PULL (by_user), not the whole roster — only people the pull
    # actually mentions get touched. For each, write ONLY the relationship
    # types present in their row; the other column is left as-is.
    assigned = 0
    for hris_id, m in by_user.items():
        person = people_by_hris.get(hris_id)
        if person is None:
            continue  # relationship for someone outside the caller's set
        changed = False
        if SF_REL_HRBP in m:
            hrbp = targets.get(m[SF_REL_HRBP])
            # Self-reference is meaningless for HRBP (you can't be your own
            # HR partner). An unresolvable target (HRBP left) clears it.
            new_hrbp = hrbp.id if hrbp and hrbp.id != person.id else None
            if new_hrbp != person.hris_hrbp_person_id:
                person.hris_hrbp_person_id = new_hrbp
                changed = True
        if SF_REL_BUDGET_OWNER in m:
            owner = targets.get(m[SF_REL_BUDGET_OWNER])
            # A person CAN be their own comp/budget owner (a senior lead).
            new_owner = owner.id if owner else None
            if new_owner != person.hris_budget_owner_person_id:
                person.hris_budget_owner_person_id = new_owner
                changed = True
        if SF_REL_HR_OPS in m:
            hrops = targets.get(m[SF_REL_HR_OPS])
            # Self-reference is meaningless for HR Ops; unresolvable target
            # (HR Ops person left) clears it. Mirrors the HRBP rule.
            new_hrops = hrops.id if hrops and hrops.id != person.id else None
            if new_hrops != person.hris_hrops_person_id:
                person.hris_hrops_person_id = new_hrops
                changed = True
        if changed:
            assigned += 1
    return assigned


def _department_name(department: dict | None, default_code: str) -> str:
    if not department:
        return default_code
    return (
        department.get("name_zh_CN")
        or department.get("name")
        or department.get("externalCode")
        or default_code
    )


def _build_org_path(department: dict | None, departments: dict[str, dict]) -> str | None:
    if not department:
        return None

    parts: list[str] = []
    root_name = department.get("cust_L0")
    if root_name:
        parts.append(str(root_name))

    for key in ("cust_L1", "cust_L2", "cust_L3", "cust_L4", "cust_L5"):
        code = department.get(key)
        if not code:
            continue
        name = _department_name(departments.get(code), str(code))
        if not parts or parts[-1] != name:
            parts.append(name)

    if not parts:
        return None
    return " / ".join(parts)


def _parse_org_level(department: dict | None) -> int | None:
    if not department:
        return None
    raw_level = department.get("cust_orglevel")
    if raw_level in (None, ""):
        return None
    try:
        return int(raw_level)
    except (TypeError, ValueError):
        return None


def _normalize_employment_status(user: dict, job: dict) -> str:
    """Map SF user/job to canonical Helix ``employment_status`` enum.

    Pre-launch P2 fix (Round 1, CR-010): the canonical states are
    ``active``, ``leaver_pending``, ``left`` (PLATFORM.md §2.1). The
    prior ``"inactive"`` literal was not a recognized state in the
    rest of the JML pipeline — if this dormant full-roster sync gets
    re-activated and writes ``employment_status='inactive'`` it would
    bypass ``LeaverBlockGuard`` (the listener fires on writes to
    ``'left'``) and the leaver-blocked dashboard would not classify
    the row. Map terminations to ``leaver_pending`` so the
    OffboardingProcess flow takes over; the ``'left'`` flip is then
    routed through ``LeaverBlockGuard.flip_to_left`` from
    ``hr_lifecycle.advance_service`` once blockers clear.
    """
    if not job:
        return "leaver_pending"
    raw_status = str(user.get("status") or "").strip().lower()
    if raw_status in {"inactive", "terminated", "f"}:
        return "leaver_pending"
    return "active"


def _is_china_employee(company: dict | None, location_code: str | None) -> bool:
    country = str((company or {}).get("country") or "").upper()
    if country == "CHN":
        return True
    return bool(location_code and location_code.upper().startswith("CN"))


def _parse_date(value: str | None) -> date | None:
    if not value:
        return None
    if isinstance(value, date) and not isinstance(value, datetime):
        return value
    if isinstance(value, datetime):
        return value.date()
    match = _ODATA_DATE_RE.fullmatch(value)
    if match:
        milliseconds = int(match.group("ms"))
        return datetime.fromtimestamp(milliseconds / 1000, tz=UTC).date()
    try:
        return datetime.fromisoformat(value).date()
    except (ValueError, TypeError):
        logger.warning(
            "successfactors-sync: unrecognized date format %r — skipping field",
            value,
        )
        return None


def _parse_datetime(value: str | None) -> datetime | None:
    if not value:
        return None
    if isinstance(value, datetime):
        return value
    match = _ODATA_DATE_RE.fullmatch(value)
    if match:
        milliseconds = int(match.group("ms"))
        return datetime.fromtimestamp(milliseconds / 1000, tz=UTC)
    try:
        normalized = value.replace("Z", "+00:00")
        return datetime.fromisoformat(normalized)
    except (ValueError, TypeError):
        logger.warning(
            "successfactors-sync: unrecognized datetime format %r — skipping field",
            value,
        )
        return None