import asyncio
import os
import re
import shlex
from typing import Any
from urllib.parse import urlsplit

import httpx

GITEA_URL = os.getenv("GITEA_URL", "http://127.0.0.1:3000").rstrip("/")
GITEA_TOKEN = os.getenv("GITEA_TOKEN", "")
REVIEW_DIFF_MAX_BYTES = 64 * 1024
REVIEW_DIFF_MAX_LINES = 400
_client: httpx.AsyncClient | None = None


class WorkItems(list[dict]):
    """A list-compatible first page carrying truthful per-stream totals."""

    def __init__(self, items: list[dict], pagination: dict[str, dict]):
        super().__init__(items)
        self.pagination = pagination


class StaleReviewError(ValueError):
    """Raised before mutation when a pull request head changed during review."""


class InvalidReviewCommentError(ValueError):
    """Raised before mutation when an inline comment cannot target this change."""


class StalePullError(ValueError):
    """Raised before merge when an assigned pull request head changed."""


class PullNotMergeableError(ValueError):
    """Raised before merge when current pull state or checks prohibit it."""


class IssueNotAvailableError(ValueError):
    """Raised before assignment when an issue is no longer open and unassigned."""


def _auth() -> dict[str, str]:
    headers: dict[str, str] = {"Accept": "application/json"}
    if GITEA_TOKEN:
        headers["Authorization"] = f"token {GITEA_TOKEN}"
    return headers


def start_client(**kwargs) -> httpx.AsyncClient:
    """Create the application-lifetime Gitea transport."""
    global _client
    _client = httpx.AsyncClient(base_url=GITEA_URL, timeout=10, **kwargs)
    return _client


def _get_client() -> httpx.AsyncClient:
    if _client is None or _client.is_closed:
        return start_client()
    return _client


async def stop_client() -> None:
    global _client
    if _client is not None and not _client.is_closed:
        await _client.aclose()


async def fetch(path: str) -> Any:
    r = await _get_client().get(f"/api/v1/{path}", headers=_auth())
    r.raise_for_status()
    return r.json()


async def fetch_text(path: str, max_bytes: int) -> tuple[str, bool]:
    chunks: list[bytes] = []
    size = 0
    truncated = False
    async with _get_client().stream(
        "GET", f"/api/v1/{path}", headers={**_auth(), "Accept": "text/plain"}
    ) as response:
        response.raise_for_status()
        async for chunk in response.aiter_bytes():
            remaining = max_bytes - size
            if len(chunk) > remaining:
                chunks.append(chunk[:remaining])
                truncated = True
                break
            chunks.append(chunk)
            size += len(chunk)
    return b"".join(chunks).decode("utf-8", errors="replace"), truncated


def _diff_previews(diff: str, stream_truncated: bool) -> dict[str, dict]:
    previews: dict[str, dict] = {}
    current: dict | None = None
    in_hunk = False
    remaining = REVIEW_DIFF_MAX_LINES
    for line in diff.splitlines():
        if line.startswith("diff --git "):
            try:
                target = shlex.split(line)[3]
                filename = target[2:] if target.startswith("b/") else target
            except (IndexError, ValueError):
                current = None
                continue
            current = {
                "diff_lines": [],
                "diff_available": False,
                "diff_binary": False,
                "diff_truncated": stream_truncated,
            }
            previews[filename] = current
            in_hunk = False
            continue
        if current is None:
            continue
        if line.startswith("Binary files ") or line == "GIT binary patch":
            current["diff_binary"] = True
            in_hunk = False
            continue
        if line.startswith("@@"):
            in_hunk = True
        if in_hunk and not line.startswith("\\ No newline at end of file"):
            if remaining:
                current["diff_lines"].append(line)
                current["diff_available"] = True
                remaining -= 1
            else:
                current["diff_truncated"] = True
    return previews


async def current_user() -> dict:
    return await fetch("user")


async def repos() -> list[dict]:
    return await fetch("user/repos?limit=50")


WORK_SEARCHES = {
    "issue": ("assigned=true", "issues", None),
    "pull": ("assigned=true", "pulls", "assigned_to_me"),
    "review": ("review_requested=true", "pulls", "review_requested"),
}


async def work_page(stream: str, page: int = 1, limit: int = 50) -> dict:
    """Load exactly one bounded My Work stream page with upstream totals."""
    selector, item_type, reason = WORK_SEARCHES[stream]
    response = await _get_client().get(
        "/api/v1/repos/issues/search",
        headers=_auth(),
        params={
            "state": "open", selector.split("=", 1)[0]: "true",
            "type": item_type, "limit": limit, "page": page,
        },
    )
    response.raise_for_status()
    payload = response.json()
    if not isinstance(payload, list):
        raise ValueError("Gitea work search response was not a list")
    items = [item for item in payload if isinstance(item, dict)]
    if reason:
        items = [{**item, "work_reasons": [reason]} for item in items]
    try:
        total = max(len(items), int(response.headers.get("X-Total-Count", len(items))))
    except (TypeError, ValueError):
        total = len(items)
    return {
        "stream": stream,
        "items": items,
        "page": page,
        "total": total,
        "has_more": page * limit < total,
    }


def _normalize_available_issue(item: Any) -> dict | None:
    if (
        not isinstance(item, dict)
        or item.get("state") != "open"
        or item.get("pull_request") is not None
        or item.get("assignees") not in (None, [])
    ):
        return None
    labels_value = item.get("labels")
    labels = labels_value if isinstance(labels_value, list) else []
    repository_value = item.get("repository")
    repository = repository_value if isinstance(repository_value, dict) else {}
    return {
        "id": item.get("id"),
        "number": item.get("number"),
        "title": item.get("title", "") if isinstance(item.get("title"), str) else "",
        "body": item.get("body", "") if isinstance(item.get("body"), str) else "",
        "state": "open",
        "repository": repository.get("full_name", "")
        if isinstance(repository.get("full_name"), str) else "",
        "labels": [
            label["name"] for label in labels
            if isinstance(label, dict) and isinstance(label.get("name"), str)
        ],
        "assignees": [],
        "updated_at": item.get("updated_at", "")
        if isinstance(item.get("updated_at"), str) else "",
        "url": _safe_web_url(item.get("html_url")),
    }


async def available_issue_snapshot(max_pages: int = 10, upstream_limit: int = 50) -> list[dict]:
    """Load, filter, and globally rank a bounded snapshot of available issues."""
    items: list[dict] = []
    seen_ids: set[Any] = set()
    upstream_total: int | None = None
    for upstream_page in range(1, max_pages + 1):
        response = await _get_client().get(
            "/api/v1/repos/issues/search",
            headers=_auth(),
            params={
                "state": "open", "type": "issues",
                "limit": upstream_limit, "page": upstream_page,
            },
        )
        response.raise_for_status()
        payload = response.json()
        if not isinstance(payload, list):
            raise ValueError("Gitea available issue search response was not a list")
        if upstream_total is None:
            try:
                upstream_total = max(0, int(response.headers["X-Total-Count"]))
            except (KeyError, TypeError, ValueError):
                upstream_total = None
        for raw_item in payload:
            item = _normalize_available_issue(raw_item)
            if item is not None and item["id"] not in seen_ids:
                seen_ids.add(item["id"])
                items.append(item)
        loaded = upstream_page * upstream_limit
        if not payload or len(payload) < upstream_limit or (
            upstream_total is not None and loaded >= upstream_total
        ):
            break

    priority = {"p0", "priority-high", "critical"}
    items.sort(key=lambda item: (item["repository"], item["number"] or 0))
    items.sort(key=lambda item: item["updated_at"], reverse=True)
    items.sort(key=lambda item: (
        0 if any(str(label).lower() in priority for label in item["labels"]) else 1
    ))
    return items


async def available_issue_page(page: int = 1, limit: int = 50) -> dict:
    """Return one logical page from a bounded, globally ranked available-work scan."""
    items = await available_issue_snapshot()
    total = len(items)
    start = (page - 1) * limit
    page_items = items[start:start + limit]
    return {
        "items": page_items,
        "page": page,
        "total": total,
        "has_more": start + len(page_items) < total,
    }


def _page_metadata(result: dict) -> dict:
    return {
        "page": result["page"],
        "total": result["total"],
        "has_more": result["has_more"],
    }


async def issues() -> WorkItems:
    result = await work_page("issue")
    return WorkItems(result["items"], {"issue": _page_metadata(result)})


def _safe_web_url(value: Any) -> str:
    if not isinstance(value, str):
        return ""
    parsed = urlsplit(value)
    return value if parsed.scheme in {"http", "https"} and parsed.netloc else ""


def _normalize_notifications(threads: Any) -> list[dict]:
    if not isinstance(threads, list):
        raise ValueError("Gitea notification response was not a list")
    normalized = []
    for thread in threads:
        if not isinstance(thread, dict):
            continue
        repository = thread.get("repository")
        subject = thread.get("subject")
        repository = repository if isinstance(repository, dict) else {}
        subject = subject if isinstance(subject, dict) else {}
        subject_url = _safe_web_url(subject.get("html_url"))
        latest_url = _safe_web_url(subject.get("latest_comment_html_url"))
        number_text = (
            urlsplit(subject_url).path.rstrip("/").rsplit("/", 1)[-1]
            if subject_url
            else ""
        )
        normalized.append(
            {
                "id": thread.get("id"),
                "unread": thread.get("unread") is True,
                "updated_at": (
                    thread.get("updated_at")
                    if isinstance(thread.get("updated_at"), str)
                    else ""
                ),
                "repository": (
                    repository.get("full_name")
                    if isinstance(repository.get("full_name"), str)
                    else ""
                ),
                "number": int(number_text) if number_text.isdigit() else None,
                "title": (
                    subject.get("title")
                    if isinstance(subject.get("title"), str) and subject.get("title")
                    else "Untitled update"
                ),
                "subject_type": (
                    subject.get("type")
                    if isinstance(subject.get("type"), str) and subject.get("type")
                    else "Update"
                ),
                "state": (
                    subject.get("state")
                    if isinstance(subject.get("state"), str)
                    else ""
                ),
                "url": latest_url or subject_url,
                "subject_url": subject_url,
            }
        )
    return normalized


async def notifications() -> dict:
    return await notification_page(1)


async def notification_page(page: int, limit: int = 50) -> dict:
    response = await _get_client().get(
        f"/api/v1/notifications?status-types=unread&limit={limit}&page={page}",
        headers=_auth(),
    )
    response.raise_for_status()
    items = _normalize_notifications(response.json())
    try:
        total = max(len(items), int(response.headers.get("X-Total-Count", len(items))))
    except (TypeError, ValueError):
        total = len(items)
    return {
        "items": items,
        "page": page,
        "total": total,
        "has_more": page * limit < total,
    }


async def mark_notification_read(thread_id: int) -> None:
    response = await _get_client().patch(
        f"/api/v1/notifications/threads/{thread_id}?to-status=read",
        headers=_auth(),
    )
    response.raise_for_status()


def _gitea_api_path(value: Any) -> str:
    if not isinstance(value, str):
        return ""
    parsed = urlsplit(value)
    configured = urlsplit(GITEA_URL)
    prefix = "/api/v1/"
    if (
        parsed.scheme not in {"http", "https"}
        or parsed.netloc != configured.netloc
        or not parsed.path.startswith(prefix)
    ):
        return ""
    return parsed.path[len(prefix):] + (("?" + parsed.query) if parsed.query else "")


async def notification_detail(thread_id: int) -> dict:
    thread = await fetch(f"notifications/threads/{thread_id}")
    if not isinstance(thread, dict):
        raise ValueError("Gitea notification thread response was not an object")
    repository_value = thread.get("repository")
    repository = repository_value if isinstance(repository_value, dict) else {}
    subject_value = thread.get("subject")
    subject = subject_value if isinstance(subject_value, dict) else {}
    subject_path = _gitea_api_path(subject.get("url"))
    comment_path = _gitea_api_path(subject.get("latest_comment_url"))
    subject_detail = await fetch(subject_path) if subject_path else {}
    comment = await fetch(comment_path) if comment_path else {}
    subject_detail = subject_detail if isinstance(subject_detail, dict) else {}
    comment = comment if isinstance(comment, dict) else {}
    user_value = comment.get("user")
    user = user_value if isinstance(user_value, dict) else {}
    subject_url = _safe_web_url(subject.get("html_url"))
    latest_url = _safe_web_url(comment.get("html_url")) or _safe_web_url(
        subject.get("latest_comment_html_url")
    )
    return {
        "id": thread_id,
        "repository": repository.get("full_name", "")
        if isinstance(repository.get("full_name"), str)
        else "",
        "title": subject.get("title", "")
        if isinstance(subject.get("title"), str)
        else "",
        "subject_type": subject.get("type", "Update")
        if isinstance(subject.get("type"), str)
        else "Update",
        "state": subject.get("state", "")
        if isinstance(subject.get("state"), str)
        else "",
        "url": latest_url or subject_url,
        "subject_body": subject_detail.get("body", "")
        if isinstance(subject_detail.get("body"), str)
        else "",
        "latest_comment": {
            "author": user.get("login", "")
            if isinstance(user.get("login"), str)
            else "",
            "body": comment.get("body", "")
            if isinstance(comment.get("body"), str)
            else "",
            "created_at": comment.get("created_at", "")
            if isinstance(comment.get("created_at"), str)
            else "",
            "url": latest_url,
        },
    }


async def reply_to_notification(thread_id: int, body: str) -> dict:
    thread = await fetch(f"notifications/threads/{thread_id}")
    if not isinstance(thread, dict):
        raise ValueError("Gitea notification thread response was not an object")
    repository = thread.get("repository")
    subject = thread.get("subject")
    if not isinstance(repository, dict) or not isinstance(subject, dict):
        raise ValueError("Notification does not identify a conversation")
    repository_name = repository.get("full_name")
    subject_path = _gitea_api_path(subject.get("url"))
    match = re.fullmatch(
        r"repos/([^/]+/[^/]+)/(issues|pulls)/(\d+)", subject_path
    )
    if (
        not match
        or match.group(1) != repository_name
        or subject.get("type") not in {"Issue", "Pull"}
    ):
        raise ValueError("Notification subject is not a supported conversation")
    response = await _get_client().post(
        f"/api/v1/repos/{match.group(1)}/issues/{match.group(3)}/comments",
        headers=_auth(),
        json={"body": body},
    )
    response.raise_for_status()
    comment = response.json()
    if not isinstance(comment, dict):
        raise ValueError("Gitea comment response was not an object")
    return {
        "id": comment.get("id"),
        "url": _safe_web_url(comment.get("html_url")),
    }


async def close_issue(repository: str, number: int) -> dict:
    response = await _get_client().patch(
        f"/api/v1/repos/{repository}/issues/{number}",
        headers=_auth(),
        json={"state": "closed"},
    )
    response.raise_for_status()
    issue = response.json()
    if not isinstance(issue, dict) or issue.get("state") != "closed":
        raise ValueError("Gitea did not confirm issue closure")
    return {
        "number": issue.get("number"),
        "state": "closed",
        "closed_at": issue.get("closed_at", "")
        if isinstance(issue.get("closed_at"), str)
        else "",
    }


def _normalize_issue_comment(comment: dict) -> dict:
    user_value = comment.get("user")
    user: dict = user_value if isinstance(user_value, dict) else {}
    return {
        "id": comment.get("id"),
        "author": user.get("login", "") if isinstance(user.get("login"), str) else "",
        "body": comment.get("body", "") if isinstance(comment.get("body"), str) else "",
        "created_at": comment.get("created_at", "")
        if isinstance(comment.get("created_at"), str)
        else "",
        "url": _safe_web_url(comment.get("html_url")),
    }


async def comment_on_issue(repository: str, number: int, body: str) -> dict:
    response = await _get_client().post(
        f"/api/v1/repos/{repository}/issues/{number}/comments",
        headers=_auth(),
        json={"body": body},
    )
    response.raise_for_status()
    comment = response.json()
    if not isinstance(comment, dict):
        raise ValueError("Gitea comment response was not an object")
    return _normalize_issue_comment(comment)


async def repo_labels(repository: str) -> list[dict]:
    response = await _get_client().get(
        f"/api/v1/repos/{repository}/labels",
        headers=_auth(),
        params={"limit": 50, "page": 1},
    )
    response.raise_for_status()
    payload = response.json()
    if not isinstance(payload, list):
        raise ValueError("Gitea labels response was not a list")
    return [
        {
            "id": item["id"],
            "name": item["name"],
            "color": item.get("color", "") if isinstance(item.get("color"), str) else "",
            "description": item.get("description", "")
            if isinstance(item.get("description"), str)
            else "",
        }
        for item in payload
        if isinstance(item, dict)
        and isinstance(item.get("id"), int)
        and isinstance(item.get("name"), str)
    ]


async def create_issue(
    repository: str,
    title: str,
    body: str,
    assignee: str,
    label_ids: list[int] | None = None,
) -> dict:
    payload: dict = {"title": title, "body": body, "assignee": assignee}
    if label_ids:
        payload["labels"] = label_ids
    response = await _get_client().post(
        f"/api/v1/repos/{repository}/issues",
        headers=_auth(),
        json=payload,
    )
    response.raise_for_status()
    issue = response.json()
    if not isinstance(issue, dict) or not isinstance(issue.get("number"), int):
        raise ValueError("Gitea did not confirm issue creation")
    assignees_value = issue.get("assignees")
    assignees = assignees_value if isinstance(assignees_value, list) else []
    confirmed_assignees = [
        item["login"]
        for item in assignees
        if isinstance(item, dict) and isinstance(item.get("login"), str)
    ]
    if assignee not in confirmed_assignees:
        raise ValueError("Gitea did not confirm issue self-assignment")
    labels_value = issue.get("labels")
    labels = labels_value if isinstance(labels_value, list) else []
    return {
        "id": issue.get("id"),
        "number": issue["number"],
        "title": issue.get("title", "")
        if isinstance(issue.get("title"), str)
        else "",
        "state": issue.get("state", "")
        if isinstance(issue.get("state"), str)
        else "",
        "repository": repository,
        "labels": [
            item["name"]
            for item in labels
            if isinstance(item, dict) and isinstance(item.get("name"), str)
        ],
        "assignees": confirmed_assignees,
        "updated_at": issue.get("updated_at", "")
        if isinstance(issue.get("updated_at"), str)
        else "",
        "url": _safe_web_url(issue.get("html_url")),
    }


async def claim_available_issue(repository: str, number: int) -> dict:
    issue = await fetch(f"repos/{repository}/issues/{number}")
    user = await current_user()
    assignees_value = issue.get("assignees") if isinstance(issue, dict) else None
    if (
        not isinstance(issue, dict)
        or issue.get("state") != "open"
        or issue.get("pull_request") is not None
        or assignees_value not in (None, [])
    ):
        raise IssueNotAvailableError("Issue is no longer available")
    login = user.get("login") if isinstance(user, dict) else None
    if not isinstance(login, str) or not login:
        raise ValueError("Gitea current user did not include a login")

    response = await _get_client().patch(
        f"/api/v1/repos/{repository}/issues/{number}",
        headers=_auth(),
        json={"assignee": login},
    )
    response.raise_for_status()
    confirmed = response.json()
    if not isinstance(confirmed, dict) or confirmed.get("number") != number:
        raise ValueError("Gitea did not confirm issue assignment")
    confirmed_assignees_value = confirmed.get("assignees")
    confirmed_assignees = (
        confirmed_assignees_value if isinstance(confirmed_assignees_value, list) else []
    )
    logins = [
        assignee["login"] for assignee in confirmed_assignees
        if isinstance(assignee, dict) and isinstance(assignee.get("login"), str)
    ]
    if login not in logins:
        raise ValueError("Gitea did not confirm issue assignment")
    labels_value = confirmed.get("labels")
    labels = labels_value if isinstance(labels_value, list) else []
    return {
        "id": confirmed.get("id"),
        "number": number,
        "title": confirmed.get("title", "")
        if isinstance(confirmed.get("title"), str) else "",
        "state": "open",
        "repository": repository,
        "labels": [
            label["name"] for label in labels
            if isinstance(label, dict) and isinstance(label.get("name"), str)
        ],
        "assignees": logins,
        "updated_at": confirmed.get("updated_at", "")
        if isinstance(confirmed.get("updated_at"), str) else "",
        "url": _safe_web_url(confirmed.get("html_url")),
    }


async def update_issue_labels(repository: str, number: int, label_ids: list[int]) -> dict:
    response = await _get_client().patch(
        f"/api/v1/repos/{repository}/issues/{number}",
        headers=_auth(),
        json={"labels": label_ids},
    )
    response.raise_for_status()
    issue = response.json()
    if not isinstance(issue, dict) or issue.get("number") != number:
        raise ValueError("Gitea did not confirm the label update")
    labels_value = issue.get("labels")
    labels = labels_value if isinstance(labels_value, list) else []
    confirmed_ids = {
        item["id"] for item in labels
        if isinstance(item, dict) and isinstance(item.get("id"), int)
    }
    if confirmed_ids != set(label_ids):
        raise ValueError("Gitea did not confirm the requested label set")
    return {
        "number": number,
        "labels": [
            item["name"] for item in labels
            if isinstance(item, dict) and isinstance(item.get("name"), str)
        ],
    }


async def issue_detail(repository: str, number: int) -> dict:
    base = f"repos/{repository}/issues/{number}"
    issue, comments = await asyncio.gather(
        fetch(base), fetch(f"{base}/comments?limit=20&page=1")
    )
    if not isinstance(issue, dict):
        raise ValueError("Gitea issue response was not an object")
    labels_value = issue.get("labels")
    labels: list = labels_value if isinstance(labels_value, list) else []
    assignees_value = issue.get("assignees")
    assignees: list = assignees_value if isinstance(assignees_value, list) else []
    comments_value: list = comments if isinstance(comments, list) else []
    normalized_comments = [
        _normalize_issue_comment(comment)
        for comment in comments_value
        if isinstance(comment, dict)
    ]
    return {
        "repository": repository,
        "number": number,
        "title": issue.get("title", "") if isinstance(issue.get("title"), str) else "",
        "state": issue.get("state", "") if isinstance(issue.get("state"), str) else "",
        "body": issue.get("body", "") if isinstance(issue.get("body"), str) else "",
        "url": _safe_web_url(issue.get("html_url")),
        "labels": [
            label["name"] for label in labels
            if isinstance(label, dict) and isinstance(label.get("name"), str)
        ],
        "assignees": [
            assignee["login"] for assignee in assignees
            if isinstance(assignee, dict) and isinstance(assignee.get("login"), str)
        ],
        "comments": normalized_comments,
    }


async def _current_login_and_target(path: str) -> tuple[str, dict]:
    user, target = await asyncio.gather(current_user(), fetch(path))
    login = user.get("login") if isinstance(user, dict) else None
    if not isinstance(login, str) or not login or not isinstance(target, dict):
        return "", {}
    return login, target


def _login_in_users(login: str, value: object) -> bool:
    return isinstance(value, list) and any(
        isinstance(user, dict) and user.get("login") == login for user in value
    )


async def is_assigned_issue(repository: str, number: int) -> bool:
    login, issue = await _current_login_and_target(
        f"repos/{repository}/issues/{number}"
    )
    return (
        issue.get("state") == "open"
        and not isinstance(issue.get("pull_request"), dict)
        and _login_in_users(login, issue.get("assignees"))
    )


async def pull_requests() -> WorkItems:
    assigned, review_requested = await asyncio.gather(
        work_page("pull"),
        work_page("review"),
    )
    merged: dict[int, dict] = {}
    for result in (
        assigned,
        review_requested,
    ):
        for pull in result["items"]:
            identity = pull.get("id")
            if identity not in merged:
                merged[identity] = {**pull, "work_reasons": []}
            for reason in pull.get("work_reasons", []):
                if reason not in merged[identity]["work_reasons"]:
                    merged[identity]["work_reasons"].append(reason)
    return WorkItems(
        list(merged.values()),
        {
            "pull": _page_metadata(assigned),
            "review": _page_metadata(review_requested),
        },
    )


async def is_requested_review(repository: str, number: int) -> bool:
    login, pull = await _current_login_and_target(
        f"repos/{repository}/pulls/{number}"
    )
    return (
        pull.get("state") == "open"
        and _login_in_users(login, pull.get("requested_reviewers"))
    )


async def is_assigned_pull(repository: str, number: int) -> bool:
    login, pull = await _current_login_and_target(
        f"repos/{repository}/pulls/{number}"
    )
    return (
        pull.get("state") == "open"
        and _login_in_users(login, pull.get("assignees"))
    )


async def pull_completion_detail(repository: str, number: int) -> dict:
    base = f"repos/{repository}/pulls/{number}"
    pull = await fetch(base)
    if not isinstance(pull, dict):
        raise ValueError("Gitea pull request response was not an object")
    head = pull.get("head") if isinstance(pull.get("head"), dict) else {}
    sha = head.get("sha") if isinstance(head.get("sha"), str) else ""
    files, status, comments = await asyncio.gather(
        fetch(f"{base}/files"),
        fetch(f"repos/{repository}/commits/{sha}/status"),
        fetch(f"repos/{repository}/issues/{number}/comments?limit=20&page=1"),
    )
    user = pull.get("user") if isinstance(pull.get("user"), dict) else {}
    return {
        "repository": repository,
        "number": number,
        "title": pull.get("title") if isinstance(pull.get("title"), str) else "",
        "body": pull.get("body") if isinstance(pull.get("body"), str) else "",
        "url": _safe_web_url(pull.get("html_url")),
        "author": user.get("login") if isinstance(user.get("login"), str) else "",
        "head_sha": sha,
        "state": pull.get("state") if isinstance(pull.get("state"), str) else "",
        "draft": pull.get("draft") is True,
        "mergeable": pull.get("mergeable") is True,
        "merged": pull.get("merged") is True,
        "ci_state": status.get("state", "unknown") if isinstance(status, dict) else "unknown",
        "files": [
            {
                "filename": item.get("filename", ""),
                "status": item.get("status") or "changed",
                "additions": item.get("additions") or 0,
                "deletions": item.get("deletions") or 0,
            }
            for item in (files if isinstance(files, list) else [])[:100]
            if isinstance(item, dict) and isinstance(item.get("filename"), str)
        ],
        "comments": [
            _normalize_issue_comment(item)
            for item in (comments if isinstance(comments, list) else [])[:20]
            if isinstance(item, dict)
        ],
    }


async def merge_assigned_pull(
    repository: str, number: int, expected_head_sha: str
) -> dict:
    base = f"repos/{repository}/pulls/{number}"
    pull = await fetch(base)
    if not isinstance(pull, dict):
        raise PullNotMergeableError("Pull request state is unavailable")
    head_value = pull.get("head")
    head: dict = head_value if isinstance(head_value, dict) else {}
    current_sha = head.get("sha")
    if current_sha != expected_head_sha:
        raise StalePullError("Pull request changed before merge")
    status = await fetch(f"repos/{repository}/commits/{current_sha}/status")
    ci_state = status.get("state") if isinstance(status, dict) else "unknown"
    if (
        pull.get("state") != "open"
        or pull.get("draft") is True
        or pull.get("mergeable") is not True
        or pull.get("merged") is True
        or ci_state != "success"
    ):
        raise PullNotMergeableError("Pull request is not currently safe to merge")
    response = await _get_client().post(
        f"/api/v1/{base}/merge",
        headers=_auth(),
        json={"Do": "merge", "head_commit_id": current_sha},
    )
    response.raise_for_status()
    return {"number": number, "merged": True, "state": "closed"}


async def pull_review_detail(repository: str, number: int) -> dict:
    base = f"repos/{repository}/pulls/{number}"
    pull = await fetch(base)
    if not isinstance(pull, dict):
        raise ValueError("Gitea pull request response was not an object")
    head_value = pull.get("head")
    head: dict = head_value if isinstance(head_value, dict) else {}
    sha_value = head.get("sha")
    sha = sha_value if isinstance(sha_value, str) else ""
    files, status, reviews, diff_result = await asyncio.gather(
        fetch(f"{base}/files"),
        fetch(f"repos/{repository}/commits/{sha}/status"),
        fetch(f"{base}/reviews"),
        fetch_text(
            f"repos/{repository}/pulls/{number}.diff", REVIEW_DIFF_MAX_BYTES
        ),
    )
    diff, diff_truncated = diff_result
    previews = _diff_previews(diff, diff_truncated)
    user_value = pull.get("user")
    user: dict = user_value if isinstance(user_value, dict) else {}
    normalized_files = [
        {
            "filename": file["filename"],
            "status": file.get("status") or "changed",
            "additions": file.get("additions") or 0,
            "deletions": file.get("deletions") or 0,
            **previews.get(
                file["filename"],
                {
                    "diff_lines": [],
                    "diff_available": False,
                    "diff_binary": False,
                    "diff_truncated": diff_truncated,
                },
            ),
        }
        for file in (files if isinstance(files, list) else [])[:100]
        if isinstance(file, dict) and isinstance(file.get("filename"), str)
    ]
    normalized_reviews = [
        {
            "user": {
                "login": (
                    review["user"].get("login", "")
                    if isinstance(review.get("user"), dict)
                    else ""
                )
            },
            "state": review.get("state") or "COMMENT",
            "body": review.get("body") or "",
        }
        for review in (reviews if isinstance(reviews, list) else [])[:50]
        if isinstance(review, dict)
    ]
    return {
        "repository": repository,
        "number": number,
        "title": pull.get("title", ""),
        "body": pull.get("body") or "",
        "url": pull.get("html_url", ""),
        "author": user.get("login", ""),
        "head_sha": sha,
        "ci_state": status.get("state", "unknown") if isinstance(status, dict) else "unknown",
        "files": normalized_files,
        "reviews": normalized_reviews,
    }


async def submit_pull_review(
    repository: str,
    number: int,
    expected_head_sha: str,
    decision: str,
    body: str,
    comments: list[dict] | None = None,
) -> dict:
    base = f"repos/{repository}/pulls/{number}"
    pull = await fetch(base)
    head = pull.get("head") if isinstance(pull, dict) else None
    current_sha = head.get("sha") if isinstance(head, dict) else None
    if current_sha != expected_head_sha:
        raise StaleReviewError("Pull request changed while it was being reviewed")
    if comments:
        files = await fetch(f"{base}/files")
        changed_paths = {
            item.get("filename")
            for item in (files if isinstance(files, list) else [])
            if isinstance(item, dict) and isinstance(item.get("filename"), str)
        }
        if any(comment.get("path") not in changed_paths for comment in comments):
            raise InvalidReviewCommentError("Inline comment path is not in this pull request")
    response = await _get_client().post(
        f"/api/v1/{base}/reviews",
        headers=_auth(),
        json={
            "body": body,
            "event": {
                "comment": "COMMENT",
                "approve": "APPROVE",
                "request_changes": "REQUEST_CHANGES",
            }[decision],
            "commit_id": current_sha,
            **({"comments": comments} if comments else {}),
        },
    )
    response.raise_for_status()
    review = response.json()
    if not isinstance(review, dict):
        raise ValueError("Gitea review response was not an object")
    return {
        "id": review.get("id"),
        "state": review.get("state") or "COMMENT",
        "url": _safe_web_url(review.get("html_url")),
    }


async def activity_events(user: dict | None = None) -> list[dict]:
    if user is None:
        user = await current_user()
    events = await fetch(f"users/{user['login']}/activities/feeds?limit=20")
    if events is None:
        events = []
    elif not isinstance(events, list):
        raise ValueError("Gitea activity feed response was not a list")
    return [
        {
            "type": (
                event.get("op_type")
                if isinstance(event.get("op_type"), str) and event.get("op_type")
                else "activity"
            ),
            "actor": (
                event.get("act_user")
                if isinstance(event.get("act_user"), dict)
                else {}
            ),
            "repo": (
                event.get("repo")
                if isinstance(event.get("repo"), dict)
                else {}
            ),
            "created_at": (
                event.get("created")
                if isinstance(event.get("created"), str)
                else ""
            ),
        }
        for event in (events or [])
        if isinstance(event, dict)
    ]