#!/usr/bin/env python3
# /// script
# requires-python = ">= 3.10
# dependencies = []
# ///
import contextlib
import hashlib
import json
import os
import pathlib
import re
import sys
import threading
import typing
import urllib.error
import urllib.parse
import urllib.request
from subprocess import run, PIPE, DEVNULL

__VERSION__ = "0.1"
SCRIPT_URL = 'https://raw.githubusercontent.com/odoo/runbot/refs/heads/17.0/runbot_merge/scripts/git-fw'
GIT_DIR = None


class PrInfo(typing.TypedDict):
    display: str
    repository: str
    target: str
    number: int
    state: str
    commits_map: dict[str, str]
    batch: int
    ref: str
    head: str
    parent: str | None
    child: str | None


HELP = """\
NAME
\tgit-fw - Apply the changes introduced by some existing PR

SYNOPSYS
\tgit fw redo <pr>
\tgit fw (--continue | --skip | --abort | --quit)

DESCRIPTION
\tGiven a pull request (by number, <owner>/<repo>#<number> short form, or full
\tgithub or mergebot URL), re-execute its forward porting via git-cherry-pick(1),
\tproviding the ability to interactively fix conflicts then push the result
\tback onto the forward-port branch.

\tIf used with the git-cherry-pick(1) subcommands, forwards them to the
\tcurrent cherry-pick.
"""


ENV = {**os.environ, 'GIT_EDITOR': 'git fw edit'}
def main() -> int:
    match sys.argv[1:]:
        # case ['do', pr]:
        #     return do(pr)
        case ["redo", pr]:
            threading.Thread(target=check_version, args=(__file__,), daemon=True).start()
            return redo(pr)
        case [("--continue" | "--skip" | "--abort" | "--quit") as subcommand]:
            r = run(["git", "cherry-pick", subcommand], env=ENV)
            exit(r.returncode)
        case ["edit", msgpath]:
            return edit(msgpath)
        case ["--version"]:
            print(f"git-fw version {__VERSION__}")
            check_version(__file__)
            exit(0)
        case _:
            print(HELP)
            check_version(__file__)
            exit(0)


def redo(pr: str) -> int:
    repo, number = parse_pr(pr)

    print(f"Resolving {repo}#{number}")
    res = info(f"https://mergebot.odoo.com/{repo}/pull/{number}.json")
    if not res["parent"]:
        exit("not a forward port")
    print(".", end="", flush=True)
    parent = info(res["parent"])
    print(".", end="", flush=True)
    repository_url, reflist = ls_remote(
        res["repository"],
        res["target"],
        parent["target"],
        res['ref'],
    )
    print(".", end="", flush=True)
    refs = dict(reflist)
    commits_of_interest = [*refs.values(), parent["head"], res["head"], res['ref']]
    if cm := parent["commits_map"]:
        commits_of_interest.append(cm[""])
    if f'refs/heads/{res["ref"]}' in refs:
        devname = res['repository']
    else:
        # try to probe a dev repository for the branch
        devname = res['repository'].replace('odoo/', 'odoo-dev/')
        dev_url, reflist = ls_remote(devname, res['ref'])
        if not reflist:
            exit(f"\nunable to find ref {res['ref']!r} in {res['repository']!r} or {devname!r}")

    urls = tuple(
        f"{root}{devname}{ext}"
        for root in ("https://github.com/", 'git@github.com:')
        for ext in ('.git', '')
    )

    for name, url, kind in remotes():
        if kind == 'fetch' and url in urls:
            remotename = name
            break
    else:
        exit(f"\nfound {res['ref']!r} in {devname!r} but no configured remote for it")
    print(".", end="", flush=True)

    r = run(
        ["git", "fetch", "--no-tags", "-q", remotename, *commits_of_interest],
        stdout=DEVNULL,
    )
    if r.returncode:
        exit(f"\nunable to retrieve commits of interest from {remotename!r}")
    print(".", end="", flush=True)

    # list the source's commits
    r = run(
        [
            "git",
            "rev-list",
            "--reverse",
            parent["head"],
            "--not",
            refs[f"refs/heads/{parent['target']}"],
        ],
        stdout=PIPE,
        encoding="utf-8",
    )
    if r.returncode:
        exit(r.returncode)
    print(".")
    commits = r.stdout.splitlines(keepends=False)
    if cm:
        # remap commits to forward port to those which got merged
        commits = [cm[c] for c in commits]
        with GIT_DIR.joinpath("FW_CMAP").open("w", encoding="utf-8") as f:
            json.dump(cm, f)

    run(["git", "switch", "--guess", res["ref"]], check=True)
    run(["git", "branch", "--set-upstream-to", f"{remotename}/{res['ref']}"], check=True, stdout=DEVNULL)
    merge_base = run(
        ["git", "merge-base", refs[f"refs/heads/{res['target']}"], res["head"]],
        stdout=PIPE,
        encoding="utf-8",
        check=True,
    )
    run(["git", "reset", "--hard", merge_base.stdout.strip()], check=True)
    return run(
        ["git", "cherry-pick", "-e", f"{commits[0]}~..{commits[-1]}"],
        env=ENV,
    ).returncode


def parse_pr(pr: str) -> tuple[str, str]:
    if m := re.fullmatch(r'#?(\d+)', pr):
        return from_remotes(), m[1]

    if m := re.fullmatch(r"(\w+/\w+)#(\d+)", pr):
        return m[1], m[2]

    u = urllib.parse.urlsplit(pr)
    if u.netloc not in ("github.com", "mergebot.odoo.com"):
        exit(f"invalid domain {u.netloc!r}, should be github.com or mergebot.odoo.com")
    if m := re.fullmatch(r"/(\w+/\w+)/pull/(\d+)", u.path):
        return m[1], m[2]

    exit(f"failed to find PR information in {pr!r}")


def remotes() -> typing.Iterator[tuple[str, str, str]]:
    r = run(["git", "remote", "-v"], check=True, stdout=PIPE, encoding="utf-8")
    for line in r.stdout.splitlines(keepends=False):
        # a line should be name TAB url SPACE ( kind )
        if m := re.fullmatch(
            r'''
            (?P<name>[^\t]+)
            \t
            (?P<url>[^ ]+)
            \x20
            \((?P<kind>.+)\)
            # promisor filters
            (?: \[.*])?
            ''',
            line,
            re.VERBOSE,
        ):
            yield typing.cast(tuple[str, str, str], m.groups())


def from_remotes() -> str:
    for _name, url, _kind in remotes():
        for prefix in ('git@github.com:', 'https://github.com/'):
            if url.startswith(prefix):
                return url.removeprefix(prefix)\
                    .removesuffix('.git')\
                    .replace('odoo-dev/', 'odoo/')
    exit(f"Found no github remote in:\n{r.stdout}")


def ls_remote(
    repository: str, *patterns: str
) -> tuple[str, typing.Iterator[tuple[str, str]]]:
    # TODO: maybe support credentials / tokens to be injected in the URL? Not
    #       sure if that's compatible in any way with ssh
    for repository_url in [
        # https seems reliably faster, but doesn't necessarily work if credentials are required
        f"https://github.com/{repository}",
        # ssh is slower, but works better with private repos
        f"git@github.com:{repository}",
    ]:
        r = run(
            ["git", "-c", "credential.interactive=false", "ls-remote", "-q", repository_url, *patterns],
            stdout=PIPE,
            stderr=DEVNULL,
        )
        if r.returncode == 0:
            break
    else:
        exit(f"unable to retrieve branches from {repository!r}")

    return repository_url, (
        (ref, oid)
        for oid, _, ref in (
            line.partition("\t")
            for line in r.stdout.decode().splitlines(keepends=False)
        )
    )


def info(url: str) -> PrInfo:
    try:
        return json.load(urllib.request.urlopen(url))
    except urllib.error.URLError as e:
        exit(str(e.reason))
    except json.JSONDecodeError as e:
        exit(f"invalid response (not json): {e}")


def edit(msgpath: str) -> int:
    source_commit = GIT_DIR.joinpath('CHERRY_PICK_HEAD').read_text(encoding="utf-8").strip()

    original = source_commit
    with contextlib.suppress(FileNotFoundError):
        # When the parent PR is merged, `x-original-commit` could (/ should) be
        # the commit we merged, however we don't want its `closes` or
        # `signed-off-by`, so retrieve and copy the commit message from the PR.
        with GIT_DIR.joinpath('FW_CMAP').open(encoding="utf-8") as f:
            cmap = json.load(f)
        original = next(o for o, f in cmap.items() if f == source_commit)
    r = run(
        ['git', 'show', '--format=%B', '--no-patch', original],
        stdout=PIPE,
        stderr=DEVNULL,
    )
    if r.returncode:
        exit(f"Unable to get commit message for {original} (source of {source_commit})")

    with open(msgpath, "wb") as f:
        r = run([
            'git', 'interpret-trailers',
                # WARNING: order is important
                '--if-exists=replace',
                f'--trailer=X-original-commit:{source_commit}',
        ], input=r.stdout, stdout=f, stderr=DEVNULL)
    return r.returncode


def check_version(f):
    if not f:
        return

    with urllib.request.urlopen(SCRIPT_URL) as r:
        hh = hashlib.blake2b(r.read(), usedforsecurity=False)
    h = hashlib.blake2b(pathlib.Path(f).read_bytes(), usedforsecurity=False)
    if h.digest() != hh.digest():
        print(f"You may want to update {f}:\nA different version is available at {SCRIPT_URL}")


if __name__ == "__main__":
    r = run(["git", "rev-parse", "--git-dir"], stdout=PIPE, stderr=DEVNULL, encoding="utf-8")
    if r.returncode == 0:
        GIT_DIR = pathlib.Path(r.stdout.strip())
    exit(main())