"""API 路由 — 工具链事件中枢（Toolchain Event Hub）

接收 Gitea Webhook，翻译成 Mail 通知推送给 Agent。

端点: POST /webhook/gitea
支持事件: pull_request, pull_request_review, issues, issue_comment
"""

from __future__ import annotations

import asyncio
import hashlib
import hmac
import json
import logging
import os
import re
import time
from datetime import datetime
from pathlib import Path, PurePath
from typing import Any, Dict, List, Optional, Set, Tuple

import httpx
from fastapi import APIRouter, Header, Request, Response

from src.blackboard.db import init_db
from src.blackboard.models import Task
from src.blackboard.operations import Blackboard
from src.config.agents import AGENT_IDS
from src.daemon.toolchain_templates import render_template
from src.utils import get_data_root

logger = logging.getLogger(__name__)

router = APIRouter(tags=["toolchain"])

# ---------------------------------------------------------------------------
# 幂等检查：内存 set，保留最近 7 天
# ---------------------------------------------------------------------------
# 使用内存 set 而非 SQLite（设计文档原计划 SQLite，简化实现：daemon 重启不频繁，
# 重启后丢失可接受，Webhook 重试窗口内不会重复）

_delivery_cache: Set[str] = set()
_delivery_timestamps: List[Tuple[float, str]] = []
_TTL_SECONDS = 7 * 24 * 3600
_idempotency_lock = asyncio.Lock()


def _is_duplicate(event: str, delivery: str,
                  payload: Optional[Dict[str, Any]] = None) -> bool:
    """检查 Webhook 是否重复投递，自动清理过期条目。

    双重去重策略：
    1. delivery UUID 去重（标准幂等）
    2. payload 内容去重（应对 Gitea v1.23.4 的 webhookNotifier + actionsNotifier
       对同一 review 生成不同 UUID 的双投递问题）
    """
    now = time.time()
    # 清理过期条目
    while _delivery_timestamps and (
            now - _delivery_timestamps[0][0]) > _TTL_SECONDS:
        _, key = _delivery_timestamps.pop(0)
        _delivery_cache.discard(key)

    # 检查 delivery UUID 去重
    key = f"{event}-{delivery}"
    if key in _delivery_cache:
        return True

    # 检查 payload 内容去重（review 事件：同一 PR + 同一用户 + 同一内容）
    # 注意：Gitea webhookNotifier 用 review.body，actionsNotifier 用 review.content
    # 所以去重 key 需要同时取两个字段，确保两种格式生成相同 key
    if payload and "review" in event:
        pr_num = payload.get("pull_request", {}).get("number")
        sender = payload.get("sender", {}).get("login")
        review = payload.get("review", {})
        # 取 body 或 content，优先 body（webhookNotifier 格式）
        content = review.get("body", "") or review.get("content", "")
        content_hash = hashlib.sha256(content.encode()).hexdigest()[:16]
        review_id = review.get("id", "")
        content_key = f"content:{event}:{pr_num}:{sender}:{review_id}:{content_hash}"
        if content_key in _delivery_cache:
            logger.info(
                "Content-based duplicate detected: %s PR#%s by %s",
                event,
                pr_num,
                sender)
            return True
        _delivery_cache.add(content_key)
        _delivery_timestamps.append((now, content_key))

    _delivery_cache.add(key)
    _delivery_timestamps.append((now, key))
    return False


# ---------------------------------------------------------------------------
# 签名验证
# ---------------------------------------------------------------------------

_WEBHOOK_SECRET: Optional[str] = os.environ.get("GITEA_WEBHOOK_SECRET")


def _verify_signature(body: bytes, signature: Optional[str]) -> bool:
    """验证 HMAC-SHA256 签名。secret 为空时跳过验签。"""
    if not _WEBHOOK_SECRET:
        return True
    if not signature:
        return False
    expected = hmac.new(
        _WEBHOOK_SECRET.encode(), body, hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature)


# ---------------------------------------------------------------------------
# Gitea API 调用
# ---------------------------------------------------------------------------

_GITEA_TOKEN: str = os.environ.get("GITEA_TOKEN", "")
_GITEA_BASE = "http://192.168.2.154:3000/api/v1"


async def _fetch_pr_files(repo: str, pr_number: int) -> Tuple[List[str], str]:
    """获取 PR 文件列表，含重试机制。

    Returns:
        (文件列表, 错误信息) — 成功时错误信息为空字符串
    """
    if not _GITEA_TOKEN:
        return [], "GITEA_TOKEN 未配置"

    url = f"{_GITEA_BASE}/repos/{repo}/pulls/{pr_number}/files"
    headers = {"Authorization": f"token {_GITEA_TOKEN}"}
    last_error = ""
    for attempt in range(3):
        try:
            async with httpx.AsyncClient(timeout=5.0) as client:
                resp = await client.get(url, headers=headers)
                resp.raise_for_status()
                files: List[Dict[str, Any]] = resp.json()
                return [f.get("filename", "") for f in files], ""
        except Exception as e:
            last_error = str(e)
            if attempt < 2:
                await asyncio.sleep(0.5 * (attempt + 1))
                logger.warning(
                    "Retry %d/3 fetching PR files: %s/pulls/%d",
                    attempt + 1,
                    repo,
                    pr_number)
    logger.warning(
        "Failed to fetch PR files after 3 retries: %s/pulls/%d - %s",
        repo,
        pr_number,
        last_error)
    return [], f"获取文件列表失败（重试3次）: {last_error}"


# ---------------------------------------------------------------------------
# 风险级别判定
# ---------------------------------------------------------------------------

_HIGH_PATTERNS = [
    "**/spawner*", "**/ticker*", "**/dispatcher*",
    "**/router*", "**/guardrails*", "**/strategy*", "**/risk*",
]


def _calc_risk_level(changed_files: List[str]) -> str:
    """根据改动文件列表判定风险级别。"""
    for filepath in changed_files:
        for pattern in _HIGH_PATTERNS:
            if PurePath(filepath).match(pattern):
                return "high"
    return "standard"


# ---------------------------------------------------------------------------
# Mail 创建
# ---------------------------------------------------------------------------


MAIL_PROJECT_ID = "_mail"


def _mail_db_path() -> Path:
    """获取 Mail 数据库路径，确保目录存在。"""
    root = get_data_root()
    db = root / MAIL_PROJECT_ID / "blackboard.db"
    db.parent.mkdir(parents=True, exist_ok=True)
    init_db(db)
    return db


def _send_mail(
    to_agent: str,
    title: str,
    description: str,
    source: str = "webhook",
) -> str:
    """创建 Mail Task 并写入数据库。

    Args:
        to_agent: 收件人 Agent ID
        title: 邮件标题
        description: 邮件正文
        source: 来源标识

    Returns:
        创建的 Mail ID

    Raises:
        Exception: 数据库写入失败
    """
    if to_agent not in AGENT_IDS:
        logger.warning("Unknown agent: %s, skipping mail", to_agent)
        return ""

    mail_id = f"mail-{int(datetime.now().timestamp() * 1000)}"
    notify_meta = {
        "type": "inform",
        "performative": "inform",
        "is_read": False,
        "conversation_id": f"conv-{mail_id}",
        "from": "system",
        "source": source,
    }
    task = Task(
        id=mail_id,
        title=title,
        description=description,
        assignee=to_agent,
        assigned_by="system",
        must_haves=json.dumps(notify_meta, ensure_ascii=False),
        task_type="mail",
        status="pending",
    )
    bb = Blackboard(_mail_db_path())
    bb.create_task(task)
    logger.info("Mail sent: %s → %s [%s]", title[:40], to_agent, mail_id)
    return mail_id


# ---------------------------------------------------------------------------
# 辅助：从 payload 提取仓库全名
# ---------------------------------------------------------------------------


def _repo_fullname(payload: Dict[str, Any]) -> str:
    """从 Webhook payload 提取仓库全名（owner/repo）。"""
    repo = payload.get("repository") or {}
    return repo.get("full_name", "")


# ---------------------------------------------------------------------------
# 事件处理函数
# ---------------------------------------------------------------------------


async def _handle_pull_request(payload: Dict[str, Any]) -> None:
    """处理 pull_request 事件：opened → 通知 reviewer；closed → merge 通知。"""
    action = payload.get("action", "")
    if action == "opened":
        await _handle_pr_opened(payload)
    elif action == "closed":
        await _handle_pr_closed(payload)


async def _handle_pr_opened(payload: Dict[str, Any]) -> None:
    """PR opened → 通知 simayi-challenger。"""
    pr = payload.get("pull_request")
    if not pr or not isinstance(pr, dict):
        logger.warning(
            "pull_request event missing pull_request field, skipping")
        return
    repo = _repo_fullname(payload)
    pr_number = pr.get("number", 0)
    pr_title = pr.get("title", "")
    pr_author = pr.get("user", {}).get("login", "unknown")
    branch = pr.get("head", {}).get("ref", "unknown")

    # 获取改动文件列表
    changed_files, fetch_error = await _fetch_pr_files(repo, pr_number)
    risk_level = _calc_risk_level(changed_files)
    if fetch_error:
        file_list = f"⚠️ {fetch_error}"
    else:
        file_list = "\n".join(
            f"- {f}" for f in changed_files) if changed_files else "(无文件变更)"

    text = render_template("review_request", {
        "repo": repo,
        "pr_number": str(pr_number),
        "pr_title": pr_title,
        "pr_author": pr_author,
        "branch": branch,
        "risk_level": risk_level,
        "file_list": file_list,
    })

    title = f"Review 请求: {pr_title} ({repo}#{pr_number})"
    _send_mail("simayi-challenger", title, text)


async def _handle_pull_request_review(payload: Dict[str, Any]) -> None:
    """处理 pull_request_review 事件：非 COMMENTED → 通知 PR 作者。

    支持两种 payload 格式：
    - repo webhook: review.state = "APPROVED" / "REQUEST_CHANGES"
    - org webhook (Gitea v1.23.4): review.type = "pull_request_review_approved" / "pull_request_review_rejected"
    """
    review = payload.get("review")
    if not review or not isinstance(review, dict):
        logger.warning(
            "pull_request_review event missing review field, skipping")
        return
    pr = payload.get("pull_request")
    if not pr or not isinstance(pr, dict):
        logger.warning(
            "pull_request_review event missing pull_request field, skipping")
        return

    # 兼容两种 payload 格式提取 state
    state = review.get("state", "")
    if not state:
        # org webhook 格式：review.type = "pull_request_review_approved"
        review_type = review.get("type", "")
        type_map = {
            "pull_request_review_approved": "APPROVED",
            "pull_request_review_rejected": "REQUEST_CHANGES",
            "pull_request_review_comment": "COMMENTED",
        }
        state = type_map.get(review_type, "")

    repo = _repo_fullname(payload)
    pr_number = pr.get("number", 0)
    pr_title = pr.get("title", "")
    pr_author = pr.get("user", {}).get("login", "unknown")
    # 兼容：org webhook 的 review 没有 user，从 sender 取
    reviewer = review.get(
        "user",
        {}).get(
        "login",
        "") or payload.get(
            "sender",
            {}).get(
                "login",
        "unknown")
    review_body = review.get("body", "") or review.get("content", "(无评论)")

    if state == "COMMENTED":
        # Review 评论 → 通知 PR 作者
        review_body = review.get("body", "") or review.get("content", "(无评论)")
        reviewer = review.get("user", {}).get("login", "") or payload.get("sender", {}).get("login", "unknown")

        text = render_template("review_comment", {
            "repo": repo,
            "pr_number": str(pr_number),
            "pr_title": pr_title,
            "reviewer": reviewer,
            "comment_body": review_body,
        })

        title = f"Review 评论: {pr_title} ({repo}#{pr_number})"
        _send_mail(pr_author, title, text)
        return

    result_map = {"APPROVED": "通过 ✓", "REQUEST_CHANGES": "驳回 ✗"}
    if state not in result_map:
        return
    result = result_map[state]

    text = render_template("review_result", {
        "repo": repo,
        "pr_number": str(pr_number),
        "pr_title": pr_title,
        "reviewer": reviewer,
        "result": result,
        "review_body": review_body,
    })

    title = f"Review {result}: {pr_title} ({repo}#{pr_number})"
    _send_mail(pr_author, title, text)


async def _fetch_latest_reviewer(repo: str, pr_number: int) -> str:
    """查询 PR 最近一次非 PENDING review 的提交者。

    Returns:
        reviewer login 或空字符串
    """
    if not _GITEA_TOKEN:
        return ""

    url = f"{_GITEA_BASE}/repos/{repo}/pulls/{pr_number}/reviews"
    headers = {"Authorization": f"token {_GITEA_TOKEN}"}

    try:
        async with httpx.AsyncClient(timeout=5.0) as client:
            resp = await client.get(url, headers=headers)
            resp.raise_for_status()
            reviews = resp.json()

            # 取最后一个非 PENDING 的 review 的 user
            for review in reversed(reviews):
                state = review.get("state", "")
                if state in ("APPROVED", "REQUEST_CHANGES", "COMMENTED"):
                    user = review.get("user", {})
                    return user.get("login", "")
    except Exception as e:
        logger.warning("Failed to fetch reviews for %s#%d: %s", repo, pr_number, e)

    return ""


async def _handle_pr_synchronize(payload: Dict[str, Any]) -> None:
    """PR 更新（新 push）→ 通知 reviewer 重新 review。

    查询最近一次 review 的提交者作为通知目标。
    只在有 review 历史时才通知（避免和 opened 重复）。
    """
    pr = payload.get("pull_request")
    if not pr or not isinstance(pr, dict):
        return

    repo = _repo_fullname(payload)
    pr_number = pr.get("number", 0)
    pr_title = pr.get("title", "")
    pr_author = pr.get("user", {}).get("login", "unknown")
    new_sha = pr.get("head", {}).get("sha", "unknown")[:12]

    # 查询最近 review 的提交者
    reviewer = await _fetch_latest_reviewer(repo, pr_number)
    if not reviewer:
        # 没有已有 review 历史，fallback 到默认 reviewer
        reviewer = "simayi-challenger"
        logger.info("No review history for PR #%s, using default reviewer %s", pr_number, reviewer)

    text = render_template("review_updated", {
        "repo": repo,
        "pr_number": str(pr_number),
        "pr_title": pr_title,
        "pr_author": pr_author,
        "new_sha": new_sha,
        "reviewer": reviewer,
    })

    title = f"PR 更新: {pr_title} ({repo}#{pr_number})"
    _send_mail(reviewer, title, text)


def _send_deploy_failure_mail(repo: str, pr_number: int, pr_title: str, reason: str) -> None:
    """CD 部署失败通知，复用 deploy_failure 模板"""
    text = render_template("deploy_failure", {
        "repo": repo,
        "commit_sha": f"PR #{pr_number}",
    })
    title = f"部署失败: {repo} (auto-deploy, PR #{pr_number})"
    full_text = f"{text}\n\n失败原因: {reason}"
    for agent_id in ("jiangwei-infra", "pangtong-fujunshi"):
        _send_mail(agent_id, title, full_text)


async def _handle_pr_closed(payload: Dict[str, Any]) -> None:
    """PR closed → 如果 merged，通知 PR 作者。"""
    pr = payload.get("pull_request")
    if not pr or not isinstance(pr, dict):
        return

    # 只处理 merged 的 PR
    if not pr.get("merged", False):
        return

    repo = _repo_fullname(payload)
    pr_number = pr.get("number", 0)
    pr_title = pr.get("title", "")
    pr_author = pr.get("user", {}).get("login", "unknown")
    # merged_by 可能不在 payload 中，fallback 到 sender
    merged_by = (
        pr.get("merged_by", {}).get("login", "")
        or payload.get("sender", {}).get("login", "unknown")
    )

    text = render_template("review_merged", {
        "repo": repo,
        "pr_number": str(pr_number),
        "pr_title": pr_title,
        "pr_author": pr_author,
        "merged_by": merged_by,
    })

    title = f"PR 已合并: {pr_title} ({repo}#{pr_number})"
    _send_mail(pr_author, title, text)

    # 自动部署：git pull + rsync + 按需 post_deploy
    try:
        import yaml

        # 加载部署配置
        config_path = Path(__file__).parent.parent.parent / "config" / "deploy-targets.yaml"
        if not config_path.exists():
            return

        with open(config_path, "r", encoding="utf-8") as f:
            deploy_config = yaml.safe_load(f) or {}

        targets = deploy_config.get("targets", {})
        target = targets.get(repo)
        if not target:
            return  # 该仓库不在部署配置中，跳过

        dev_dir = os.path.expanduser(target["dev_dir"])
        install_dir = os.path.expanduser(target.get("install_dir", target["dev_dir"]))
        rsync_excludes = target.get("rsync_exclude", [])

        # Step 1: git pull in dev dir
        proc = await asyncio.create_subprocess_exec(
            "git", "pull", "origin", "main",
            cwd=dev_dir,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=30)

        if proc.returncode != 0:
            logger.warning("Auto-deploy: git pull failed for %s: %s", repo, stderr.decode())
            return

        logger.info("Auto-deploy: git pull success for %s", repo)

        # Step 2: rsync to install dir
        rsync_args = ["rsync", "-a"]
        for exc in rsync_excludes:
            rsync_args.extend(["--exclude", exc])
        rsync_args.extend([f"{dev_dir}/", f"{install_dir}/"])

        rsync_proc = await asyncio.create_subprocess_exec(
            *rsync_args,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        _, rsync_err = await asyncio.wait_for(rsync_proc.communicate(), timeout=60)

        if rsync_proc.returncode != 0:
            logger.error("Auto-deploy: rsync failed: %s", rsync_err.decode())
            _send_deploy_failure_mail(repo, pr_number, pr_title, f"rsync 失败: {rsync_err.decode()}")
            return

        # Step 3: 判断是否需要执行 post_deploy
        files = await _fetch_pr_files(repo, pr_number)
        file_list = files[0]
        needs_restart = any(
            f.startswith("src/") or f.startswith("templates/") or f.startswith("frontend/") or f.endswith(".py")
            for f in file_list
        )

        if needs_restart:
            post_deploy_cmds = target.get("post_deploy", [])
            pm2_name = target.get("pm2_name", "")
            for cmd in post_deploy_cmds:
                logger.info("Auto-deploy: executing post_deploy: %s", cmd)

                # M2: 检测当前进程是否会被此命令杀掉（而非脆弱的字符串匹配）
                # 通过 PM2 环境变量判断：pm2 启动的进程有 PM2_HOME
                self_restart = False
                if pm2_name and os.environ.get("PM2_HOME") and "pm2 restart" in cmd:
                    # 检查命令是否包含当前进程名
                    import re
                    if re.search(rf'pm2\s+restart\s+{re.escape(pm2_name)}', cmd):
                        self_restart = True

                if self_restart:
                    # M1: 用 asyncio.sleep 延迟而非 nohup，保留子进程输出和错误检测
                    # 先 sleep 让 handler 正常返回，再启动 restart 命令
                    # restart 的子进程会在父进程死后被 pm2 新进程接管
                    logger.info("Auto-deploy: self-restart detected, deferring 2s: %s", cmd)
                    await asyncio.sleep(2)
                    deploy_proc = await asyncio.create_subprocess_exec(
                        "sh", "-c", cmd,
                        stdout=asyncio.subprocess.PIPE,
                        stderr=asyncio.subprocess.PIPE,
                    )
                    # restart 会杀掉当前进程，communicate 可能不会完成
                    # 但我们至少尝试读取输出
                    try:
                        _, deploy_err = await asyncio.wait_for(
                            deploy_proc.communicate(), timeout=10)
                    except (asyncio.TimeoutError, ProcessLookupError):
                        # 预期行为：进程被 pm2 restart 杀掉
                        logger.info("Auto-deploy: process killed by self-restart (expected)")
                        break
                else:
                    deploy_proc = await asyncio.create_subprocess_exec(
                        "sh", "-c", cmd,
                        stdout=asyncio.subprocess.PIPE,
                        stderr=asyncio.subprocess.PIPE,
                    )
                    _, deploy_err = await asyncio.wait_for(deploy_proc.communicate(), timeout=30)

                    if deploy_proc.returncode != 0:
                        logger.error("Auto-deploy: post_deploy failed: %s", deploy_err.decode())
                        _send_deploy_failure_mail(repo, pr_number, pr_title, f"post_deploy 失败 ({cmd}): {deploy_err.decode()}")
                        break
            else:
                logger.info("Auto-deploy: all post_deploy commands succeeded (files: %s)", ", ".join(file_list[:5]))
        else:
            logger.info("Auto-deploy: docs-only change for %s, skip post_deploy", repo)

    except asyncio.TimeoutError:
        logger.error("Auto-deploy: timeout for %s", repo)
        _send_deploy_failure_mail(repo, pr_number, pr_title, "部署超时")
    except Exception as e:
        logger.error("Auto-deploy: unexpected error: %s", e)


async def _handle_issues(payload: Dict[str, Any]) -> None:
    """处理 issues 事件：assigned → 通知被指派人；opened+部署失败 → 通知运维。"""
    action = payload.get("action", "")
    issue = payload.get("issue")
    if not issue or not isinstance(issue, dict):
        logger.warning("issues event missing issue field, skipping")
        return
    repo = _repo_fullname(payload)
    issue_number = issue.get("number", 0)
    issue_title = issue.get("title", "")

    if action == "assigned":
        assignee = ""
        assignees = issue.get("assignees") or []
        if not assignees:
            single = issue.get("assignee")
            if single and isinstance(single, dict):
                assignees = [single]
        if assignees:
            assignee = assignees[-1].get("login", "")
        else:
            assignee = ""
        if not assignee:
            logger.debug("Issue assigned but no assignee found, skipping")
            return

        labels_list = [lbl.get("name", "")
                       for lbl in (issue.get("labels") or [])]
        labels = ", ".join(labels_list) if labels_list else "(无标签)"
        issue_body = issue.get("body", "(无描述)")
        brief = issue_title[:20].replace(" ", "-").lower()

        text = render_template("issue_assigned", {
            "repo": repo,
            "issue_number": str(issue_number),
            "issue_title": issue_title,
            "labels": labels,
            "issue_body": issue_body or "(无描述)",
            "brief": brief,
        })

        title = f"Issue 指派: {issue_title} ({repo}#{issue_number})"
        _send_mail(assignee, title, text)

    elif action == "opened" and "部署失败" in issue_title:
        # 从 Issue body 提取 commit hash（Gitea deploy workflow 格式）
        sha_match = re.search(r'[0-9a-f]{40}', issue.get("body", ""))
        commit_sha = sha_match.group(0) if sha_match else "(未知)"

        text = render_template("deploy_failure", {
            "repo": repo,
            "commit_sha": commit_sha or "(未知)",
        })

        title = f"部署失败: {repo}"
        for agent_id in ("jiangwei-infra", "pangtong-fujunshi"):
            _send_mail(agent_id, title, text)


async def _handle_issue_comment(payload: Dict[str, Any]) -> None:
    """处理 issue_comment 事件：CI 失败关键词 → 通知 PR 作者。"""
    comment = payload.get("comment")
    if not comment or not isinstance(comment, dict):
        logger.warning("issue_comment event missing comment field, skipping")
        return
    body = comment.get("body", "")

    # 检查是否包含 CI 失败关键词
    if "[CI]" not in body and "CI 失败" not in body:
        return

    issue = payload.get("issue")
    if not issue or not isinstance(issue, dict):
        logger.warning("issue_comment event missing issue field, skipping")
        return

    # 已关闭的 Issue/PR 不再发送 CI 失败通知
    if issue.get("state") == "closed":
        logger.debug(
            "Skipping CI failure notification for closed issue #%s",
            issue.get("number"))
        return

    repo = _repo_fullname(payload)
    issue_number = issue.get("number", 0)

    # 尝试从关联 PR 获取信息
    pr_author = issue.get("user", {}).get("login", "unknown")
    branch_match = re.search(r"分支:\s*(\S+)", body)
    branch = branch_match.group(1) if branch_match else "(未知)"

    # 提取错误摘要（取 comment body 前 500 字符）
    error_summary = body[:500] if body else "(无错误信息)"

    text = render_template("ci_failure", {
        "repo": repo,
        "pr_number": str(issue_number),
        "branch": branch,
        "error_summary": error_summary,
    })

    title = f"CI 失败: {repo}#{issue_number}"
    _send_mail(pr_author, title, text)


# ---------------------------------------------------------------------------
# 事件分发
# ---------------------------------------------------------------------------

_EVENT_HANDLERS: Dict[str, Any] = {
    "pull_request": _handle_pull_request,
    "pull_request_sync": _handle_pr_synchronize,  # Gitea: PR branch push 是独立事件类型
    "pull_request_review": _handle_pull_request_review,
    "pull_request_review_approved": _handle_pull_request_review,
    "pull_request_review_rejected": _handle_pull_request_review,
    "pull_request_review_comment": _handle_pull_request_review,
    "pull_request_comment": _handle_pull_request_review,  # Gitea: review comment 独立事件类型
    # Gitea v1.23.4 实际发出的 review 子事件（无 _review_ 中间段）
    "pull_request_approved": _handle_pull_request_review,
    "pull_request_rejected": _handle_pull_request_review,
    "issues": _handle_issues,
    "issue_comment": _handle_issue_comment,
}


# ---------------------------------------------------------------------------
# Webhook 端点
# ---------------------------------------------------------------------------


@router.post("/webhook/gitea")
async def gitea_webhook(
    request: Request,
    x_gitea_event: Optional[str] = Header(None, alias="X-Gitea-Event"),
    x_gitea_delivery: Optional[str] = Header(None, alias="X-Gitea-Delivery"),
    x_gitea_signature: Optional[str] = Header(None, alias="X-Gitea-Signature"),
) -> Response:
    """Gitea Webhook 接收端点。

    处理流程：签名验证 → 幂等检查 → 事件分发 → Mail 推送。

    返回策略：
    - payload 解析失败 / 未知事件 / 幂等重复 → 200（不触发重试）
    - Mail 创建失败 → 500（触发 Gitea 重试）
    """
    body = await request.body()

    # 1. 签名验证
    if not _verify_signature(body, x_gitea_signature):
        logger.warning("Webhook signature verification failed")
        return Response(status_code=403,
                        content="signature verification failed")

    # 3. 解析 payload（提前解析，用于幂等检查）
    try:
        payload = await request.json()
    except Exception:
        logger.warning("Failed to parse webhook payload")
        return Response(status_code=200, content="invalid payload")

    # 2. 幂等检查（需要在 payload 解析后，以支持内容去重）
    if x_gitea_event and x_gitea_delivery:
        async with _idempotency_lock:
            if _is_duplicate(x_gitea_event, x_gitea_delivery, payload):
                logger.debug(
                    "Duplicate webhook: %s/%s",
                    x_gitea_event,
                    x_gitea_delivery)
                return Response(status_code=200, content="duplicate")

    # 4. 查找 handler
    action = payload.get("action", "")
    logger.info("[WEBHOOK] event=%s action=%s delivery=%s", x_gitea_event, action, x_gitea_delivery)
    handler = _EVENT_HANDLERS.get(x_gitea_event or "")
    if not handler:
        logger.info("[WEBHOOK] Unhandled event type: %s", x_gitea_event)
        return Response(status_code=200,
                        content=f"unhandled event: {x_gitea_event}")

    # 5. 执行 handler
    try:
        await handler(payload)
    except Exception:
        logger.exception("Mail creation failed for %s event", x_gitea_event)
        return Response(status_code=500, content="mail creation failed")

    return Response(status_code=200, content="ok")
