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

from __future__ import annotations

import gzip
import os
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_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

_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) -> None:
	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)


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:
	_begin_smart_http(request, project, write=True, operation="git-receive-pack")
	raw_body = await request.body()
	body = _decode_request_body(raw_body, content_encoding)
	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())