"""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())