from __future__ import annotations

import email.parser
import email.policy
import re
import shutil
import subprocess
import typing as t

from mercurial import extensions
from mercurial import wireprotoserver
from mercurial.thirdparty import attr

from . import _export as xp

if t.TYPE_CHECKING:
    import mercurial.hgweb.hgweb_mod_inner as web_inner
    import mercurial.hgweb.request as hgreq
    import mercurial.ui as hgui

    PermissionCheck = t.Callable[
        [web_inner.requestcontext, hgreq.parsedrequest, bytes],
        None,
    ]


_CGI_VAR = re.compile(rb'[A-Z0-9_]+$')
"""Environment variables that we need to pass to git-as-cgi."""


def _build_git_environ(
    req_ctx: web_inner.requestcontext,
    request: hgreq.parsedrequest,
) -> dict[bytes, bytes]:
    """Builds the environment to be sent to Git to serve HTTP."""
    fixed = {
        k: v
        for (k, v) in request.rawenv.items()
        if isinstance(v, bytes) and _CGI_VAR.match(k)
    }
    repo = req_ctx.repo
    assert xp.is_gitty(repo)
    fixed.update(
        {
            b'GIT_HTTP_EXPORT_ALL': b'yes',
            b'GIT_PROJECT_ROOT': repo.githandler.gitdir,
            b'PATH_INFO': b'/' + request.dispatchpath,
            # Since Mercurial is taking care of authorization checking,
            # we tell Git to always allow push.
            b'GIT_CONFIG_COUNT': b'1',
            b'GIT_CONFIG_KEY_0': b'http.receivepack',
            b'GIT_CONFIG_VALUE_0': b'true',
        }
    )
    return fixed


def _parse_cgi_response(
    output: t.IO[bytes],
) -> tuple[bytes, dict[bytes, bytes], t.IO[bytes]]:
    """Parses a CGI response into a status, headers, and everyhting else."""
    parser = email.parser.BytesFeedParser(policy=email.policy.HTTP)
    while line := output.readline():
        if not line.rstrip(b'\r\n'):
            # We've reached the end of the headers.
            # Leave the rest in the output for later.
            break
        parser.feed(line)
    msg = parser.close()
    status = msg.get('Status', '200 OK I guess').encode('utf-8')
    del msg['Status']  # this won't raise an exception
    byte_headers = {
        k.encode('utf-8'): v.encode('utf-8') for (k, v) in msg.items()
    }
    return status, byte_headers, output


def _git_service_permission(request: hgreq.parsedrequest) -> bytes | None:
    """Figures out what Mercurial permission corresponds to a request from Git.

    If the request is a supported Git action, returns the permission it needs.
    If the request is not a Git action, returns None.
    """
    if perm := xp.SERVICE_PERMISSIONS.get(request.dispatchpath):
        return perm
    if request.dispatchpath != b'info/refs':
        return None
    qs = request.querystring
    service = qs.removeprefix(b'service=')
    if qs == service:
        # Nothing was stripped.
        return None
    return xp.SERVICE_PERMISSIONS.get(service)


def _handle_git_protocol(
    original: t.Callable[..., bool],
    req_ctx: web_inner.requestcontext,
    request: hgreq.parsedrequest,
    response: hgreq.wsgiresponse,
    check_permission: PermissionCheck,
) -> bool:
    """Intercepts requests from Git, if needed."""
    perm = _git_service_permission(request)
    repo = req_ctx.repo
    if not perm or not xp.is_gitty(repo):
        # We only handle Git requests to Gitty repos.
        return original(req_ctx, request, response, check_permission)

    # Permission workaround: Mercurial requires POSTs for push,
    # but the advertisement request from Git will be a GET.
    # We just lie to Mercurial about what we're doing.
    check_permission(
        req_ctx,
        (
            attr.evolve(req_ctx.req, method=b'POST')
            if perm == xp.PUSH
            else req_ctx.req
        ),
        perm,
    )
    cgi_env = _build_git_environ(req_ctx, request)
    http_backend = repo.ui.configlist(
        b'hggit-serve', b'http-backend', default=(b'git', b'http-backend')
    )
    call = subprocess.Popen(
        http_backend,
        close_fds=True,
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.DEVNULL,
        env=cgi_env,
        text=False,
    )
    assert call.stdout
    assert call.stdin
    # Git will not start writing output until stdin is fully closed.
    with call.stdin:
        # This is how we know if there's anything to read from bodyfh.
        # If we try to read from bodyfh on a request with no content,
        # it hangs forever.
        if b'CONTENT_LENGTH' in request.rawenv:
            shutil.copyfileobj(request.bodyfh, call.stdin)

    status, headers, rest = _parse_cgi_response(call.stdout)
    response.status = status
    for k, v in headers.items():
        response.headers[k] = v

    def write_the_rest() -> t.Iterator[bytes]:
        with call, rest:
            # if it's good enough for shutil it's good enough for me
            # technically not in the docs but everybody it
            bs = shutil.COPY_BUFSIZE  # type: ignore[attr-defined]
            while more := rest.read(bs):
                yield more
        if perm == xp.PUSH:
            # If we pushed, we need to import any new refs back into Mercurial.
            xp.import_all(repo, b'git-http-push')

    response.setbodygen(write_the_rest())
    response.sendresponse()
    return True


#
# Interfacing with Mercurial
#


def uisetup(_: hgui.ui) -> None:
    extensions.wrapfunction(
        wireprotoserver, 'handlewsgirequest', _handle_git_protocol
    )