"""Identity reconciliation primitives shared across the lifecycle.

Two convergence helpers that close the gap between a stored identity
field and the lifecycle event that *should* have maintained it — the
class of bug that left the data-quality scanner reporting stale
``orphan_login_identity`` and ``missing_manager`` findings:

* :func:`backfill_manager_links` — re-resolves ``manager_person_id`` for
  active persons whose HRIS-declared manager (``manager_external_id``)
  wasn't yet in Helix when their last sync event ran. The per-event
  resolvers (``hr.service._assign_manager`` /
  ``identity.successfactors_sync._resolve_managers``) deliberately
  *clear* the pointer on a miss (a stale pointer is worse than a brief
  gap — Round-20 audit), but nothing re-linked the subordinate once the
  manager finally onboarded. This pass runs at the tail of every sync so
  a late-arriving manager closes the gap on the very next run.

* :func:`deactivate_login_identities` — flips a departed person's
  Google / Feishu login bindings to non-login + inactive. The auth layer
  already rejects ``is_effectively_departed`` users on every request, so
  this is data-hygiene / defence-in-depth rather than the security gate:
  without it the binding row stays ``is_login_identity=True`` +
  ``identity_status='active'`` forever and the data-quality scanner keeps
  reporting a false "已离职但仍可登录" finding. The ``successfactors``
  provider row is intentionally left to the HR-sync owner
  (``hr.service._upsert_external_identity``) — it is never a login
  binding and mirrors the upstream employment status there.

Both operate only on identity-domain rows (``identity_persons`` /
``identity_external_identities``), so this module is import-safe from the
HR domain (``app/domains/hr``) and the leaver-blocked guard without
crossing a domain boundary the wrong way.
"""
from __future__ import annotations

import logging
import uuid
from collections.abc import Iterable
from datetime import UTC, datetime

from sqlalchemy import or_, select, update
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.sql.elements import ColumnElement

from app.core.safe_logging import get_safe_logger
from app.identity.person_models import (
    IdentityExternalIdentity,
    IdentityHrisUserAlias,
    IdentityOrgNode,
    IdentityPerson,
)

# Everything this module logs names a person: HRIS userIds, person UUIDs,
# org codes. All of it goes through the pseudonymizing SafeLogger — there
# is deliberately no plain ``logging.getLogger`` here, and
# ``test_safe_logging.py::test_alerted_modules_use_the_safe_logging_boundary``
# fails the build if one comes back.
safe_logger = get_safe_logger(__name__)

# External-identity providers that confer a Helix login (SSO). The
# ``successfactors`` binding is HRIS provenance, never a login identity.
LOGIN_PROVIDERS: tuple[str, ...] = ("google", "feishu")

# SF EmpJob.managerId carries the sentinel ``"NO_MANAGER"`` for the org
# apex (CEO) — it means "no declared manager", NOT a real userId. Persist
# it as NULL so it's never stored as a bogus external id, never chased by
# ``backfill_manager_links`` every sync, and so the data-quality scanner
# reads it as a genuine "HRIS declares no manager" (apex/gap) rather than
# a Helix link failure. Prod 2026-06-09: P0001 刘奇 (CEO) carries it.
_NO_MANAGER_SENTINELS = frozenset({"NO_MANAGER", "no_manager"})


def normalize_manager_external_id(raw: str | None) -> str | None:
    """Coerce SF's ``managerId`` to a real external id or ``None``.

    Strips whitespace and maps the empty string / ``NO_MANAGER`` sentinel
    to ``None`` (no declared manager). Anything else is returned verbatim.
    """
    if raw is None:
        return None
    v = raw.strip()
    if not v or v in _NO_MANAGER_SENTINELS:
        return None
    return v


# ── HRIS userId → person, alias-aware ─────────────────────────────────
#
# SF keys a *person* by ``personIdExternal`` and an *employment* by
# ``userId``; Helix keys a person by ``userId``
# (``identity_persons.hris_person_id``). The two coincide until someone
# holds two concurrent employments, at which point every HRIS reference
# to them (``EmpJob.managerId``, ``EmpJobRelationships.relUserId``) may
# name the *other* userId — one Helix deliberately keeps no row for.
# ``identity_hris_user_aliases`` is that indirection; everything that
# resolves an inbound userId goes through the helpers below so the
# primary-then-alias order is stated once.


async def resolve_person_by_hris_user_id(
    db: AsyncSession,
    user_id: str | None,
    *,
    hris_instance_id: uuid.UUID | None = None,
    exclude_person_id: uuid.UUID | None = None,
    source_system: str = "successfactors",
) -> IdentityPerson | None:
    """Resolve an inbound HRIS ``userId`` to a live person.

    Order: the primary column (``hris_person_id``) first, then the alias
    table. Both apply the live filters every per-event resolver uses
    (``deleted_at IS NULL``, ``employment_status != 'left'``) so a
    tombstoned or departed row is never wired up as someone's manager.

    ``hris_instance_id`` scopes ONLY the primary lookup — the Round-49
    anti-collision rule for two SF tenants seeded from a shared
    directory. The alias lookup is global because
    ``(source_system, alias_user_id)`` is UNIQUE, so a given userId maps
    to at most one alias row and cross-tenant confusion is impossible by
    construction (same argument :func:`backfill_manager_links` makes for
    resolving the primary column globally).
    """
    if not user_id:
        return None

    stmt = select(IdentityPerson).where(
        IdentityPerson.source_system == source_system,
        IdentityPerson.hris_person_id == user_id,
        IdentityPerson.deleted_at.is_(None),
        IdentityPerson.employment_status != "left",
    )
    if hris_instance_id is not None:
        stmt = stmt.where(
            or_(
                IdentityPerson.hris_instance_id == hris_instance_id,
                IdentityPerson.hris_instance_id.is_(None),
            )
        )
    if exclude_person_id is not None:
        stmt = stmt.where(IdentityPerson.id != exclude_person_id)
    person = await db.scalar(stmt)
    if person is not None:
        return person

    alias_stmt = (
        select(IdentityPerson)
        .join(IdentityHrisUserAlias, IdentityHrisUserAlias.person_id == IdentityPerson.id)
        .where(
            IdentityHrisUserAlias.source_system == source_system,
            IdentityHrisUserAlias.alias_user_id == user_id,
            IdentityPerson.deleted_at.is_(None),
            IdentityPerson.employment_status != "left",
        )
    )
    if exclude_person_id is not None:
        alias_stmt = alias_stmt.where(IdentityPerson.id != exclude_person_id)
    via_alias: IdentityPerson | None = await db.scalar(alias_stmt)
    return via_alias


async def resolve_persons_by_hris_user_ids(
    db: AsyncSession,
    user_ids: Iterable[str],
    *,
    source_system: str = "successfactors",
    include_departed: bool = False,
) -> dict[str, IdentityPerson]:
    """Batch form of :func:`resolve_person_by_hris_user_id`.

    Maps each **requested** userId to the live person it resolves to —
    keyed by what the caller asked for, not by the person's own
    ``hris_person_id``, so an alias hit lands under the alias id the HRIS
    payload actually carried. Two queries regardless of batch size:
    the primary column, then the alias table for whatever is left over.

    Not instance-scoped: batch callers are full-roster syncs that already
    resolve globally, and the UNIQUE on both key columns makes a userId
    map to at most one row either way.

    ``include_departed=True`` drops the ``employment_status != 'left'``
    filter, keeping the pre-existing semantics of the caller that only
    ever filtered soft-deletes (``apply_hris_role_relationships``). Adding
    the leaver filter there would ALSO clear the HRBP / HR-Ops / budget
    owner of everyone whose role-holder has since left — a real behaviour
    change unrelated to alias resolution, so it stays opt-in.
    """
    wanted = sorted({u for u in user_ids if u})
    if not wanted:
        return {}

    live_only: list[ColumnElement[bool]] = [IdentityPerson.deleted_at.is_(None)]
    if not include_departed:
        live_only.append(IdentityPerson.employment_status != "left")

    resolved: dict[str, IdentityPerson] = {}
    rows = (await db.execute(
        select(IdentityPerson).where(
            IdentityPerson.source_system == source_system,
            IdentityPerson.hris_person_id.in_(wanted),
            *live_only,
        )
    )).scalars().all()
    for person in rows:
        if person.hris_person_id:
            resolved[person.hris_person_id] = person

    missing = [u for u in wanted if u not in resolved]
    if not missing:
        return resolved

    alias_rows = (await db.execute(
        select(IdentityHrisUserAlias.alias_user_id, IdentityPerson)
        .join(IdentityPerson, IdentityPerson.id == IdentityHrisUserAlias.person_id)
        .where(
            IdentityHrisUserAlias.source_system == source_system,
            IdentityHrisUserAlias.alias_user_id.in_(missing),
            *live_only,
        )
    )).all()
    for alias_user_id, person in alias_rows:
        resolved[alias_user_id] = person
    return resolved


async def hris_user_alias_owner(
    db: AsyncSession,
    user_id: str | None,
    *,
    source_system: str = "successfactors",
) -> IdentityPerson | None:
    """The canonical person a secondary ``userId`` belongs to, or None.

    Unlike :func:`resolve_person_by_hris_user_id` this consults ONLY the
    alias table and does NOT filter the owner by employment status —
    callers use it to answer "is this userId a duplicate of somebody we
    already have?", and a leaver's duplicate employment is still a
    duplicate, not a new hire.
    """
    if not user_id:
        return None
    owner: IdentityPerson | None = await db.scalar(
        select(IdentityPerson)
        .join(IdentityHrisUserAlias, IdentityHrisUserAlias.person_id == IdentityPerson.id)
        .where(
            IdentityHrisUserAlias.source_system == source_system,
            IdentityHrisUserAlias.alias_user_id == user_id,
        )
    )
    return owner


async def record_hris_user_alias(
    db: AsyncSession,
    *,
    person: IdentityPerson,
    alias_user_id: str,
    reason: str,
    person_id_external: str | None = None,
    hris_instance_id: uuid.UUID | None = None,
    note: str | None = None,
    source_system: str = "successfactors",
) -> bool:
    """Map a secondary HRIS ``userId`` onto ``person``. Idempotent.

    Returns True when a row was created, False when the mapping already
    existed (``last_seen_at`` is refreshed either way) or when a guard
    refused it.

    Two guards, both of which would otherwise make a person invisible to
    HRIS sync — the alias table short-circuits the person upsert, so
    aliasing a userId that IS somebody's primary key would stop syncing
    that human entirely:

    * never alias a userId to the person whose own ``hris_person_id`` it
      is (self-alias, a no-op that only adds a way to be wrong later);
    * never alias a userId that a LIVE person holds as their primary
      ``hris_person_id``. A tombstoned (soft-deleted) row does not block
      it — retiring a ghost row and aliasing its userId onto the real
      person is exactly the intended cleanup.
    """
    alias_user_id = (alias_user_id or "").strip()
    if not alias_user_id:
        return False
    if alias_user_id == (person.hris_person_id or ""):
        safe_logger.event(
            "HRIS_USER_ALIAS_REFUSED",
            {"user_id": alias_user_id, "person_id": str(person.id),
             "reason": "self_alias"},
            level=logging.WARNING,
        )
        return False

    live_primary = await db.scalar(
        select(IdentityPerson.id).where(
            IdentityPerson.source_system == source_system,
            IdentityPerson.hris_person_id == alias_user_id,
            IdentityPerson.deleted_at.is_(None),
        )
    )
    if live_primary is not None:
        # Aliasing a live person's own primary key would short-circuit their
        # person upsert and stop syncing them entirely — refuse loudly.
        safe_logger.event(
            "HRIS_USER_ALIAS_REFUSED",
            {"user_id": alias_user_id, "person_id": str(live_primary),
             "reason": "live_primary_hris_person_id"},
            level=logging.WARNING,
        )
        return False

    existing = await db.scalar(
        select(IdentityHrisUserAlias).where(
            IdentityHrisUserAlias.source_system == source_system,
            IdentityHrisUserAlias.alias_user_id == alias_user_id,
        )
    )
    now = datetime.now(UTC)
    if existing is not None:
        existing.last_seen_at = now
        if existing.person_id != person.id:
            safe_logger.event(
                "HRIS_USER_ALIAS_REPOINTED",
                {"user_id": alias_user_id,
                 "previous_person_id": str(existing.person_id),
                 "person_id": str(person.id), "reason": reason},
                level=logging.WARNING,
            )
            existing.person_id = person.id
            existing.reason = reason
            existing.note = note
        if person_id_external and not existing.person_id_external:
            existing.person_id_external = person_id_external
        if hris_instance_id is not None and existing.hris_instance_id is None:
            existing.hris_instance_id = hris_instance_id
        return False

    db.add(IdentityHrisUserAlias(
        id=uuid.uuid4(),
        person_id=person.id,
        source_system=source_system,
        alias_user_id=alias_user_id,
        person_id_external=person_id_external,
        hris_instance_id=hris_instance_id,
        reason=reason,
        note=note,
        first_seen_at=now,
        last_seen_at=now,
    ))
    await db.flush()
    safe_logger.event(
        "HRIS_USER_ALIAS_RECORDED",
        {"user_id": alias_user_id, "person_id": str(person.id),
         "reason": reason, "person_id_external": person_id_external or ""},
        level=logging.WARNING,
    )
    return True


async def backfill_manager_links(
    db: AsyncSession,
    *,
    hris_instance_id: uuid.UUID | None = None,  # noqa: ARG001 — see below
) -> int:
    """Resolve ``manager_person_id`` for active persons that carry a
    declared ``manager_external_id`` but have no live link yet.

    This is the **reconciliation** counterpart to the conservative
    per-event resolver (``hr.service._assign_manager`` /
    ``successfactors_sync._resolve_managers``). The per-event resolver
    instance-scopes its manager lookup (``hris_instance_id == sync OR
    NULL``) — a Round-49 anti-collision rule for two SF tenants seeded
    from a shared directory reusing the same ``userId``. But that scope
    ALSO refuses legitimate **cross-instance reporting lines** (PingCAP
    prod: a 6-person secondary SF tenant whose members report to — and
    are reported to by — the main tenant), leaving those managers
    permanently NULL.

    This pass resolves the declared manager **globally** across instances,
    which is safe because ``(source_system, hris_person_id)`` is UNIQUE
    (``uq_identity_persons_source``, spanning soft-deletes) — so a given
    ``hris_person_id`` maps to **at most one** Helix row, ever. The
    Round-49 "two tenants reuse a userId" collision the per-event scope
    guards against is therefore impossible at the row level; cross-tenant
    confusion cannot occur here. Because this pass only ever **sets** a
    NULL pointer (never clears, so it can't introduce a stale pointer)
    and runs at the tail of every sync *after* the per-event clear, it
    safely "upgrades" a same-tenant-only link to the correct
    cross-instance manager.

    Resolution rule, mirroring the per-event live-manager filters:
    * exactly one live (non-deleted, non-``left``) person carries that
      ``hris_person_id`` → link it;
    * none (manager gone / soft-deleted duplicate / not yet synced) →
      skip; the data-quality scanner surfaces it as a genuine gap.

    ``hris_instance_id`` is accepted for call-site compatibility but no
    longer used to scope — see above. Idempotent: a re-run resolves the
    same target and writes nothing new. Returns the number newly linked.
    """
    rows = (await db.execute(
        select(IdentityPerson).where(
            IdentityPerson.deleted_at.is_(None),
            IdentityPerson.employment_status == "active",
            IdentityPerson.manager_person_id.is_(None),
            IdentityPerson.manager_external_id.is_not(None),
            IdentityPerson.manager_external_id != "",
        )
    )).scalars().all()
    if not rows:
        return 0

    linked = 0
    for person in rows:
        # Alias-aware (2026-08-18): the declared manager may be a
        # SECONDARY userId of someone Helix already has — a concurrent
        # employment whose assignment owns the team. Chasing only the
        # primary column left four active people manager-less on every
        # sync (prod: 孙若曦 P0050 / 80100061). See
        # ``resolve_person_by_hris_user_id``.
        manager = await resolve_person_by_hris_user_id(
            db,
            person.manager_external_id,
            exclude_person_id=person.id,
        )
        if manager is None:
            continue
        person.manager_person_id = manager.id
        linked += 1
        safe_logger.event(
            "IDENTITY_MANAGER_BACKFILL_LINKED",
            {
                "person_id": str(person.id),
                "hris_person_id": person.hris_person_id or "",
                "manager_person_id": str(manager.id),
                "manager_external_id": person.manager_external_id or "",
            },
        )
    if linked:
        await db.flush()
    return linked


async def deactivate_login_identities(
    db: AsyncSession,
    person_id: uuid.UUID,
) -> int:
    """Flip a departed person's Google / Feishu login bindings to
    ``is_login_identity=False`` + ``identity_status='inactive'``.

    Only touches still-live login rows (``deleted_at IS NULL`` and
    actually login-enabled or active), so it is idempotent — a second
    call finds nothing to flip and writes nothing. Returns the number of
    binding rows changed. Best-effort by contract: callers run this as a
    leaver-hygiene sweep and swallow/​audit failures; the auth layer's
    ``is_effectively_departed`` gate is the real lock.
    """
    rows = (await db.execute(
        select(IdentityExternalIdentity).where(
            IdentityExternalIdentity.person_id == person_id,
            IdentityExternalIdentity.provider.in_(LOGIN_PROVIDERS),
            IdentityExternalIdentity.deleted_at.is_(None),
            or_(
                IdentityExternalIdentity.is_login_identity.is_(True),
                IdentityExternalIdentity.identity_status == "active",
            ),
        )
    )).scalars().all()
    changed = 0
    for ext in rows:
        ext.is_login_identity = False
        ext.identity_status = "inactive"
        changed += 1
    if changed:
        await db.flush()
        safe_logger.event(
            "IDENTITY_LOGIN_BINDINGS_DEACTIVATED",
            {"person_id": str(person_id), "count": changed},
        )
    return changed


async def deactivate_absent_org_nodes(
    db: AsyncSession,
    active_source_ids: Iterable[str],
    *,
    source_system: str = "successfactors",
    org_node_type: str = "department",
) -> int:
    """Mark HRIS org nodes that vanished from the upstream active set inactive.

    The SF org sync is an *active-only upsert*: ``list_active_departments``
    filters ``status eq 'A'``, so a department deactivated in SF simply drops
    out of the response — and because the upsert only ever touches the codes
    that came back (and hard-codes ``node_status='active'``), the orphaned
    ``identity_org_nodes`` row keeps its last-written ``active`` status
    forever. This pass closes that gap: every currently-active node NOT in
    the freshly-synced active set is flipped to ``node_status='inactive'``.

    Round-trips cleanly — if a department is re-activated in SF it reappears
    in the active pull and the upsert sets it back to ``active``.

    Two invariants the CALLER must uphold (both already hold at the two SF
    sync call sites, which only run this after a *full* active-department
    pull and gate the upsert on a non-empty result):

    * ``active_source_ids`` must be the COMPLETE active set for the run — a
      partial roster slice would wrongly deactivate live orgs.
    * Never call with an empty set: a transient upstream failure yielding
      ``{}`` would otherwise tombstone the whole tree. This function
      no-ops on empty input as a second line of defence.

    Multi-instance note (verified 2026-07-02): ``identity_org_nodes`` carries
    NO ``hris_instance_id`` discriminator — every SF instance keys its org
    nodes by ``(source_system='successfactors', org_node_source_id)``. This is
    SAFE only because the PingCAP CN + EU SF instances share ONE org
    foundation: each instance's ``list_active_departments`` returns the SAME
    global active-department set (confirmed against prod — EU employees'
    department codes are present-and-active after a CN-instance reconcile
    run). So a per-instance call site still passes the complete active set and
    the precondition above holds. If the instances are ever split onto
    disjoint org foundations, this reconcile MUST be made instance-aware (add
    ``hris_instance_id`` to the node key + filter) before that lands, or each
    instance will tombstone the other's tree.
    """
    codes = set(active_source_ids)
    if not codes:
        return 0

    now = datetime.now(UTC)
    result = await db.execute(
        update(IdentityOrgNode)
        .where(
            IdentityOrgNode.source_system == source_system,
            IdentityOrgNode.org_node_type == org_node_type,
            IdentityOrgNode.node_status == "active",
            IdentityOrgNode.deleted_at.is_(None),
            IdentityOrgNode.org_node_source_id.not_in(codes),
        )
        .values(node_status="inactive", last_synced_at=now, updated_at=now)
    )
    count = result.rowcount or 0
    if count:
        await db.flush()
        safe_logger.event(
            "IDENTITY_ORG_NODES_DEACTIVATED",
            {
                "source_system": source_system,
                "org_node_type": org_node_type,
                "count": count,
                "upstream_active": len(codes),
            },
        )
    return count