"""Git Smart HTTP transport routes (clone/fetch/push over HTTP)."""

from __future__ import annotations

import gzip
import os
import re
import subprocess
import sys
import threading
from datetime import UTC, datetime
from pathlib import Path
from typing import Annotated, Literal

import pygit2
from fastapi import APIRouter, Depends, Header, HTTPException, Query, Request, status
from fastapi.responses import Response

from pygitweb.auth import (
    ACCESS_TOKEN_COOKIE_NAME,
    SmartHttpCredentials,
    can_push_ref,
    can_read_project,
    load_smart_http_credentials,
    principal_can_read_project,
    principal_can_write_project,
    raise_for_smart_http_credentials,
    unauthorized_smart_http,
)
from pygitweb.config import settings
from pygitweb.dependencies import project_visible_in_list, require_valid_project
from pygitweb.validation import is_valid_project

SmartHttpService = Literal["git-upload-pack", "git-receive-pack"]

http_router = APIRouter(prefix="/http", tags=["smart_http"])

_SMART_HTTP_LOG_LOCK = threading.Lock()
_SMART_HTTP_LOG_MODE = 0o600
_REF_UPDATE_LINE = re.compile(rb"^[0-9a-fA-F]{4,64} [0-9a-fA-F]{4,64} (\S+)$")

_SERVICE_CONTENT_TYPES: dict[SmartHttpService, tuple[str, str]] = {
    "git-upload-pack": (
        "application/x-git-upload-pack-advertisement",
        "application/x-git-upload-pack-result",
    ),
    "git-receive-pack": (
        "application/x-git-receive-pack-advertisement",
        "application/x-git-receive-pack-result",
    ),
}


def _decode_request_body(body: bytes, content_encoding: str | None) -> bytes:
    if not content_encoding or not content_encoding.strip():
        return body
    encoding = content_encoding.strip().lower()
    if encoding in ("identity", "none"):
        return body
    if encoding == "gzip":
        try:
            return gzip.decompress(body)
        except OSError as exc:
            raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid gzip body") from exc
    raise HTTPException(
        status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE,
        detail=f"Unsupported Content-Encoding: {content_encoding}",
    )


def _pkt_line(data: bytes) -> bytes:
    length = len(data) + 4
    return f"{length:04x}".encode("ascii") + data


def _pkt_flush() -> bytes:
    return b"0000"


def _is_valid_http_project(project: str) -> bool:
    return is_valid_project(
        project,
        settings.PROJECTROOT,
        settings.EXPORT_OK,
        settings.STRICT_EXPORT,
        project_visible_in_list,
    )


def require_http_project(project: str) -> str:
    candidate = project.removesuffix(".git")
    if candidate != project and _is_valid_http_project(candidate):
        return candidate
    return require_valid_project(project)


ValidatedHttpProject = Annotated[str, Depends(require_http_project)]


def smart_http_log_path() -> Path:
    return Path.home() / ".pygitweb" / "smart_http.log"


def _request_ip(request: Request) -> str:
    client = request.client
    if client is None or not client.host:
        return "-"
    return client.host.replace("\r", "").replace("\n", "")


def _log_token(value: str) -> str:
    return value.replace("\r", "").replace("\n", "") or "-"


def append_smart_http_log(ip: str, actor: str, *, project: str, operation: str) -> None:
    stamp = datetime.now(UTC).isoformat(timespec="seconds").replace("+00:00", "Z")
    safe_actor = actor.replace("\r", "").replace("\n", "")
    line = f"{stamp} {_log_token(ip)} {_log_token(project)} {_log_token(operation)}"
    if safe_actor:
        line = f"{line} {safe_actor}"
    path = smart_http_log_path()
    path.parent.mkdir(parents=True, exist_ok=True)
    with _SMART_HTTP_LOG_LOCK:
        created = not path.exists()
        with path.open("a", encoding="utf-8", newline="\n") as handle:
            handle.write(f"{line}\n")
        if created and sys.platform != "win32":
            os.chmod(path, _SMART_HTTP_LOG_MODE)


def _begin_smart_http(request: Request, project: str, *, write: bool, operation: str) -> SmartHttpCredentials:
    creds = load_smart_http_credentials(
        request.headers.get("authorization"),
        request.cookies.get(ACCESS_TOKEN_COOKIE_NAME),
        project=project,
        write=write,
    )
    append_smart_http_log(_request_ip(request), creds.actor, project=project, operation=operation)
    raise_for_smart_http_credentials(creds)
    _authorize_smart_http(project, creds, write=write)
    return creds


def receive_pack_updated_refs(body: bytes) -> list[str]:
    """Ref names from a stateless ``git-receive-pack`` command list.

    Stops at the first flush pkt, which separates commands from the pack. Lines that
    are not ``<old> <new> <ref>`` (capability text, push certificates) are ignored.
    """
    refs: list[str] = []
    offset = 0
    total = len(body)
    while offset + 4 <= total:
        header = body[offset : offset + 4]
        try:
            size = int(header, 16)
        except ValueError as exc:
            raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid receive-pack request") from exc
        if size == 0:
            return refs
        if size < 4 or offset + size > total:
            raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid receive-pack request")
        payload = body[offset + 4 : offset + size].split(b"\0", 1)[0].rstrip(b"\n")
        offset += size
        if not payload:
            continue
        match = _REF_UPDATE_LINE.match(payload)
        if match is None:
            continue
        try:
            refs.append(match.group(1).decode("utf-8"))
        except UnicodeDecodeError as exc:
            raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid receive-pack request") from exc
    if offset != total:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid receive-pack request")
    return refs


def _enforce_branch_pushes(project: str, creds: SmartHttpCredentials, body: bytes) -> None:
    if not creds.required or creds.principal is None:
        return
    for ref_name in receive_pack_updated_refs(body):
        if can_push_ref(project, creds.principal, ref_name):
            continue
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail=f"Insufficient permissions for {ref_name}",
        )


def _authorize_smart_http(project: str, creds: SmartHttpCredentials, *, write: bool) -> None:
    if not creds.required:
        return
    principal = creds.principal
    if principal is None:
        if not write and can_read_project(project, None):
            return
        raise unauthorized_smart_http()
    if write and creds.readonly:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Read-only key")
    if write:
        if principal_can_write_project(project, principal):
            return
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Insufficient permissions")
    if principal_can_read_project(project, principal):
        return
    raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="No such project")


def _repo_cwd(project: str) -> Path:
    full = os.path.join(settings.PROJECTROOT, project)
    try:
        git_dir = pygit2.discover_repository(full)
    except (KeyError, pygit2.GitError, OSError) as exc:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="No such project") from exc
    if not git_dir:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="No such project")
    repo = pygit2.Repository(git_dir)
    if repo.is_bare:
        return Path(git_dir)
    worktree = repo.workdir
    if worktree is None:
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Repository has no worktree")
    return Path(worktree)


def _run_git_service(
    service: SmartHttpService,
    project: str,
    *,
    advertise: bool,
    body: bytes | None = None,
    git_protocol: str | None = None,
) -> bytes:
    cwd = _repo_cwd(project)
    args = [settings.GIT, service.removeprefix("git-"), "--stateless-rpc" if not advertise else "--advertise-refs", "."]
    env = os.environ.copy()
    if git_protocol:
        env["GIT_PROTOCOL"] = git_protocol
    try:
        completed = subprocess.run(
            args,
            cwd=str(cwd),
            input=body,
            capture_output=True,
            check=False,
            env=env,
        )
    except OSError as exc:
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(exc)) from exc
    if completed.returncode != 0:
        detail = completed.stderr.decode("utf-8", errors="replace").strip() or f"git {service} failed"
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=detail)
    return completed.stdout


def _service_advertisement(service: SmartHttpService, project: str) -> bytes:
    refs = _run_git_service(service, project, advertise=True)
    announce = _pkt_line(f"# service={service}\n".encode("ascii"))
    return announce + _pkt_flush() + refs


def _smart_http_cache_headers() -> dict[str, str]:
    return {
        "Cache-Control": "no-cache",
        "Pragma": "no-cache",
        "Expires": "Fri, 01 Jan 1980 00:00:00 GMT",
    }


@http_router.get("/{project:path}/info/refs")
def smart_http_info_refs(
    project: ValidatedHttpProject,
    service: Annotated[SmartHttpService, Query(description="Git Smart HTTP service name")],
    request: Request,
) -> Response:
    _begin_smart_http(
        request,
        project,
        write=service == "git-receive-pack",
        operation=f"info/refs?service={service}",
    )
    body = _service_advertisement(service, project)
    advertise_type, _ = _SERVICE_CONTENT_TYPES[service]
    return Response(
        content=body,
        media_type=advertise_type,
        headers=_smart_http_cache_headers(),
    )


@http_router.post("/{project:path}/git-upload-pack")
async def smart_http_upload_pack(
    project: ValidatedHttpProject,
    request: Request,
    content_encoding: Annotated[str | None, Header(alias="Content-Encoding")] = None,
    git_protocol: Annotated[str | None, Header(alias="Git-Protocol")] = None,
) -> Response:
    _begin_smart_http(request, project, write=False, operation="git-upload-pack")
    raw_body = await request.body()
    body = _decode_request_body(raw_body, content_encoding)
    result = _run_git_service(
        "git-upload-pack",
        project,
        advertise=False,
        body=body,
        git_protocol=git_protocol,
    )
    _, result_type = _SERVICE_CONTENT_TYPES["git-upload-pack"]
    return Response(content=result, media_type=result_type, headers=_smart_http_cache_headers())


@http_router.post("/{project:path}/git-receive-pack")
async def smart_http_receive_pack(
    project: ValidatedHttpProject,
    request: Request,
    content_encoding: Annotated[str | None, Header(alias="Content-Encoding")] = None,
    git_protocol: Annotated[str | None, Header(alias="Git-Protocol")] = None,
) -> Response:
    creds = _begin_smart_http(request, project, write=True, operation="git-receive-pack")
    raw_body = await request.body()
    body = _decode_request_body(raw_body, content_encoding)
    _enforce_branch_pushes(project, creds, body)
    result = _run_git_service(
        "git-receive-pack",
        project,
        advertise=False,
        body=body,
        git_protocol=git_protocol,
    )
    _, result_type = _SERVICE_CONTENT_TYPES["git-receive-pack"]
    return Response(content=result, media_type=result_type, headers=_smart_http_cache_headers())