"""「申请加入 GitHub 组织」的提交、审批回调与执行。

形状与 ``vpn_application`` / ``sso_grant`` 对齐 —— 提交落 pending、飞书
审批回调翻状态、通过后执行 —— 但有两处这个域独有的东西:

**1. 两个来源共用一个执行分支。**

飞书那边的表单入口不会在上线当天就消失,而且已经有在途的单子。所以
`github_org_join_request` 这条流同时开着 ``accept_external_origin``:

* **Helix 发起**:本地有一行 ``github_org_join_requests``,handle 与组织
  都是提交时校验过的。
* **飞书直接提交**:没有本地行,父行由 ``_synthesize_external_origin_request``
  合成。执行时从表单里现场解析 handle 与组织。

两条路最终都落到 :func:`execute_approved` 上,所以「审批通过之后发生
什么」只有一份实现。少了外来那一半,过渡期里从飞书提交的人会**静默地**
拿不到自动化 —— 而这正是本来要修的那个问题。

**2. 执行失败不等于审批失败。**

驱动可能没配(转人工)、GitHub 可能报 handle 不存在、也可能只是网络抖。
这三种的处置完全不同,所以它们落在不同的状态与不同的通知上,而不是
一律记成「失败」。特别地:**没配凭证不是失败** —— 申请单照常记成
approved,台账落一行 pending_invite 的人工待办,通知发给 GitHub 管理员
去手工点邀请。这条路径让整套东西在凭证到位之前就能用。
"""
from __future__ import annotations

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

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

from app.approval.models import ApprovalRequest
from app.audit.service import AuditService
from app.core.exceptions import AppException, NotFoundError
from app.identity.person_models import IdentityPerson

from . import constants as const
from . import service as svc
from .models import GithubOrg, GithubOrgJoinRequest
from .validators import bounded_text, normalize_github_login

logger = logging.getLogger(__name__)

FEISHU_FLOW_KEY = "github_org_join_request"

#: 提交时的说明上限。超了就拒,不静默截断 —— 一段 500 字的理由在审批人
#: 眼里变成半句话,而提交的人以为写全了。
_REASON_MAX = 1000


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


# ════════════════════════════════════════════════════════════════════
# 飞书表单载荷
# ════════════════════════════════════════════════════════════════════


def _widget(schema: dict[str, Any], field: str) -> dict[str, Any]:
    spec = (schema or {}).get(field)
    if isinstance(spec, str):
        return {"id": spec, "type": "input"}
    if isinstance(spec, dict) and spec.get("id"):
        return dict(spec)
    raise AppException(
        code="GITHUB_FLOW_SCHEMA_INCOMPLETE",
        message=(
            f"The Feishu flow mapping for '{FEISHU_FLOW_KEY}' is missing "
            f"widget '{field}'. Fix it under 集成管理 → 审批流程映射."
        ),
        status_code=503,
    )


def build_feishu_form(
    *,
    payload_schema: dict[str, Any],
    org: GithubOrg,
    github_login: str,
    reason: str | None,
) -> str:
    """把一张申请单渲染成飞书表单载荷。

    只填 Helix 入口收集的那几个控件。**社区成员固定填「否」,不提供
    社区 team 选择** —— 社区 team 的加入由社区自己的流程决定,Helix 替
    申请人替社区做主是越权。员工要申请 team,仍然走飞书那张表单,而那
    条路由 external-origin 通道接住。
    """
    if not org.feishu_option_value:
        raise AppException(
            code="GITHUB_ORG_NOT_REQUESTABLE",
            message=(
                f"Organization '{org.slug}' has no Feishu form option value, "
                f"so Helix cannot raise an approval for it. Set it under "
                f"GitHub 组织 → 组织登记."
            ),
            status_code=409,
        )

    login_w = _widget(payload_schema, "github_login")
    ack_w = _widget(payload_schema, "lark_profile_ack")
    notice_w = _widget(payload_schema, "notice")
    org_w = _widget(payload_schema, "org")
    community_w = _widget(payload_schema, "is_community_member")
    reason_w = _widget(payload_schema, "reason")

    community_no = (
        (community_w.get("choice_values") or {}).get("no")
    )
    if not community_no:
        raise AppException(
            code="GITHUB_FLOW_SCHEMA_INCOMPLETE",
            message=(
                "The Feishu flow mapping is missing "
                "is_community_member.choice_values.no."
            ),
            status_code=503,
        )

    form = [
        {"id": login_w["id"], "type": login_w.get("type", "input"),
         "value": github_login},
        # 「已设置」—— 走 Helix 入口意味着 handle 就在这张单子上。
        {"id": ack_w["id"], "type": ack_w.get("type", "radioV2"),
         "value": ack_w.get("value", "")},
        # 只读说明控件,飞书标成必填,原样回填。
        {"id": notice_w["id"], "type": notice_w.get("type", "text"),
         "value": notice_w.get("value", "")},
        {"id": org_w["id"], "type": org_w.get("type", "radioV2"),
         "value": org.feishu_option_value},
        {"id": community_w["id"], "type": community_w.get("type", "radioV2"),
         "value": community_no},
        {"id": reason_w["id"], "type": reason_w.get("type", "textarea"),
         "value": reason or ""},
    ]
    return json.dumps(form, ensure_ascii=False)


def extract_form_values(
    instance_detail: dict[str, Any], payload_schema: dict[str, Any] | None,
) -> dict[str, Any]:
    """从飞书实例详情里读回表单值,按 widget id 定位。

    ⚠️ ``instance_detail["form"]`` 是一个 **JSON 字符串**,不是 list。
    平台为此吃过一次 P0(每个 APPROVED 回调 500,飞书重投耗尽后单子卡在
    pending),所以这里防御性解码。

    按 widget id 而不是按控件名匹配:控件名是可以被人在飞书后台改的
    (而且这张表单的中英文名在不同实例上就不一致 —— 生产实例里同一个
    控件叫 "GitHub ORG",定义里叫 "GitHub Organization")。
    """
    raw = instance_detail.get("form")
    if isinstance(raw, str):
        try:
            items = json.loads(raw or "[]")
        except (ValueError, TypeError):
            logger.warning("github_org: unparseable Feishu form payload")
            items = []
    elif isinstance(raw, list):
        items = raw
    else:
        items = []

    by_id = {
        str(it.get("id")): it for it in items if isinstance(it, dict) and it.get("id")
    }
    out: dict[str, Any] = {}
    for field in (
        "github_login", "org", "is_community_member", "community_team", "reason",
    ):
        spec = (payload_schema or {}).get(field)
        wid = spec.get("id") if isinstance(spec, dict) else spec
        if not wid:
            continue
        item = by_id.get(str(wid))
        if item is not None:
            out[field] = item.get("value")
    # 组织与社区标记在实例详情里已经被飞书翻译成了**显示文本**
    # (例如 "github.com/tidbcloud" / "No"),不是选项内部标识 ——
    # 生产实例实测如此。所以解析组织时按显示文本反查,见 resolve_org_from_form。
    return out


async def resolve_org_from_form(
    db: AsyncSession, raw_value: Any,
) -> GithubOrg | None:
    """把表单里的组织值解析成登记表里的一行。

    飞书实例详情返回的是**显示文本**(``github.com/tidbcloud``),而定义
    里的选项值是内部标识(``legl5z3l-yp7yya2lv4-1``)。两种都试:先按
    选项值精确匹配,再按显示文本里的 slug 匹配 —— 过渡期里两种形态都
    可能出现,而认不出组织意味着这张已批准的单子无处落地。
    """
    value = str(raw_value or "").strip()
    if not value:
        return None
    orgs = await svc.list_orgs(db, include_inactive=True)
    for org in orgs:
        if org.feishu_option_value and org.feishu_option_value == value:
            return org
    tail = value.rstrip("/").rsplit("/", 1)[-1].strip().lower()
    for org in orgs:
        if org.slug_lower == tail:
            return org
    return None


# ════════════════════════════════════════════════════════════════════
# 提交
# ════════════════════════════════════════════════════════════════════


def resolve_join_handle(
    requester: IdentityPerson, asked: str | None,
) -> str:
    """加入组织时用哪个 GitHub handle —— 只能是**已登记**的那一个。

    这个字段曾经是自由文本(入口预填 HRIS 值,但可以改)。那意味着员工可以
    给**任意一个 GitHub 账号**申请进组织:提交的字符串一路流到
    ``driver.invite_member(org, github_login)``,而 org owner 在飞书上看到的
    只是「申请人姓名 + 一个 handle」—— 他手里没有权威值可对照,「冯帅申请把
    someone-else 加进来」看起来并不明显是错的。

    当时收不紧:``identity_persons.github_id`` 只在入职采集、之后没有任何更新
    入口,417 人里 154 人是空的,强制它等于让这些人永远申请不了。迁移 579 的
    自助修改链路补上了那个入口(员工提交 → 本人 HR Ops 确认 → 回写 HRIS /
    飞书),所以现在有一个**被确认过**的值可以强制。

    规则:以 ``requester.github_id`` 为准。调用方仍然可以传 ``asked``,但只
    允许与登记值一致(大小写不敏感 —— GitHub 用户名本来就不区分大小写)。传
    了不一致的值是**拒绝**而不是静默忽略:静默忽略同样防住了越权,但调用方会
    以为自己指定的那个生效了,一个 MCP 客户端会照着「已提交」继续往下说。

    抽成模块级函数而不是留在 ``submit`` 里,是为了让这条规则能被直接测 ——
    在 ``submit`` 里它藏在飞书建实例、审批流查找这些外部依赖后面,测试环境
    走不到那儿,断言只能写成一个会自己早退的形状(第一版就是,变异验证时它在
    自由文本下照样通过)。
    """
    registered = (requester.github_id or "").strip()
    if not registered:
        raise AppException(
            code="GITHUB_HANDLE_NOT_ON_FILE",
            message=(
                "Your GitHub username is not on file yet. Register it first "
                "(个人设置 · 基本信息, or 「我的申请 → 创建申请 → 修改 GitHub "
                "ID」) — your HR Ops confirms it, and then you can request "
                "organization access."
            ),
            status_code=409,
        )
    login = normalize_github_login(registered, field="github_id")
    if asked is not None and str(asked).strip():
        want = normalize_github_login(asked)
        if want.lower() != login.lower():
            raise AppException(
                code="GITHUB_HANDLE_MISMATCH",
                message=(
                    f"Organization access is always requested for the GitHub "
                    f"username on file ('{login}'), not an arbitrary one. To "
                    f"use '{want}', change your GitHub ID first — that goes to "
                    f"your HR Ops for confirmation."
                ),
                status_code=409,
            )
    return login


class GithubOrgRequestService:
    def __init__(self, db: AsyncSession) -> None:
        self.db = db
        self.audit = AuditService(db)
        self._feishu_client: Any | None = None

    async def _client(self) -> Any:
        if self._feishu_client is None:
            from app.approval.feishu_client import (  # noqa: PLC0415
                FeishuApprovalClient,
            )
            from app.integrations.compat import (  # noqa: PLC0415
                get_feishu_config,
            )

            cfg = await get_feishu_config(session=self.db)
            self._feishu_client = FeishuApprovalClient(
                app_id=str(cfg.get("app_id") or ""),
                app_secret=str(cfg.get("app_secret") or ""),
            )
        return self._feishu_client

    # ── 提交 ──────────────────────────────────────────────────────

    async def submit(
        self,
        *,
        requester: IdentityPerson,
        org_id: str,
        github_login: str | None,
        reason: str | None,
        trace_id: str,
    ) -> GithubOrgJoinRequest:
        """提交一张申请单并在飞书上建实例。

        ``github_login`` 是**可选的一致性检查**,不是数据来源:真正用的是
        ``requester.github_id``(已登记、经 HR Ops 确认)。传了不一致的值会
        被拒 —— 见下面那段注释。

        服务层自带离职闸:路由上的 ``_leaver_runtime_gate`` 只挡 HTTP,
        而 MCP 工具(PAT 鉴权的 agent)是直接调服务的 —— 少了这道,一个
        待离职的人拿着有效 PAT 就能绕过整个离职锁定。
        """
        from app.core.grant_eligibility import (  # noqa: PLC0415
            is_blocked_from_new_grant,
        )

        if is_blocked_from_new_grant(requester):
            # 比硬编码 employment_status 更全:它还认软删与 HRIS
            # ``purged`` 墓碑,而那两种状态的人同样不该拿到新访问权。
            raise AppException(
                code="GITHUB_REQUESTER_INELIGIBLE",
                message=(
                    "Departing or departed employees cannot request new "
                    "GitHub organization access."
                ),
                status_code=403,
            )

        clean_reason = bounded_text(reason, _REASON_MAX, "reason")

        org = await svc.get_org(self.db, org_id)
        if not org.is_active:
            raise AppException(
                code="GITHUB_ORG_INACTIVE",
                message=f"Organization '{org.slug}' is retired.",
                status_code=409,
            )
        if not org.request_enabled:
            raise AppException(
                code="GITHUB_ORG_NOT_REQUESTABLE",
                message=(
                    f"Organization '{org.slug}' is not open for self-service "
                    f"requests."
                ),
                status_code=409,
            )

        # 已经在里面了就别再走一遍审批 —— 让 org owner 去批一个已经成立的
        # 事实,是在消耗那条流程唯一的稀缺资源(审批人的注意力)。
        #
        # 按**人**查,不按 handle 查:一个换过 GitHub 账号的人,按新 handle
        # 查不到他用旧 handle 建立的成员资格,于是会带着一张注定重复的
        # 申请去占用审批人的时间。错误信息里点名旧 handle,这样他一眼就
        # 知道为什么被挡,而不是以为系统坏了。
        existing_member = await svc.find_person_membership_in_org(
            self.db, org_id=org.id, person_id=str(requester.id),
        )
        if existing_member is not None:
            verb = (
                "a member of"
                if existing_member.state == const.MEMBERSHIP_STATE_ACTIVE
                else "invited to"
            )
            # 直接读登记值,不走 ``resolve_join_handle`` —— 这一段要报的是
            # 「你已经在里面了」,而那个函数对没登记 handle 的人会抛。一个
            # 已经是成员、但还没登记 handle 的人,应该看到「你已经在里面
            # 了」,不是「你还没登记 handle」。
            registered_lower = (requester.github_id or "").strip().lower()
            as_handle = (
                ""
                if existing_member.github_login_lower == registered_lower
                else f" (as '{existing_member.github_login}')"
            )
            raise AppException(
                code="GITHUB_ALREADY_MEMBER",
                message=(
                    f"You are already {verb} {org.slug}{as_handle}."
                ),
                status_code=409,
            )

        # handle 闸放在这里,不放在方法开头:组织已下线、你已经在组织里了
        # —— 这些和 handle 无关的情况应该报它们自己的原因。放在最前面会让
        # 一个「已经是成员」的人收到「你还没登记 handle」,而那句话既不对
        # 也没告诉他真正发生了什么。(既有测试抓到的。)
        login = resolve_join_handle(requester, github_login)

        from app.core.distributed_lock import (  # noqa: PLC0415
            LockBusy,
            distributed_lock,
        )

        lock_key = f"github_org_submit:{requester.id}"
        try:
            async with distributed_lock(lock_key, ttl=120):
                # 锁内复查:两次并发提交(双击 / agent 重试)否则会各自
                # 建一个飞书实例,把审批人淹掉。
                dupe = await self.db.scalar(
                    select(GithubOrgJoinRequest.id)
                    .where(
                        GithubOrgJoinRequest.person_id == str(requester.id),
                        GithubOrgJoinRequest.org_id == org.id,
                        GithubOrgJoinRequest.status.in_(
                            const.JOIN_REQUEST_OPEN_STATUSES,
                        ),
                    )
                    .limit(1)
                    .with_for_update()
                )
                if dupe:
                    raise AppException(
                        code="GITHUB_REQUEST_ALREADY_PENDING",
                        message=(
                            f"You already have a pending request for "
                            f"{org.slug}. Cancel it first or wait for it to "
                            f"be processed."
                        ),
                        status_code=409,
                    )
                return await self._submit_locked(
                    requester=requester, org=org, login=login,
                    reason=clean_reason, trace_id=trace_id,
                )
        except LockBusy as exc:
            raise AppException(
                code="GITHUB_SUBMIT_IN_FLIGHT",
                message=(
                    "Another submission of yours is being processed. "
                    "Please retry in a few seconds."
                ),
                status_code=409,
            ) from exc

    async def _submit_locked(
        self,
        *,
        requester: IdentityPerson,
        org: GithubOrg,
        login: str,
        reason: str | None,
        trace_id: str,
    ) -> GithubOrgJoinRequest:
        from app.approval.submission_intents import (  # noqa: PLC0415
            submit_with_durable_intent,
        )
        from app.integrations.service import get_approval_flow  # noqa: PLC0415

        flow = await get_approval_flow(FEISHU_FLOW_KEY, session=self.db)
        client = await self._client()
        form_payload = build_feishu_form(
            payload_schema=flow.payload_schema or {},
            org=org,
            github_login=login,
            reason=reason,
        )
        feishu_user_id = await client.resolve_user_id_by_email(
            requester.work_email,
        )

        request_id = str(uuid.uuid4())
        approval_request_id = str(uuid.uuid4())
        intent = await submit_with_durable_intent(
            self.db,
            client=client,
            scope=FEISHU_FLOW_KEY,
            applicant_id=requester.id,
            semantic_payload={
                "org_id": org.id, "github_login": login, "reason": reason,
            },
            provider_payload={
                "approval_code": flow.external_code,
                "identity_type": "user_id",
                "identity_value": feishu_user_id,
                "form": form_payload,
            },
            finalization_payload={
                "request_id": request_id,
                "approval_request_id": approval_request_id,
                "approval_code": flow.external_code,
                "org_id": org.id,
                "org_slug": org.slug,
                "github_login": login,
                "reason": reason,
            },
            trace_id=trace_id,
            # 同一个人可以同时申请多个组织 —— 用组织把未完成的意图分区,
            # 否则第二个组织的提交会去恢复第一个组织那张未完成的单子。
            recovery_key=f"org:{org.id}",
        )
        return await self._finalize(intent.id, recovered=intent.attempt_count > 1)

    async def _finalize(
        self, intent_id: uuid.UUID, *, recovered: bool,
    ) -> GithubOrgJoinRequest:
        from app.approval.models import (  # noqa: PLC0415
            ApprovalSubmissionIntent,
        )
        from app.approval.submission_intents import (  # noqa: PLC0415
            compact_terminal_intent_payload,
            terminal_intent_resource_id,
        )

        intent = await self.db.scalar(
            select(ApprovalSubmissionIntent)
            .where(ApprovalSubmissionIntent.id == intent_id)
            .with_for_update()
            .execution_options(populate_existing=True)
        )
        if intent is None or intent.scope != FEISHU_FLOW_KEY:
            raise RuntimeError("GitHub org submission intent disappeared")
        if intent.state == "completed":
            rid = terminal_intent_resource_id(
                intent, resource_type="github_org_join_request",
            )
            existing = await self.db.get(GithubOrgJoinRequest, rid)
            if existing is None:
                raise RuntimeError("completed GitHub intent has no row")
            return existing

        payload = dict((intent.payload or {}).get("finalization") or {})
        request_id = str(payload["request_id"])
        if intent.state != "provider_created" or not intent.external_id:
            raise RuntimeError(
                f"cannot finalize GitHub org intent in {intent.state!r}",
            )

        row = await self.db.get(GithubOrgJoinRequest, request_id)
        if row is None:
            row = GithubOrgJoinRequest(
                id=request_id,
                person_id=str(intent.applicant_id),
                org_id=str(payload["org_id"]),
                github_login=str(payload["github_login"]),
                is_community_member=False,
                reason=payload.get("reason"),
                status=const.JOIN_REQUEST_STATUS_PENDING,
                feishu_instance_code=intent.external_id,
            )
            self.db.add(row)
            await self.db.flush()
            approval_request = ApprovalRequest(
                id=uuid.UUID(str(payload["approval_request_id"])),
                external_id=intent.external_id,
                approval_type=FEISHU_FLOW_KEY,
                applicant_id=intent.applicant_id,
                status=const.JOIN_REQUEST_STATUS_PENDING,
                source_payload={
                    "github_org_join_request_id": request_id,
                    "approval_code": payload["approval_code"],
                    "org_id": payload["org_id"],
                    "org_slug": payload.get("org_slug"),
                    "github_login": payload["github_login"],
                },
            )
            self.db.add(approval_request)
            await self.db.flush()
            row.approval_request_id = str(approval_request.id)
            await self.db.flush()

        intent.approval_request_id = uuid.UUID(
            str(payload["approval_request_id"]),
        )
        intent.state = "completed"
        intent.completed_at = _now()
        intent.last_error = None
        await self.audit.log(
            trace_id=(
                f"approval-intent-reconcile:{intent.id}"
                if recovered
                else str((intent.payload or {}).get("trace_id") or intent.id)
            ),
            actor_type="system" if recovered else "user",
            actor_user_id=None if recovered else intent.applicant_id,
            action=(
                "approval.submission_intent.reconciled"
                if recovered
                else "github_org.request.submit"
            ),
            status="success",
            resource_type="github_org_join_request",
            resource_id=request_id,
            detail={
                "intent_id": str(intent.id),
                "feishu_instance_code": intent.external_id,
                "org_slug": payload.get("org_slug"),
                "github_login": payload["github_login"],
                "recovered": recovered,
            },
        )
        compact_terminal_intent_payload(
            intent,
            resource_type="github_org_join_request",
            resource_id=request_id,
        )
        await self.db.commit()
        await self.db.refresh(row)
        return row

    # ── 撤销 ──────────────────────────────────────────────────────

    async def cancel(
        self, *, request_id: str, actor: IdentityPerson, trace_id: str,
    ) -> GithubOrgJoinRequest:
        row = await self.db.get(GithubOrgJoinRequest, str(request_id))
        if row is None:
            raise NotFoundError("GithubOrgJoinRequest", str(request_id))
        if row.person_id != str(actor.id):
            # 与「查无此单」同一个回答 —— 不做存在性预言机。
            raise NotFoundError("GithubOrgJoinRequest", str(request_id))
        if row.status != const.JOIN_REQUEST_STATUS_PENDING:
            raise AppException(
                code="GITHUB_REQUEST_NOT_PENDING",
                message=f"Request is already {row.status}.",
                status_code=409,
            )
        row.status = const.JOIN_REQUEST_STATUS_CANCELLED
        row.decided_at = _now()

        # 飞书侧一并撤 —— 尽力而为。撤不掉不该让本地状态回滚:本地已经
        # 不认这张单子了,而飞书上留着一张会被 org owner 看到的孤儿单
        # 比状态不一致更糟的是**两边都还开着**。
        synced = True
        if row.feishu_instance_code:
            try:
                client = await self._client()
                await client.cancel_instance(row.feishu_instance_code)
            except Exception as e:  # noqa: BLE001
                synced = False
                logger.warning(
                    "github_org: Feishu cancel failed for %s: %s",
                    row.feishu_instance_code, e,
                )
        await self.audit.log(
            trace_id=trace_id, actor_type="user", actor_user_id=actor.id,
            action="github_org.request.cancel", status="success",
            resource_type="github_org_join_request", resource_id=row.id,
            detail={"feishu_cancel_synced": synced},
        )
        await self.db.commit()
        await self.db.refresh(row)
        self.last_feishu_cancel_synced = synced
        return row

    last_feishu_cancel_synced: bool = True