#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# ===----------------------------------------------------------------------===##
#
# Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
# See https://llvm.org/LICENSE.txt for license information.
# SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
#
# ===----------------------------------------------------------------------===##
"""Checks for reverts of commits across a given git commit.

To clarify the meaning of 'across' with an example, if we had the following
commit history (where `a -> b` notes that `b` is a direct child of `a`):

123abc -> 223abc -> 323abc -> 423abc -> 523abc

And where 423abc is a revert of 223abc, this revert is considered to be 'across'
323abc. More generally, a revert A of a parent commit B is considered to be
'across' a commit C if C is a parent of A and B is a parent of C.

Please note that revert detection in general is really difficult, since merge
conflicts/etc always introduce _some_ amount of fuzziness. This script just
uses a bundle of heuristics, and is bound to ignore / incorrectly flag some
reverts. The hope is that it'll easily catch the vast majority (>90%) of them,
though.

This is designed to be used in one of two ways: an import in Python, or run
directly from a shell. If you want to import this, the `find_reverts`
function is the thing to look at. If you'd rather use this from a shell, have a
usage example:

```
./revert_checker.py c47f97169 origin/main origin/release/12.x
```

This checks for all reverts from the tip of origin/main to c47f97169, which are
across the latter. It then does the same for origin/release/12.x to c47f97169.
Duplicate reverts discovered when walking both roots (origin/main and
origin/release/12.x) are deduplicated in output.
"""

import argparse
import collections
import itertools
import logging
import re
import subprocess
import sys
from typing import Dict, Generator, Iterable, List, NamedTuple, Optional, Tuple

assert sys.version_info >= (3, 6), "Only Python 3.6+ is supported."

# People are creative with their reverts, and heuristics are a bit difficult.
# At a glance, most reverts have "This reverts commit ${full_sha}". Many others
# have `Reverts llvm/llvm-project#${PR_NUMBER}`.
#
# By their powers combined, we should be able to automatically catch something
# like 80% of reverts with reasonable confidence. At some point, human
# intervention will always be required (e.g., I saw
# ```
# This reverts commit ${commit_sha_1} and
# also ${commit_sha_2_shorthand}
# ```
# during my sample)

_CommitMessageReverts = NamedTuple(
    "_CommitMessageReverts",
    [
        ("potential_shas", List[str]),
        ("potential_pr_numbers", List[int]),
    ],
)


def _try_parse_reverts_from_commit_message(
    commit_message: str,
) -> _CommitMessageReverts:
    """Tries to parse revert SHAs and LLVM PR numbers form the commit message.

    Returns:
        A namedtuple containing:
        - A list of potentially reverted SHAs
        - A list of potentially reverted LLVM PR numbers
    """
    if not commit_message:
        return _CommitMessageReverts([], [])

    sha_reverts = re.findall(
        r"This reverts commit ([a-f0-9]{40})\b",
        commit_message,
    )

    first_line = commit_message.splitlines()[0]
    initial_revert = re.match(r'Revert ([a-f0-9]{6,}) "', first_line)
    if initial_revert:
        sha_reverts.append(initial_revert.group(1))

    pr_numbers = [
        int(x)
        for x in re.findall(
            r"Reverts llvm/llvm-project#(\d+)",
            commit_message,
        )
    ]

    return _CommitMessageReverts(
        potential_shas=sha_reverts,
        potential_pr_numbers=pr_numbers,
    )


def _stream_stdout(
    command: List[str], cwd: Optional[str] = None
) -> Generator[str, None, None]:
    with subprocess.Popen(
        command,
        cwd=cwd,
        stdout=subprocess.PIPE,
        encoding="utf-8",
        errors="replace",
    ) as p:
        assert p.stdout is not None  # for mypy's happiness.
        yield from p.stdout


def _resolve_sha(git_dir: str, sha: str) -> str:
    if len(sha) == 40:
        return sha

    return subprocess.check_output(
        ["git", "-C", git_dir, "rev-parse", sha],
        encoding="utf-8",
        stderr=subprocess.DEVNULL,
    ).strip()


_LogEntry = NamedTuple(
    "_LogEntry",
    [
        ("sha", str),
        ("commit_message", str),
    ],
)


def _log_stream(git_dir: str, root_sha: str, end_at_sha: str) -> Iterable[_LogEntry]:
    sep = 50 * "<>"
    log_command = [
        "git",
        "-C",
        git_dir,
        "log",
        "^" + end_at_sha,
        root_sha,
        "--format=" + sep + "%n%H%n%B%n",
    ]

    stdout_stream = iter(_stream_stdout(log_command))

    # Find the next separator line. If there's nothing to log, it may not exist.
    # It might not be the first line if git feels complainy.
    found_commit_header = False
    for line in stdout_stream:
        if line.rstrip() == sep:
            found_commit_header = True
            break

    while found_commit_header:
        sha = next(stdout_stream, None)
        assert sha is not None, "git died?"
        sha = sha.rstrip()

        commit_message = []

        found_commit_header = False
        for line in stdout_stream:
            line = line.rstrip()
            if line.rstrip() == sep:
                found_commit_header = True
                break
            commit_message.append(line)

        yield _LogEntry(sha, "\n".join(commit_message).rstrip())


def _shas_between(git_dir: str, base_ref: str, head_ref: str) -> Iterable[str]:
    rev_list = [
        "git",
        "-C",
        git_dir,
        "rev-list",
        "--first-parent",
        f"{base_ref}..{head_ref}",
    ]
    return (x.strip() for x in _stream_stdout(rev_list))


def _rev_parse(git_dir: str, ref: str) -> str:
    return subprocess.check_output(
        ["git", "-C", git_dir, "rev-parse", ref],
        encoding="utf-8",
    ).strip()


Revert = NamedTuple(
    "Revert",
    [
        ("sha", str),
        ("reverted_sha", str),
    ],
)


def _find_common_parent_commit(git_dir: str, ref_a: str, ref_b: str) -> str:
    """Finds the closest common parent commit between `ref_a` and `ref_b`.

    Returns:
        A SHA. Note that `ref_a` will be returned if `ref_a` is a parent of
        `ref_b`, and vice-versa.
    """
    return subprocess.check_output(
        ["git", "-C", git_dir, "merge-base", ref_a, ref_b],
        encoding="utf-8",
    ).strip()


def _load_pr_commit_mappings(
    git_dir: str, root: str, min_ref: str
) -> Dict[int, List[str]]:
    git_log = ["git", "log", "--format=%H %s", f"{min_ref}..{root}"]
    results = collections.defaultdict(list)
    pr_regex = re.compile(r"\s\(#(\d+)\)$")
    for line in _stream_stdout(git_log, cwd=git_dir):
        m = pr_regex.search(line)
        if not m:
            continue

        pr_number = int(m.group(1))
        sha = line.split(None, 1)[0]
        # N.B., these are kept in log (read: reverse chronological) order,
        # which is what's expected by `find_reverts`.
        results[pr_number].append(sha)
    return results


# N.B., max_pr_lookback's default of 20K commits is arbitrary, but should be
# enough for the 99% case of reverts: rarely should someone land a cleanish
# revert of a >6 month old change...
def find_reverts(
    git_dir: str,
    across_ref: str,
    root: str,
    max_pr_lookback: int = 20000,
    stop_at_sha: Optional[str] = None,
) -> List[Revert]:
    """Finds reverts across `across_ref` in `git_dir`, starting from `root`.

    These reverts are returned in order of oldest reverts first.

    Args:
        git_dir: git directory to find reverts in.
        across_ref: the ref to find reverts across.
        root: the 'main' ref to look for reverts on.
        max_pr_lookback: this function uses heuristics to map PR numbers to
            SHAs. These heuristics require that commit history from `root` to
            `some_parent_of_root` is loaded in memory. `max_pr_lookback` is how
            many commits behind `across_ref` should be loaded in memory.
        stop_at_sha: If non-None and `stop_at_sha` is encountered while walking
            to `across_ref` from `root`, stop checking for reverts. This allows for
            faster incremental checking between `find_reverts` calls.
    """
    across_sha = _rev_parse(git_dir, across_ref)
    root_sha = _rev_parse(git_dir, root)

    common_ancestor = _find_common_parent_commit(git_dir, across_sha, root_sha)
    if common_ancestor != across_sha:
        raise ValueError(
            f"{across_sha} isn't an ancestor of {root_sha} "
            "(common ancestor: {common_ancestor})"
        )

    intermediate_commits = set(_shas_between(git_dir, across_sha, root_sha))
    assert across_sha not in intermediate_commits

    logging.debug(
        "%d commits appear between %s and %s",
        len(intermediate_commits),
        across_sha,
        root_sha,
    )

    commit_log_stream: Iterable[_LogEntry] = _log_stream(git_dir, root_sha, across_sha)
    if stop_at_sha:
        commit_log_stream = itertools.takewhile(
            lambda x: x.sha != stop_at_sha, commit_log_stream
        )

    all_reverts = []
    # Lazily load PR <-> commit mappings, since it can be expensive.
    pr_commit_mappings = None
    for sha, commit_message in commit_log_stream:
        reverts, pr_reverts = _try_parse_reverts_from_commit_message(
            commit_message,
        )
        if pr_reverts:
            if pr_commit_mappings is None:
                logging.info(
                    "Loading PR <-> commit mappings. This may take a moment..."
                )
                pr_commit_mappings = _load_pr_commit_mappings(
                    git_dir, root_sha, f"{across_sha}~{max_pr_lookback}"
                )
                logging.info(
                    "Loaded %d PR <-> commit mappings", len(pr_commit_mappings)
                )

            for reverted_pr_number in pr_reverts:
                reverted_shas = pr_commit_mappings.get(reverted_pr_number)
                if not reverted_shas:
                    logging.warning(
                        "No SHAs for reverted PR %d (commit %s)",
                        reverted_pr_number,
                        sha,
                    )
                    continue
                logging.debug(
                    "Inferred SHAs %s for reverted PR %d (commit %s)",
                    reverted_shas,
                    reverted_pr_number,
                    sha,
                )
                reverts.extend(reverted_shas)

        if not reverts:
            continue

        resolved_reverts = sorted(set(_resolve_sha(git_dir, x) for x in reverts))
        for reverted_sha in resolved_reverts:
            if reverted_sha in intermediate_commits:
                logging.debug(
                    "Commit %s reverts %s, which happened after %s",
                    sha,
                    reverted_sha,
                    across_sha,
                )
                continue

            try:
                object_type = subprocess.check_output(
                    ["git", "-C", git_dir, "cat-file", "-t", reverted_sha],
                    encoding="utf-8",
                    stderr=subprocess.DEVNULL,
                ).strip()
            except subprocess.CalledProcessError:
                logging.warning(
                    "Failed to resolve reverted object %s (claimed to be reverted "
                    "by sha %s)",
                    reverted_sha,
                    sha,
                )
                continue

            if object_type != "commit":
                logging.error(
                    "%s claims to revert the %s %s, which isn't a commit",
                    sha,
                    object_type,
                    reverted_sha,
                )
                continue

            # Rarely, reverts will cite SHAs on other branches (e.g., revert
            # commit says it reverts a commit with SHA ${X}, but ${X} is not a
            # parent of the revert). This can happen if e.g., the revert has
            # been mirrored to another branch. Treat them the same as
            # reverts of non-commits.
            if _find_common_parent_commit(git_dir, sha, reverted_sha) != reverted_sha:
                logging.error(
                    "%s claims to revert %s, which is a commit that is not "
                    "a parent of the revert",
                    sha,
                    reverted_sha,
                )
                continue

            all_reverts.append(Revert(sha, reverted_sha))


    # Since `all_reverts` contains reverts in log order (e.g., newer comes before
    # older), we need to reverse this to keep with our guarantee of older =
    # earlier in the result.
    all_reverts.reverse()
    return all_reverts


def _main() -> None:
    parser = argparse.ArgumentParser(
        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
    )
    parser.add_argument("base_ref", help="Git ref or sha to check for reverts around.")
    parser.add_argument("-C", "--git_dir", default=".", help="Git directory to use.")
    parser.add_argument("root", nargs="+", help="Root(s) to search for commits from.")
    parser.add_argument("--debug", action="store_true")
    parser.add_argument(
        "-u",
        "--review_url",
        action="store_true",
        help="Format SHAs as llvm review URLs",
    )
    opts = parser.parse_args()

    logging.basicConfig(
        format="%(asctime)s: %(levelname)s: %(filename)s:%(lineno)d: %(message)s",
        level=logging.DEBUG if opts.debug else logging.INFO,
    )

    # `root`s can have related history, so we want to filter duplicate commits
    # out. The overwhelmingly common case is also to have one root, and it's way
    # easier to reason about output that comes in an order that's meaningful to
    # git.
    seen_reverts = set()
    all_reverts = []
    for root in opts.root:
        for revert in find_reverts(opts.git_dir, opts.base_ref, root):
            if revert not in seen_reverts:
                seen_reverts.add(revert)
                all_reverts.append(revert)

    sha_prefix = (
        "https://github.com/llvm/llvm-project/commit/" if opts.review_url else ""
    )
    for revert in all_reverts:
        sha_fmt = f"{sha_prefix}{revert.sha}"
        reverted_sha_fmt = f"{sha_prefix}{revert.reverted_sha}"
        print(f"{sha_fmt} claims to revert {reverted_sha_fmt}")


if __name__ == "__main__":
    _main()