"""与 GitHub 的双向对账 —— 台账说的和 GitHub 上真实的,差在哪。

邀请生命周期(``invitation_lifecycle``)只推进**我们自己发出去的**那些
邀请。这里管的是另外两个方向,而它们才是治理面真正关心的:

* **GitHub 上有、Helix 台账里没有** —— 有人绕过 Helix 直接被加进了组织。
  这不是错误(org owner 本来就能在 GitHub 上直接点),但它必须被**看见**:
  一条 Helix 不知道的成员资格,离职时不会被回收。落一行 ``source=reconcile``
  的台账记录,``person_id`` 尽力解析,解析不出就留空进「认不出是谁」。
* **Helix 说 active、GitHub 上没有** —— 有人在 GitHub 上被直接移除了。
  台账要跟上,否则它会一直宣称一个已经不存在的访问权。

另外顺手做两件小事:

* **回填 person_id**。台账里 ``person_id`` 为空的行,如果它的 handle 现在
  能对上某个人的 ``github_id`` 了(那个人后来补登记了),就把它接上。
  这条路能修的只是「以后新登记的人」—— 实测存量的 217 个认不出的 handle
  里,能靠字面对上的是 0 个,那批要靠治理面的画像 + AI 建议去清。
* **刷坐席用量**。要的凭证和组织清单与对账完全相同,单开一个 job 只会
  得到第二处会和这里说法不一致的地方。

## 一道刹车,不能省

「GitHub 返回了空成员列表」和「这个组织真的一个人都没有」在 HTTP 层
长得一模一样。没有刹车的话,一次组织改名、一次权限被撤、一次 API 行为
变化,都会让对账把整个组织的成员**全部标成已移除** —— 而那正是离职回收
读的那张表。所以:成功拿到的成员数为 0、而台账里有 ≥
:data:`EMPTY_FETCH_BRAKE_MIN_LIVE` 条活着的记录时,**拒绝对账**并把原因
记在组织行上。与 HRIS 那边的群体离职刹车同一个道理。
"""
from __future__ import annotations

import logging
from datetime import UTC, datetime
from typing import Any

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

from app.audit.service import AuditService
from app.core.txn import best_effort
from app.identity.person_models import IdentityPerson

from . import constants as const
from . import service as svc
from .driver_access import resolve_org_driver
from .models import GithubOrg, GithubOrgMembership

logger = logging.getLogger(__name__)

#: 空列表刹车的触发阈值。台账里活着的记录数 ≥ 它、而 GitHub 说 0 个成员时
#: 拒绝对账。取 3 而不是 1:一个只有 1–2 人的组织真的清空是可能的,而
#: 三个人同时消失且 Helix 一无所知,更可能是我们读错了。
EMPTY_FETCH_BRAKE_MIN_LIVE = 3


def _now() -> datetime:
    return datetime.now(UTC)


def _naive(dt: datetime) -> datetime:
    return dt.replace(tzinfo=None) if dt.tzinfo is not None else dt


async def _resolve_person_ids(
    db: AsyncSession, logins: set[str],
) -> dict[str, IdentityPerson]:
    """按 handle 批量反查员工。

    只认 ``identity_persons.github_id``,**不**回退到「邮箱前缀 == handle」
    那种猜测:审计域里有一处正是这么猜的,而它猜错的代价在那边只是一条
    对不上的 MFA 记录,在这里却会把一条成员资格挂到错误的人头上 —— 然后
    那个人离职时,别人的访问权被撤了。
    """
    if not logins:
        return {}
    rows = list(
        (
            await db.execute(
                select(IdentityPerson).where(
                    IdentityPerson.deleted_at.is_(None),
                    IdentityPerson.github_id.is_not(None),
                )
            )
        ).scalars().all()
    )
    by_handle: dict[str, IdentityPerson] = {}
    for p in rows:
        key = (p.github_id or "").strip().lower()
        if not key or key not in logins:
            continue
        # 同一个 handle 登记在两个人身上是数据问题,不是这里要解决的 ——
        # 保留先出现的那个并 warning,而不是随机挑一个。
        if key in by_handle:
            logger.warning(
                "github_org: handle %s is registered on two people "
                "(%s and %s) — reconcile keeps the first",
                key, by_handle[key].work_email, p.work_email,
            )
            continue
        by_handle[key] = p
    return by_handle


async def _refresh_seat_usage(
    db: AsyncSession, org: GithubOrg, driver: Any, *,
    trace_id: str, now: datetime,
) -> None:
    """顺手刷一遍坐席用量。

    挂在对账里而不是单开一个 cron:它要的凭证、要遍历的组织清单、以及
    「组织配没配好」的判断,和对账**完全相同**。单开一个 job 只会得到
    第二份同样的解析逻辑和第二处会和这里说法不一致的地方。

    **拉不到不算对账失败。** 坐席是一条提醒,成员对齐是这套东西的本职;
    让一个 billing 权限问题把成员对账整趟打断,是把次要的事看得比主要的
    事重。拉不到就保留上一次的快照,``seat_checked_at`` 不动 —— 界面上
    「三天前的数」和「刚查过」看得出区别。
    """
    try:
        usage = await driver.get_seat_usage(org.slug)
    except NotImplementedError:
        return
    except Exception as e:  # noqa: BLE001
        logger.warning(
            "github_org: seat usage refresh failed for %s: %s", org.slug, e,
        )
        return
    org.seat_plan = usage.plan
    org.seat_filled = usage.filled_seats
    org.seat_total = usage.total_seats
    org.seat_checked_at = _naive(now)


async def reconcile_org(
    db: AsyncSession, org: GithubOrg, *, trace_id: str,
    apply_removals: bool = True,
) -> dict[str, Any]:
    """把一个组织的台账与 GitHub 上的真实成员对齐。

    ``apply_removals=False`` 用于**存量导入**那一趟:第一次导入时只该往
    台账里加,不该因为「Helix 什么都不知道」就去动任何东西。
    """
    summary: dict[str, Any] = {
        "org": org.slug, "github_members": 0, "github_pending": 0,
        "imported": 0, "linked": 0, "role_drift": 0,
        "removed_upstream": 0, "unlinked": 0, "leavers_still_in_org": 0,
    }
    resolution = await resolve_org_driver(db, org)
    if not resolution.automatic:
        summary["skipped"] = resolution.reason
        return summary

    driver = resolution.driver
    assert driver is not None  # noqa: S101
    audit = AuditService(db)

    members = await driver.list_members(org.slug)
    invitations = await driver.list_pending_invitations(org.slug)
    summary["github_members"] = len(members)
    summary["github_pending"] = len(invitations)

    live_count = int(
        (
            await db.execute(
                select(func.count())
                .select_from(GithubOrgMembership)
                .where(
                    GithubOrgMembership.org_id == org.id,
                    GithubOrgMembership.state.in_(const.MEMBERSHIP_LIVE_STATES),
                )
            )
        ).scalar_one()
    )

    # ── 空列表刹车 ────────────────────────────────────────────────
    if (
        apply_removals
        and not members
        and not invitations
        and live_count >= EMPTY_FETCH_BRAKE_MIN_LIVE
    ):
        why = (
            f"refused to reconcile {org.slug}: GitHub reported 0 members and "
            f"0 pending invitations while Helix tracks {live_count} live "
            f"memberships. Treating that as 'everybody left' would wipe the "
            f"ledger offboarding reads. Check the credential's org access."
        )
        logger.error("github_org: %s", why)
        org.last_reconcile_error = why[:512]
        await audit.log(
            trace_id=trace_id, actor_type="system",
            action="github_org.reconcile.refused_empty", status="failure",
            resource_type="github_org", resource_id=org.id,
            detail={"org": org.slug, "live_count": live_count},
        )
        summary["skipped"] = "empty_fetch_brake"
        return summary

    github_by_login = {m.login.lower(): m for m in members}
    pending_by_login = {
        i.login.lower(): i for i in invitations if i.login
    }
    people = await _resolve_person_ids(
        db, set(github_by_login) | set(pending_by_login),
    )

    ledger = list(
        (
            await db.execute(
                select(GithubOrgMembership).where(
                    GithubOrgMembership.org_id == org.id,
                )
            )
        ).scalars().all()
    )
    ledger_by_login = {m.github_login_lower: m for m in ledger}
    now = _now()

    # ── GitHub 有、Helix 没有 → 导入 ─────────────────────────────
    for login, member in github_by_login.items():
        row = ledger_by_login.get(login)
        person = people.get(login)
        if row is None:
            await svc.upsert_membership(
                db, org=org, github_login=member.login,
                person_id=(str(person.id) if person else None),
                state=const.MEMBERSHIP_STATE_ACTIVE,
                role=member.role,
                source=const.SOURCE_RECONCILE,
                actor_kind="integration", trace_id=trace_id,
                event_type=const.EVENT_RECONCILED,
                detail={"discovered_on_github": True},
            )
            summary["imported"] += 1
            continue

        # 已有行:状态 / 角色 / person 三样各自对齐。
        if row.state != const.MEMBERSHIP_STATE_ACTIVE:
            await svc.upsert_membership(
                db, org=org, github_login=row.github_login,
                person_id=row.person_id,
                state=const.MEMBERSHIP_STATE_ACTIVE,
                source=row.source, actor_kind="integration",
                trace_id=trace_id, event_type=const.EVENT_JOINED,
            )
        if member.role != row.role:
            await svc.upsert_membership(
                db, org=org, github_login=row.github_login,
                person_id=row.person_id, state=row.state,
                role=member.role, source=row.source,
                actor_kind="integration", trace_id=trace_id,
            )
            summary["role_drift"] += 1
        # 回填 person_id。注意它只能救「后来补登记了 github_id 的人」;
        # 存量那批认不出的靠这里救不回来(实测 217 个里 0 命中),
        # 那是治理面画像 + AI 建议要清的东西。
        if row.person_id is None and person is not None:
            await svc.upsert_membership(
                db, org=org, github_login=row.github_login,
                person_id=str(person.id), state=row.state,
                source=row.source, actor_kind="integration",
                trace_id=trace_id,
            )
            summary["linked"] += 1
        row.last_verified_at = _naive(now)

    # ── Helix 说在、GitHub 说不在 → 标已移除 ─────────────────────
    if apply_removals:
        for row in ledger:
            if row.state not in const.MEMBERSHIP_LIVE_STATES:
                continue
            login = row.github_login_lower
            if login in github_by_login:
                continue
            if login in pending_by_login:
                # 待接受的邀请仍然挂着 —— 那是 invitation_lifecycle 的地盘,
                # 这里不碰。
                continue
            if row.state == const.MEMBERSHIP_STATE_PENDING_INVITE:
                # 同上:pending 的推进(接受 / 过期 / 拒绝)由
                # invitation_lifecycle 按时间判,这里凭一次成员快照
                # 判不出是哪一种。
                continue
            if row.state == const.MEMBERSHIP_STATE_AWAITING_MANUAL:
                # 等人工去点的记录**本来就不该在 GitHub 上** —— 邀请还没
                # 发。把「不在 GitHub 上」读成「上游把他移除了」,等于把
                # 一件待办悄悄标成已完成,而那个人其实一直没拿到访问权。
                continue
            await svc.mark_removed(
                db, row,
                reason="removed on GitHub outside Helix",
                removal_source=const.REMOVAL_SOURCE_RECONCILE,
                actor_kind="integration", trace_id=trace_id,
            )
            summary["removed_upstream"] += 1

    summary["unlinked"] = int(
        (
            await db.execute(
                select(func.count())
                .select_from(GithubOrgMembership)
                .where(
                    GithubOrgMembership.org_id == org.id,
                    GithubOrgMembership.person_id.is_(None),
                    GithubOrgMembership.state.in_(const.MEMBERSHIP_LIVE_STATES),
                )
            )
        ).scalar_one()
    )

    # ── 已离职的人还在组织里 ──────────────────────────────────────
    #
    # 🔴 这是整套东西承诺的**反面验证**,不是一个可选的统计。对账把
    # GitHub 上的事实如实记下来是对的,但一条「已离职 + active」的记录
    # 如果只是安静地躺在台账里,那这套自动化就等于没做。
    #
    # 它有两条真实的产生路径,而**两条都绕开了离职流程**:
    #   * 有人在 GitHub 上手工把一个已离职的人加了回去;
    #   * 这条成员资格在这个人离职时 ``person_id`` 还是空的(认不出是谁),
    #     所以离职扫描根本看不到它 —— 后来对账回填了 person_id,人已经走了。
    #
    # 每一条都单独落一条 failure 审计:一个数字告诉你「有 3 个」,而
    # 逐条审计才告诉你「是谁、在哪个组织、什么 handle」,后者才是能
    # 动手的东西。
    leaver_rows = list(
        (
            await db.execute(
                select(GithubOrgMembership, IdentityPerson)
                .join(
                    IdentityPerson,
                    IdentityPerson.id == GithubOrgMembership.person_id,
                )
                .where(
                    GithubOrgMembership.org_id == org.id,
                    GithubOrgMembership.state.in_(const.MEMBERSHIP_LIVE_STATES),
                    IdentityPerson.employment_status.in_(
                        ("left", "leaver_pending"),
                    ),
                )
            )
        ).all()
    )
    # 豁免的组织(社区 org)不算 —— 那里的成员本来就允许离职后保留,
    # 把它算成缺口只会训练人忽略这个数。
    if org.offboarding_policy != const.OFFBOARDING_POLICY_EXEMPT:
        summary["leavers_still_in_org"] = len(leaver_rows)
        for membership, person in leaver_rows:
            await audit.log(
                trace_id=trace_id, actor_type="system",
                action="github_org.reconcile.leaver_still_member",
                status="failure",
                resource_type="github_org_membership",
                resource_id=membership.id,
                detail={
                    "org": org.slug,
                    "github_login": membership.github_login,
                    "person": person.work_email,
                    "employment_status": person.employment_status,
                    "membership_state": membership.state,
                    "membership_source": membership.source,
                },
            )
        if leaver_rows:
            logger.warning(
                "github_org: %d departed employee(s) still in %s: %s",
                len(leaver_rows), org.slug,
                ", ".join(m.github_login for m, _ in leaver_rows),
            )

    await _refresh_seat_usage(db, org, driver, trace_id=trace_id, now=now)

    org.last_reconciled_at = _naive(now)
    org.last_reconcile_error = None
    if (
        summary["imported"]
        or summary["removed_upstream"]
        or summary["leavers_still_in_org"]
    ):
        await audit.log(
            trace_id=trace_id, actor_type="system",
            action="github_org.reconcile", status="success",
            resource_type="github_org", resource_id=org.id,
            detail=summary,
        )
    return summary


async def reconcile_all(
    db: AsyncSession, *, trace_id: str, apply_removals: bool = True,
) -> dict[str, Any]:
    """对每个启用的组织跑一遍。单个失败不中断其余。"""
    orgs = await svc.list_orgs(db)
    totals: dict[str, Any] = {
        "orgs": len(orgs), "imported": 0, "linked": 0, "role_drift": 0,
        "removed_upstream": 0, "unlinked": 0, "leavers_still_in_org": 0,
        "failed_orgs": [],
        "skipped": {},
    }
    for org in orgs:
        # 每个组织一个 SAVEPOINT。没有它,「单个失败不中断其余」只对纯
        # Python 异常成立 —— 而真实的故障是 **flush 失败**:那之后 session
        # 被毒化,后面每个组织都撞 PendingRollbackError,连
        # ``last_reconcile_error`` 都提交不上去,整个 job 以一条与真因无关
        # 的错误收场,真因只留在日志里。
        #
        # 用 ``best_effort`` 而不是裸 ``begin_nested()``:锁等待 / 死锁
        # (1205/1213/1305)和连接断开(2006/2013)时事务在服务端就已经
        # 没了,savepoint 回滚救不回来,那两类必须**往上抛**让 cron 重跑,
        # 而不是被记成这个组织的「对账错误」然后继续对着一个死事务循环。
        # 它顺带把失败**归到正确的组织头上**:退出 savepoint 会 flush,
        # 一条延迟到下一个组织才炸的 INSERT 不会再冤枉别人。
        #
        # ``slug`` 先取出来:回滚 savepoint 会让这个组织上的属性过期,
        # 而在 async session 上碰一个过期属性是一次同步 IO —— MissingGreenlet,
        # 不是「再查一次」。要写它得先 ``await db.refresh``。
        slug = org.slug
        out: dict[str, Any] | None = None
        async with best_effort(
            db, label=f"member reconcile {slug}",
            log=logger, log_level=logging.DEBUG,
        ) as step:
            out = await reconcile_org(
                db, org, trace_id=trace_id,
                apply_removals=apply_removals,
            )
        # ``step.ok``,不是 ``out is None``:body 正常返回了,失败发生在
        # 退出 savepoint 时的那次 flush —— 变量早就被赋上值了。
        if not step.ok or out is None:
            # ERROR + 栈:cron 的失败只有日志这一个现场。
            logger.error(
                "github_org: reconcile failed for %s: %s",
                slug, step.reason(), exc_info=step.error,
            )
            await db.refresh(org)
            org.last_reconcile_error = step.reason()
            totals["failed_orgs"].append(slug)
            continue
        if out.get("skipped"):
            totals["skipped"][slug] = out["skipped"]
            continue
        for key in ("imported", "linked", "role_drift", "removed_upstream"):
            totals[key] += out.get(key, 0)
        totals["unlinked"] += out.get("unlinked", 0)
        totals["leavers_still_in_org"] += out.get("leavers_still_in_org", 0)

    # 对账刚把台账刷新过一遍,紧接着把**已启用**的标签规则落一次 ——
    # 标签是从台账派生的,同一个节奏跑最省事,也不用再解释一个新 cron。
    #
    # 没有任何规则被启用时它是空转(默认就是全部关着),所以这一步不会
    # 在管理员没点过同意之前改动任何数据。计数进 summary,好让「昨天
    # 谁给这些人打了标签」在任务详情页上答得出来。
    #
    # SAVEPOINT,和上面每个组织那圈同理:``apply_enabled_rules`` 会经
    # ``attach()`` 往 ``github_org_member_labels`` 写。裸 ``except`` 只挡得住
    # 纯 Python 异常 —— 一次失败的 flush 会让整个事务失活,那之后连
    # ``totals`` 都提交不上去,一次**成功**的离职回收对账会以一条打标签的
    # 错误收场。而这两件事的严重性差一个数量级,正是上面那句注释想说的。
    from . import labels as _labels  # noqa: PLC0415

    label_summary: dict[str, Any] | None = None
    async with best_effort(
        db, label="github_org auto label apply",
        log=logger, log_level=logging.DEBUG,
    ) as step:
        label_summary = await _labels.apply_enabled_rules(db, trace_id=trace_id)
    if not step.ok or label_summary is None:
        logger.warning(
            "github_org: auto label apply failed: %s", step.reason(),
            exc_info=step.error,
        )
        totals["labels_auto_applied"] = {"error": step.reason()}
    else:
        totals["labels_auto_applied"] = label_summary
    return totals


async def import_existing_members(
    db: AsyncSession, *, trace_id: str,
) -> dict[str, Any]:
    """存量导入:把 GitHub 上现有的成员一次性拉进台账。

    与例行对账**只差一个开关**(``apply_removals=False``),刻意如此:
    第一次导入时 Helix 什么都不知道,任何「Helix 说在、GitHub 说不在」
    的判断都是无意义的 —— 那时台账本来就是空的。用同一份实现,是为了
    不让导入路径长成第二套语义。

    幂等:重复跑只会把新出现的人补进去。
    """
    return await reconcile_all(db, trace_id=trace_id, apply_removals=False)