#!/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 [--exact] <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.

\tBy default, the forward-port is redone on the current state of the target
\tbranch (on the remote, not locally, so may include changes which have not
\tbeen fetched yet), the conflicts might be different but it includes
\teverything that would otherwise have to be rebased on.

\t--exact can be passed to redo the forward-port exactly as it was done
\toriginally instead.

\tSupports the git-cherry-pick(1) subcommands so that it can properly rewrite
\tcommit messages when resuming.
"""


ENV = {**os.environ, 'GIT_EDITOR': 'git fw edit'}
def main() -> int:
    match sys.argv[1:]:
        # case ['do', pr]:
        #     return do(pr)
        case ["redo", *flags, pr]:
            threading.Thread(target=check_version, args=(__file__,), daemon=True).start()
            exit(redo(pr, exact='--exact' in flags))
        case [("--continue" | "--skip" | "--abort" | "--quit") as subcommand]:
            r = run(["git", "cherry-pick", subcommand], env=ENV, stderr=PIPE)
            if r.stderr:
                sys.stderr.buffer.write(r.stderr.replace(b'git cherry-pick', b'git fw'))
            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, *, exact: bool = False) -> int | str:
    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"]:
        return "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:
            return 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:
        return 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:
        return 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:
        return r.returncode
    print(".")
    commits = r.stdout.splitlines(keepends=False)
    commitsmap = GIT_DIR.joinpath("FW_CMAP")
    if cm:
        # remap commits to forward port to those which got merged
        commits = [cm[c] for c in commits]
        with commitsmap.open("w", encoding="utf-8") as f:
            json.dump(cm, f)
    else:
        commitsmap.unlink(missing_ok=True)

    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,
    )
    if exact:
        replay_base = merge_base.stdout.strip()
    else:
        replay_base = refs[f"refs/heads/{res['target']}"]
    run(["git", "reset", "--hard", replay_base], check=True)
    r = run(
        ["git", "cherry-pick", "-e", f"{commits[0]}~..{commits[-1]}"],
        env=ENV, stderr=PIPE,
    )
    if r.stderr:
        sys.stderr.buffer.write(r.stderr.replace(b'git cherry-pick', b'git fw'))
    return r.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
            (?:\x20\[.*])?
            ''',
            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())