"""One-shot: backfill SF status='f' leavers into ``identity_persons``.

Why this exists
---------------
The 2026-04-21 bootstrap full sync called ``list_active_users()`` which
filters ``status eq 't'``. Every SF user whose ``status`` was ``'f'``
(Forfeited / terminated) at bootstrap time was permanently skipped:
incremental sync only fetches rows whose ``lastModifiedDateTime`` is
after the previous cursor, so anyone SF flipped to ``'f'`` before the
bootstrap moment lives in a sync blind spot.

Concrete prod fallout (verified 2026-05-30):

* 48 real PingCAP ex-employees were absent from ``identity_persons``
  despite having full SF records.
* A verified former-employee case was absent from Helix. It had:
    * multiple ``cmdb_servers`` rows with a legacy username owner —
      ``owner_person_id`` could not link to a real person.
    * multiple ``cmdb_network_owner_history`` rows from
      ``restore_dns_owners_from_legacy.py`` with
      an unresolved ``context.legacy_owner`` and
      ``context.resolution='unresolved'`` because the script could not
      resolve the username to an IdentityPerson.
    * ``software_external_holder`` rows (Slack / Feishu / Google)
      with ``person_id=NULL`` (the SaaS accounts are still ACTIVE —
      the company is paying for seats two months after termination).

This script:

1. Pulls SF ``status='f'`` users from every enabled ``hris`` integration.
2. Skips users whose work_email matches ``HR_SKIP_EMAIL_DOMAINS``
   (catches known SAP service-account placeholder patterns).
3. Skips users already in ``identity_persons`` (by ``hris_person_id``
   OR ``work_email`` — handles the SF-side dup-userId pattern where the
   canonical row owns the real email and the dup hits an orphan ID).
4. For each genuinely-missing user:
   a. ``INSERT`` ``identity_persons`` with:
      * ``source_system='successfactors'``
      * ``hris_person_id=user.userId``
      * ``work_email`` (resolved with the same uniqueness logic the
        live sync uses)
      * ``employment_status='left'`` if endDate is past, else
        ``'leaver_pending'`` (matches the live ``_upsert_person``
        contract introduced in the same PR)
      * ``termination_date`` = EmpEmployment endDate (or
        lastModifiedDateTime fallback if SF didn't give us an endDate)
      * ``hris_record_state='present'`` — SF still has the row, we
        just hadn't imported it yet.
   b. ``INSERT`` ``identity_external_identities`` provider=``successfactors``
      so future incremental syncs find the row.
   c. ``INSERT`` ``hr_lifecycle_events`` event_type=``offboard`` with
      effective_date=termination_date so the lifecycle audit is
      reconstructable from the same ``hr_lifecycle_events`` table the
      rest of the platform pivots on.
   d. ``INSERT`` ``audit_logs`` action=``identity_persons.backfill``
      so the operation is traceable.
5. Re-link orphan anchors for the newly-created persons:
   a. ``cmdb_servers``: where ``JSON_EXTRACT(raw_data,'$.owner')``
      matches the SF username (case-insensitive), set
      ``owner_person_id`` to the freshly-created person.
   b. ``cmdb_network_owner_history``: where
      ``JSON_EXTRACT(context,'$.legacy_owner')`` matches and
      ``new_owner_person_id`` is NULL, set
      ``new_owner_person_id`` + flip resolution to ``resolved``.
   c. ``software_external_holder``: where ``email`` matches the
      person's ``work_email`` (case-insensitive) AND ``person_id`` is
      NULL, set ``person_id``.

Dry-run by default. Pass ``--commit`` to actually persist.

Usage::

    cd server
    python scripts/backfill_sf_leavers_2026_05_30.py             # dry-run
    python scripts/backfill_sf_leavers_2026_05_30.py --commit    # apply
"""
from __future__ import annotations

import argparse
import asyncio
import json
import logging
import re as _re
import uuid
from datetime import UTC, date, datetime
from typing import Any

from sqlalchemy import func, or_, select

# Side-effect model imports so SQLAlchemy mappers resolve. The script
# touches identity_persons + audit + cmdb_servers + cmdb_network_owner_history
# + software_external_holder + hr_lifecycle_events → load everything.
import app.audit.models  # noqa: F401
import app.auth.models  # noqa: F401
import app.domains.asset.models  # noqa: F401
import app.domains.cmdb.models  # noqa: F401
import app.domains.hr.models  # noqa: F401
import app.identity.person_models  # noqa: F401
from app.audit.models import AuditLog
from app.core.database import async_session_factory
from app.core.safe_logging import get_safe_logger
from app.domains.asset.models import SoftwareExternalHolder
from app.domains.cmdb.models import CmdbNetworkOwnerHistory, CmdbServer
from app.domains.hr.models import HrLifecycleEvent
from app.domains.hr.service import SuccessFactorsLifecycleService
from app.identity.person_models import (
    IdentityExternalIdentity,
    IdentityPerson,
)
from app.identity.successfactors_auth import build_auth_from_config
from app.identity.successfactors_client import SuccessFactorsClient
from app.integrations.compat import list_hris_instances

logger = get_safe_logger("backfill_sf_leavers_2026_05_30")

# SF endDate format is ``/Date(ms)/``. Parsing helper mirrors
# ``app.identity.successfactors_sync._parse_date`` but we don't import
# it directly because the sync module has heavy SQLAlchemy mapper
# setup costs we don't need.

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


def _parse_sf_date(value: str | None) -> date | None:
    if not value:
        return None
    if isinstance(value, date):
        return value
    if isinstance(value, datetime):
        return value.date()
    match = _ODATA_DATE_RE.fullmatch(value)
    if match:
        ms = int(match.group("ms"))
        return datetime.fromtimestamp(ms / 1000, tz=UTC).date()
    try:
        return datetime.fromisoformat(value).date()
    except (ValueError, TypeError):
        return None


def _parse_sf_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:
        ms = int(match.group("ms"))
        return datetime.fromtimestamp(ms / 1000, tz=UTC)
    try:
        return datetime.fromisoformat(value.replace("Z", "+00:00"))
    except (ValueError, TypeError):
        return None


# ─── Skip predicate ──────────────────────────────────────────────────


def _is_skippable(user: dict[str, Any]) -> str | None:
    """Return reason string if user should be skipped, else None.

    Mirrors the live sync's skip semantics:

    * No userId → broken SF row.
    * No real email → ``HR_SKIP_EMAIL_DOMAINS`` (covers
      SAP service ghosts, implementation accounts,
      partners, and the ``@successfactors.local`` placeholder).
    * ``HrSyncService.is_placeholder_email`` — catches ``dummy<digits>@``
      pattern that SF fills in for employees with no real mailbox.
    """
    user_id = user.get("userId")
    if not user_id:
        return "no_userId"
    email = (user.get("email") or "").strip().lower()
    if not email:
        # No email at all — the live sync would synthesize
        # ``<userId>@successfactors.local`` and then HR_SKIP_EMAIL_DOMAINS
        # would catch that placeholder. Skip with explicit reason so
        # the operator sees the count, but don't synthesize a fake
        # row from a backfill script (we're recovering historical
        # leavers; if SF lost the email there's nothing to anchor on).
        return "no_email"
    if SuccessFactorsLifecycleService.is_placeholder_email(email):
        return "placeholder_email"
    if SuccessFactorsLifecycleService.email_domain_in_skip_list(email):
        return "skip_domain"
    return None


# ─── Anchor re-link helpers ──────────────────────────────────────────


async def _relink_cmdb_servers(
    session, *, person: IdentityPerson, sf_username: str, dry_run: bool,
) -> int:
    """Set owner_person_id on cmdb_servers whose raw_data.owner == sf_username.

    UCloud's raw payload carries ``raw_data.owner`` as a free-text
    string of the operator's UCloud sub-account name — which on the
    PingCAP tenant can differ from the SF userId because UCloud may use
    a legacy short name. We match BOTH spellings: lower-cased SF
    userId AND lower-cased local part of work_email — covers the
    common ``firstName.lastName`` vs ``firstNamelastName`` skews.
    """
    candidates: set[str] = {sf_username.lower()}
    if person.work_email and "@" in person.work_email:
        local = person.work_email.split("@", 1)[0].lower()
        candidates.add(local)
        # Also try ``firstname.lastname`` / ``firstnamelastname``
        # variants so dotted and undotted legacy names can match.
        if "." in local:
            candidates.add(local.replace(".", ""))
        else:
            # Cannot disambiguate split-points without name data; skip.
            pass

    if not candidates:
        return 0

    # JSON_UNQUOTE(JSON_EXTRACT(...)) is MySQL/TiDB syntax; SQLAlchemy
    # passes the literal through. ``LOWER(...)`` for case-insensitive
    # match. Filter to rows where owner_person_id is NULL or different
    # so we don't re-write the same value.
    matching = (await session.execute(
        select(CmdbServer).where(
            func.lower(
                func.json_unquote(func.json_extract(CmdbServer.raw_data, "$.owner"))
            ).in_(candidates),
            or_(
                CmdbServer.owner_person_id.is_(None),
                CmdbServer.owner_person_id != str(person.id),
            ),
        )
    )).scalars().all()

    if not matching:
        return 0

    print(
        f"  cmdb_servers: {len(matching)} candidate row(s)"
    )
    for srv in matching:
        if dry_run:
            continue
        else:
            # Only flip if the row truly has no owner_person_id, OR
            # the row's current owner is an anchor-less
            # NULL. We refuse to overwrite a real prior owner — that
            # would be silently stealing assets from another person
            # the legacy system had assigned to.
            if srv.owner_person_id is None:
                srv.owner_person_id = str(person.id)
            else:
                print("    SKIP one cmdb_server: refusing to overwrite its owner")
    return len(matching)


async def _relink_dns_legacy_owner(
    session, *, person: IdentityPerson, sf_username: str, dry_run: bool,
) -> int:
    """Fill in cmdb_network_owner_history rows whose legacy_owner matches.

    ``restore_dns_owners_from_legacy.py`` can produce unresolved rows.
    Now that the IdentityPerson exists we can mark them resolved. Critical:
    we ONLY flip
    ``new_owner_person_id`` from NULL → person.id; we don't touch any
    row whose ``new_owner_person_id`` was already set (that would
    overwrite a deliberate later transfer).
    """
    candidates: set[str] = {sf_username.lower()}
    if person.work_email and "@" in person.work_email:
        candidates.add(person.work_email.split("@", 1)[0].lower())

    matching = (await session.execute(
        select(CmdbNetworkOwnerHistory).where(
            func.lower(
                func.json_unquote(
                    func.json_extract(CmdbNetworkOwnerHistory.context, "$.legacy_owner")
                )
            ).in_(candidates),
            CmdbNetworkOwnerHistory.new_owner_person_id.is_(None),
        )
    )).scalars().all()
    if not matching:
        return 0
    print(
        f"  cmdb_network_owner_history: {len(matching)} unresolved row(s)"
    )
    for hist in matching:
        if dry_run:
            continue
        else:
            hist.new_owner_person_id = str(person.id)
            ctx = dict(hist.context or {})
            ctx["resolution"] = "resolved"
            ctx["resolved_via"] = "backfill_sf_leavers_2026_05_30"
            ctx["resolved_at"] = datetime.now(UTC).isoformat()
            hist.context = ctx
    return len(matching)


async def _relink_software_external_holder(
    session, *, person: IdentityPerson, dry_run: bool,
) -> int:
    """Anchor SaaS seat rows whose email matches the new person.

    These are the Slack / Feishu / Google active accounts the
    company keeps paying for after termination. Anchoring them to
    the now-existing Person row means the next ``software seat sync``
    will treat them as Helix-known, the offboarding /
    leaver_blocked code can pick them up, and the SaaS revoke flow
    can fire.

    NOTE: We anchor (set ``person_id``) but do NOT mark them resolved
    and do NOT call any external revoke. Revoke must be a deliberate
    operator action (the person IS gone, but a same-day delete of
    Slack history etc. is an irreversible operation). The anchor
    enables downstream tooling to do the right thing.
    """
    if not person.work_email:
        return 0

    matching = (await session.execute(
        select(SoftwareExternalHolder).where(
            func.lower(SoftwareExternalHolder.email)
            == person.work_email.lower(),
            SoftwareExternalHolder.person_id.is_(None),
        )
    )).scalars().all()
    if not matching:
        return 0
    print(
        f"  software_external_holder: {len(matching)} row(s)"
    )
    for holder in matching:
        if dry_run:
            continue
        else:
            holder.person_id = str(person.id)
    return len(matching)


# ─── Per-user create path ────────────────────────────────────────────


async def _create_person(
    session,
    *,
    user: dict[str, Any],
    employment: dict[str, Any] | None,
    instance_id: str,
    today: date,
) -> IdentityPerson:
    """Insert IdentityPerson + external_identity + lifecycle event.

    Mirrors the contract ``HrSyncService._upsert_person`` would set
    for a brand-new offboard event with ``_full_sync=True``. We
    duplicate the relevant slice rather than invoke the live sync
    pipeline because the live pipeline expects a confirmed batch +
    stages full org-tree caches we don't need for a recovery write.
    """
    user_id = user["userId"]
    email = (user.get("email") or "").strip().lower()
    if not email:
        raise ValueError("_create_person: refusing to create a person with no email")

    # Defence against the live sync's UNIQUE collision on work_email
    # (same human, multiple SF userIds → second tries the same email).
    # If a row already owns this email we MUST stop — the operator's
    # decision is whether to dedup at the SF source first, not for
    # this script to silently bypass.
    pre = (await session.execute(
        select(IdentityPerson).where(
            func.lower(IdentityPerson.work_email) == email,
            IdentityPerson.deleted_at.is_(None),
        )
    )).scalar_one_or_none()
    if pre is not None:
        raise ValueError(
            "_create_person: work email already belongs to another person; "
            "likely an SF-side duplicate userId"
        )

    termination_date = _parse_sf_date(
        (employment or {}).get("endDate")
        or user.get("lastModifiedDateTime")
    )
    hire_date = _parse_sf_date(user.get("hireDate"))
    source_updated_at = _parse_sf_datetime(user.get("lastModifiedDateTime"))

    # Match _upsert_person's same-day-termination guard: insert at
    # 'left' iff endDate is >= 30 days past, else 'leaver_pending'.
    # The leaver_blocked listener allows INSERT-as-'left' (skips
    # session.new); UPDATE-to-'left' would still need the guard.
    if termination_date and (today - termination_date).days >= 30:
        employment_status = "left"
    elif termination_date:
        employment_status = "leaver_pending"
    else:
        # SF says status='f' but gave us no endDate at all. Treat as
        # past leaver — pick today as the conservative termination
        # date so reporting shows ``left`` rather than the much
        # more-attention-grabbing ``leaver_pending``.
        employment_status = "left"
        termination_date = today

    person = IdentityPerson(
        id=uuid.uuid4(),
        source_system="successfactors",
        hris_instance_id=instance_id,
        hris_person_id=user_id,
        employee_no=user.get("empId") or user_id,
        work_email=email,
        display_name=user.get("displayName"),
        employment_status=employment_status,
        hire_date=hire_date,
        termination_date=termination_date,
        source_updated_at=source_updated_at,
        last_synced_at=datetime.now(UTC),
        hris_record_state="present",
    )
    session.add(person)

    # External identity row so the next incremental sync recognises
    # the user. Same shape ``successfactors_sync._upsert_successfactors_identity``
    # writes for a live user.
    session.add(IdentityExternalIdentity(
        id=uuid.uuid4(),
        person_id=person.id,
        provider="successfactors",
        provider_subject=user_id,
        provider_username=user_id,
        provider_email=email,
        identity_status="inactive",
        is_login_identity=False,
        source_updated_at=source_updated_at,
        last_synced_at=datetime.now(UTC),
    ))

    # Lifecycle event so downstream reports / dashboards (pivot on
    # ``hr_lifecycle_events.event_type``) reconstruct the offboard.
    if termination_date:
        session.add(HrLifecycleEvent(
            id=uuid.uuid4(),
            person_id=person.id,
            event_type="offboard",
            effective_date=termination_date,
            detail={
                "source": "backfill_sf_leavers_2026_05_30",
                "userId": user_id,
                "displayName": user.get("displayName"),
                "hireDate": str(hire_date) if hire_date else None,
                "via": "one_shot_script",
            },
        ))

    return person


# ─── Driver ──────────────────────────────────────────────────────────


async def run(*, dry_run: bool) -> int:
    today = date.today()
    summary = {
        "instances_checked": 0,
        "sf_forfeited_total": 0,
        "skipped_already_in_helix": 0,
        "skipped_no_email": 0,
        "skipped_placeholder_email": 0,
        "skipped_skip_domain": 0,
        "skipped_no_userId": 0,
        "skipped_other": 0,
        "persons_created": 0,
        "lifecycle_events_created": 0,
        "audit_rows_written": 0,
        "cmdb_servers_relinked": 0,
        "dns_history_resolved": 0,
        "saas_holders_anchored": 0,
        "errors": [],
    }

    async with async_session_factory() as session:
        instances = await list_hris_instances(session=session)

    for instance_index, (row, cfg) in enumerate(instances, start=1):
        if not row.enabled:
            continue
        summary["instances_checked"] += 1
        print(f"\n===== enabled HRIS instance {instance_index} =====")
        client = SuccessFactorsClient(
            base_url=cfg["base_url"], auth=build_auth_from_config(cfg),
        )

        # Fetch all status='f' users for this instance.
        try:
            forfeited = await client._list_entities(
                "User",
                select_fields=[
                    "userId", "username", "email", "empId", "displayName",
                    "status", "hireDate", "lastModifiedDateTime",
                ],
                filter_clause="status eq 'f'",
            )
        except Exception as exc:  # noqa: BLE001
            summary["errors"].append({
                "instance_index": instance_index,
                "stage": "list_forfeited",
                "error_type": type(exc).__name__[:80],
            })
            continue

        summary["sf_forfeited_total"] += len(forfeited)
        forfeited_user_ids = [
            u["userId"] for u in forfeited if u.get("userId")
        ]

        # Fetch EmpEmployment rows so we know endDate per user. Reuse
        # ``client`` — ``SuccessFactorsClient`` does not hold a
        # persistent httpx pool (each ``_get_json`` opens its own
        # ``async with httpx.AsyncClient(...)`` block), so there's no
        # close/aclose to call and no benefit to a second instance.
        try:
            employments = (
                await client.get_employments_for_users(forfeited_user_ids)
                if forfeited_user_ids else []
            )
        except Exception as exc:  # noqa: BLE001
            summary["errors"].append({
                "instance_index": instance_index,
                "stage": "list_employments",
                "error_type": type(exc).__name__[:80],
            })
            employments = []

        # Keep latest endDate per userId (re-hires can carry multiple
        # EmpEmployment rows). Compare as parsed datetimes, NOT raw
        # ``/Date(ms)/`` strings — lexicographic compare disagrees with
        # chronological order across digit-count boundaries; harmless
        # today (every PingCAP termination is 13-digit ms) but the
        # next system with legacy 12-digit ms breaks silently.
        emp_by_uid: dict[str, dict] = {}
        for emp in employments:
            uid = emp.get("userId")
            if not uid or not emp.get("endDate"):
                continue
            prior = emp_by_uid.get(uid)
            if prior is None:
                emp_by_uid[uid] = emp
            else:
                cur_lmd = _parse_sf_datetime(emp.get("lastModifiedDateTime"))
                prior_lmd = _parse_sf_datetime(prior.get("lastModifiedDateTime"))
                if (cur_lmd or datetime.min.replace(tzinfo=UTC)) > (
                    prior_lmd or datetime.min.replace(tzinfo=UTC)
                ):
                    emp_by_uid[uid] = emp

        for user in forfeited:
            skip_reason = _is_skippable(user)
            if skip_reason:
                key = f"skipped_{skip_reason}"
                if key in summary:
                    summary[key] += 1
                else:
                    summary["skipped_other"] += 1
                continue

            user_id = user["userId"]
            email = (user.get("email") or "").strip().lower()

            # Per-user transaction so one failing user doesn't poison
            # the whole batch (mirrors the live sync's SAVEPOINT
            # discipline).
            async with async_session_factory() as session:
                # Look-up by userId OR email — either pre-existing
                # match means "we already have this person" and
                # the row should not be re-created.
                existing = (await session.execute(
                    select(IdentityPerson).where(
                        or_(
                            IdentityPerson.hris_person_id == user_id,
                            func.lower(IdentityPerson.work_email) == email,
                        ),
                        IdentityPerson.deleted_at.is_(None),
                    )
                )).scalars().first()
                if existing is not None:
                    summary["skipped_already_in_helix"] += 1
                    continue

                try:
                    person = await _create_person(
                        session,
                        user=user,
                        employment=emp_by_uid.get(user_id),
                        instance_id=str(row.id),
                        today=today,
                    )
                    print("  create one missing identity person")
                    summary["persons_created"] += 1
                    if person.termination_date:
                        summary["lifecycle_events_created"] += 1

                    # Audit row — independent of the SQLAlchemy listener
                    # (which fires from before_flush). Explicit
                    # ``identity_persons.backfill`` action so the
                    # operator can grep audit_logs for the script run.
                    session.add(AuditLog(
                        id=uuid.uuid4(),
                        trace_id=uuid.uuid4(),
                        actor_user_id=None,
                        actor_type="script",
                        action="identity_persons.backfill",
                        resource_type="identity_person",
                        resource_id=str(person.id),
                        status="success",
                        detail={
                            "script": "backfill_sf_leavers_2026_05_30",
                            "sf_user_id": user_id,
                            "sf_instance": row.slug,
                            "work_email": email,
                            "display_name": user.get("displayName"),
                            "employment_status": person.employment_status,
                            "termination_date": (
                                str(person.termination_date)
                                if person.termination_date else None
                            ),
                        },
                    ))
                    summary["audit_rows_written"] += 1

                    # Flush so the freshly-created person has a real ID
                    # the re-link helpers can write into FK columns.
                    await session.flush()

                    cnt = await _relink_cmdb_servers(
                        session, person=person, sf_username=user_id,
                        dry_run=dry_run,
                    )
                    summary["cmdb_servers_relinked"] += cnt

                    cnt = await _relink_dns_legacy_owner(
                        session, person=person, sf_username=user_id,
                        dry_run=dry_run,
                    )
                    summary["dns_history_resolved"] += cnt

                    cnt = await _relink_software_external_holder(
                        session, person=person, dry_run=dry_run,
                    )
                    summary["saas_holders_anchored"] += cnt

                    if dry_run:
                        await session.rollback()
                    else:
                        await session.commit()

                except Exception as exc:  # noqa: BLE001
                    await session.rollback()
                    summary["errors"].append({
                        "instance_index": instance_index,
                        "stage": "create_person",
                        "error_type": type(exc).__name__[:80],
                    })
                    logger.exception("backfill person failed")

    print("\n===== SUMMARY =====")
    print(json.dumps(summary, indent=2, default=str, ensure_ascii=False))
    if dry_run:
        print(
            "\nDRY-RUN — no rows written. Re-run with --commit to persist."
        )
    else:
        print("\nCOMMITTED. Verify via /admin/users + scheduled_job_runs.")
    return 0


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--commit",
        action="store_true",
        help="Actually write changes. Without this flag the script "
             "prints planned changes and rolls back.",
    )
    parser.add_argument(
        "-v", "--verbose", action="store_true",
        help="Verbose logging (DEBUG level).",
    )
    args = parser.parse_args()

    logging.basicConfig(
        level=logging.DEBUG if args.verbose else logging.INFO,
        format="%(asctime)s %(levelname)s %(name)s — %(message)s",
    )

    return asyncio.run(run(dry_run=not args.commit))


if __name__ == "__main__":
    raise SystemExit(main())