push.py

#
#  Copyright © 2021-2026 Mergify SAS
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.

from __future__ import annotations

import asyncio
import dataclasses
import datetime
import json
import os
import re
import sys
import typing

import rich.markup

from mergify_cli import console
from mergify_cli import console_error
from mergify_cli import utils
from mergify_cli.exit_codes import ExitCode
from mergify_cli.stack import approvals as approvals_mod
from mergify_cli.stack import changes
from mergify_cli.stack.note import NOTES_REF


if typing.TYPE_CHECKING:
    import httpx

    from mergify_cli import github_types
    from mergify_cli.stack import sync

DEPENDS_ON_RE = re.compile(r"Depends-On: (#[0-9]*)")
_SLUG_SUFFIX_RE = re.compile(r"--[0-9a-f]{8}$")

MAX_CONCURRENT_API_CALLS = 5


@dataclasses.dataclass
class LocalBranchInvalidError(Exception):
    message: str


def check_local_branch(branch_name: str, branch_prefix: str) -> None:
    if not branch_name.startswith(branch_prefix):
        return
    if changes.CHANGEID_SUFFIX_RE.search(branch_name):
        msg = "Local branch is a branch generated by Mergify CLI"
        raise LocalBranchInvalidError(msg)
    if _SLUG_SUFFIX_RE.search(branch_name):
        msg = "Local branch is a branch generated by Mergify CLI"
        raise LocalBranchInvalidError(msg)


def format_pull_description(
    message: str,
    depends_on: github_types.PullRequest | None,
) -> str:
    depends_on_header = ""
    if depends_on is not None:
        depends_on_header = f"\n\nDepends-On: #{depends_on['number']}"

    message = changes.CHANGEID_RE.sub("", message).rstrip("\n")
    message = DEPENDS_ON_RE.sub("", message).rstrip("\n")

    return message + depends_on_header


async def fetch_notes_ref(remote: str) -> bool:
    """Fetch ``refs/notes/mergify/stack`` from *remote* when the local ref
    does not already exist.

    Returns True when the local ref is present (either pre-existing or
    newly fetched) so a lease SHA is available for the subsequent push.
    Returns False only on first push (ref absent both locally and remotely).

    If the local ref already exists (e.g. from a prior ``stack note`` that
    hasn't been pushed yet), the fetch is skipped to avoid clobbering
    unpushed local notes. A divergent remote is then caught by the
    ``--force-with-lease`` check at push time.
    """
    try:
        await utils.git("rev-parse", "--verify", NOTES_REF)
    except utils.CommandError:
        pass
    else:
        # Local ref exists; don't overwrite.
        return True

    try:
        await utils.git(
            "fetch",
            remote,
            "--no-write-fetch-head",
            f"{NOTES_REF}:{NOTES_REF}",
        )
    except utils.CommandError as exc:
        if b"couldn't find remote ref" not in exc.stdout:
            raise
        return False
    return True


async def push_branches(
    remote: str,
    local_changes: list[changes.LocalChange],
    *,
    no_verify: bool = False,
    notes_ref_fetched: bool = False,
) -> None:
    changes_to_push = [c for c in local_changes if c.action in {"create", "update"}]
    if not changes_to_push:
        return

    lease_args: list[str] = []
    for c in changes_to_push:
        if c.action == "update":
            lease_args.append(
                f"--force-with-lease=refs/heads/{c.dest_branch}:{c.pull_head_sha}",
            )
        else:
            lease_args.append(f"--force-with-lease=refs/heads/{c.dest_branch}:")

    notes_local_sha: str | None = None
    try:
        notes_local_sha = await utils.git("rev-parse", "--verify", NOTES_REF)
    except utils.CommandError:
        pass

    if notes_local_sha is not None and notes_ref_fetched:
        lease_args.append(f"--force-with-lease={NOTES_REF}:{notes_local_sha}")

    refspecs = [f"{c.commit_sha}:refs/heads/{c.dest_branch}" for c in changes_to_push]
    if notes_local_sha is not None:
        refspecs.append(f"+{NOTES_REF}:{NOTES_REF}")

    no_verify_args = ("--no-verify",) if no_verify else ()
    os.environ["MERGIFY_STACK_PUSH"] = "1"
    try:
        await utils.git(
            "push",
            "--atomic",
            *no_verify_args,
            *lease_args,
            remote,
            *refspecs,
        )
    finally:
        os.environ.pop("MERGIFY_STACK_PUSH", None)


async def read_reasons(local_changes: list[changes.LocalChange]) -> None:
    """Populate ``c.reason`` from ``refs/notes/mergify/stack`` for each
    change in ``create`` or ``update`` state.

    Commits without a note get an empty string. Unexpected git errors
    (missing binary, permission failures, corrupted object store, …)
    propagate to the caller so real problems surface instead of being
    silently swallowed.
    """

    async def read_one(c: changes.LocalChange) -> None:
        if c.action not in {"create", "update"}:
            return
        try:
            c.reason = await utils.git(
                "notes",
                f"--ref={NOTES_REF}",
                "show",
                c.commit_sha,
            )
        except utils.CommandError as exc:
            if b"no note found" not in exc.stdout:
                raise
            c.reason = ""

    await asyncio.gather(*(read_one(c) for c in local_changes))


async def _git_patch_id(sha: str) -> str:
    """Get the patch-id of a commit, stable across rebases."""
    diff = await utils.git("show", sha)
    proc = await asyncio.subprocess.create_subprocess_exec(
        "git",
        "patch-id",
        "--stable",
        stdin=asyncio.subprocess.PIPE,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.STDOUT,
    )
    stdout, _ = await proc.communicate(input=diff.encode())
    if proc.returncode != 0:
        raise utils.CommandError(
            ("git", "patch-id"),
            proc.returncode,
            stdout or b"",
        )
    # Output format: "<patch-id> <commit-sha>"
    return stdout.decode().strip().split()[0]


async def detect_change_type(old_sha: str, new_sha: str) -> str:
    """Compare patch-ids to determine if a force-push is rebase-only or content change."""
    try:
        old_patch_id = await _git_patch_id(old_sha)
        new_patch_id = await _git_patch_id(new_sha)
    except (utils.CommandError, IndexError, UnicodeDecodeError):
        return "unknown"
    return "rebase" if old_patch_id == new_patch_id else "content"


async def fetch_old_pr_heads(remote: str, pr_numbers: list[int]) -> None:
    """Fetch current PR head refs so old SHAs are available locally for patch-id comparison."""
    if not pr_numbers:
        return
    refspecs = [f"refs/pull/{n}/head" for n in pr_numbers]
    await utils.git("fetch", remote, *refspecs)


@dataclasses.dataclass
class ChangeTask:
    index: int
    change: changes.LocalChange
    depends_on_index: int | None
    pull_ready: asyncio.Event = dataclasses.field(default_factory=asyncio.Event)
    resolved_pull: github_types.PullRequest | None = None


def _build_change_tasks(
    local_changes: list[changes.LocalChange],
) -> list[ChangeTask]:
    tasks: list[ChangeTask] = []
    last_pull_index: int | None = None

    for i, change in enumerate(local_changes):
        task = ChangeTask(
            index=i,
            change=change,
            depends_on_index=last_pull_index,
        )

        # For changes with existing pulls (update/skip-*), signal immediately
        if change.action != "create" and change.pull is not None:
            task.resolved_pull = change.pull
            task.pull_ready.set()
        elif change.action not in {"create", "update"}:
            # skip-* without pull - no one should wait on this
            task.pull_ready.set()

        if change.pull is not None or change.action == "create":
            last_pull_index = i

        tasks.append(task)

    return tasks


async def _process_change_task(
    task: ChangeTask,
    tasks: list[ChangeTask],
    client: httpx.AsyncClient,
    user: str,
    repo: str,
    *,
    create_as_draft: bool,
    keep_pull_request_title_and_body: bool,
    sem: asyncio.Semaphore,
) -> None:
    # Wait for dependency's PR to be ready
    depends_on_pull: github_types.PullRequest | None = None
    if task.depends_on_index is not None:
        dep_task = tasks[task.depends_on_index]
        await dep_task.pull_ready.wait()
        depends_on_pull = dep_task.resolved_pull

    async with sem:
        pull = await create_or_update_pr(
            client,
            user=user,
            repo=repo,
            change=task.change,
            depends_on=depends_on_pull,
            create_as_draft=create_as_draft,
            keep_pull_request_title_and_body=keep_pull_request_title_and_body,
        )
    task.change.pull = pull
    task.resolved_pull = pull
    task.pull_ready.set()


def _log_rebase_performed(
    dest_branch: str,
    remote: str,
    base_branch: str,
    sync_status: sync.SyncStatus,
    decision: approvals_mod.RebaseDecision,
) -> None:
    dropped = (
        f" (dropped {len(sync_status.merged)} merged commit(s))"
        if sync_status.merged
        else ""
    )
    if decision.reason is approvals_mod.RebaseReason.CONFLICT_OVERRIDE:
        numbers = ", ".join(f"#{p['number']}" for p in decision.approved_pulls)
        console.log(
            f"branch `{dest_branch}` rebased on `{remote}/{base_branch}`{dropped} "
            f"(bottom PR has conflicts; approvals on PR(s) {numbers} may be dismissed)",
        )
    elif decision.reason is approvals_mod.RebaseReason.FORCED:
        console.log(
            f"branch `{dest_branch}` rebased on `{remote}/{base_branch}`{dropped} "
            f"(--force-rebase; approvals may be dismissed)",
        )
    else:
        console.log(
            f"branch `{dest_branch}` rebased on `{remote}/{base_branch}`{dropped}",
        )


def _log_rebase_skipped(
    dest_branch: str,
    decision: approvals_mod.RebaseDecision,
) -> None:
    if decision.reason is approvals_mod.RebaseReason.EXPLICIT_SKIP:
        console.log(f"branch `{dest_branch}` rebase skipped (--skip-rebase)")
    elif decision.reason is approvals_mod.RebaseReason.SKIPPED_FOR_APPROVALS:
        n = len(decision.approved_pulls)
        plural = "s" if n != 1 else ""
        verb = "have" if n != 1 else "has"
        console.log(
            f"branch `{dest_branch}` rebase skipped: {n} PR{plural} {verb} approvals "
            f"(use --force-rebase to rebase anyway)",
        )


def _log_rebase_dry_run(
    dest_branch: str,
    remote: str,
    base_branch: str,
    commits_behind: int,
    decision: approvals_mod.RebaseDecision,
) -> None:
    if decision.reason is approvals_mod.RebaseReason.EXPLICIT_SKIP:
        console.log(f"branch `{dest_branch}` rebase skipped (--skip-rebase)")
        return

    if decision.reason is approvals_mod.RebaseReason.SKIPPED_FOR_APPROVALS:
        n = len(decision.approved_pulls)
        plural = "s" if n != 1 else ""
        console.log(
            f"[orange]branch `{dest_branch}` rebase would be skipped: "
            f"approvals detected on {n} PR{plural}[/]",
        )
        for pull in decision.approved_pulls:
            console.log(f'  - PR #{pull["number"]} — "{pull["title"]}"')
        console.log("  Use --force-rebase to rebase anyway.")
        return

    if decision.reason is approvals_mod.RebaseReason.CONFLICT_OVERRIDE:
        numbers = ", ".join(f"#{p['number']}" for p in decision.approved_pulls)
        console.log(
            f"[orange]branch `{dest_branch}` would be rebased on `{remote}/{base_branch}` "
            f"(bottom PR has conflicts; approvals on PR(s) {numbers} would be dismissed)[/]",
        )
        return

    if decision.reason is approvals_mod.RebaseReason.FORCED:
        console.log(
            f"[orange]branch `{dest_branch}` would be rebased on `{remote}/{base_branch}` "
            f"(--force-rebase; approvals may be dismissed)[/]",
        )
        return

    # NO_APPROVALS: preserve existing dry-run behavior (only warn if behind).
    if commits_behind > 0:
        plural = "commit" if commits_behind == 1 else "commits"
        console.log(
            f"[orange]branch `{dest_branch}` is behind `{remote}/{base_branch}` "
            f"by {commits_behind} {plural}, "
            f"commit SHAs will differ after rebase[/]",
        )


# TODO(charly): fix code to conform to linter (number of arguments, local
# variables, statements, positional arguments, branches)
async def stack_push(
    github_server: str,
    token: str,
    *,
    skip_rebase: bool,
    force_rebase: bool = False,
    next_only: bool,
    branch_prefix: str | None,
    dry_run: bool,
    trunk: tuple[str, str],
    create_as_draft: bool = False,
    keep_pull_request_title_and_body: bool = False,
    only_update_existing_pulls: bool = False,
    author: str | None = None,
    revision_history: bool = True,
    no_verify: bool = False,
) -> None:
    os.chdir(await utils.git("rev-parse", "--show-toplevel"))
    dest_branch = await utils.git_get_branch_name()

    if author is None:
        async with utils.get_github_http_client(github_server, token) as client:
            r_author = await client.get("/user")
            author = r_author.json()["login"]

    if branch_prefix is None:
        branch_prefix = await utils.get_default_branch_prefix(author)

    try:
        check_local_branch(branch_name=dest_branch, branch_prefix=branch_prefix)
    except LocalBranchInvalidError as e:
        console_error(e.message)
        console.print(
            "You should run `mergify stack` on the branch you created in the first place",
        )
        sys.exit(ExitCode.INVALID_STATE)

    remote, base_branch = trunk

    user, repo = utils.get_slug(
        await utils.git("config", "--get", f"remote.{remote}.url"),
    )

    if base_branch == dest_branch:
        remote_url = await utils.git("remote", "get-url", remote)
        console_error(
            f"your local branch `{dest_branch}` targets itself: "
            f"`{remote}/{base_branch}` (at {remote_url}@{base_branch})",
        )
        console.print(
            f"You should either fix the target branch or rename your local branch.\n\n"
            f"* To fix the target branch: `git branch {dest_branch} --set-upstream-to={remote}/{base_branch}`\n"
            f"* To rename your local branch: `git branch -M {dest_branch} new-branch-name`",
        )
        sys.exit(ExitCode.INVALID_STATE)

    stack_prefix = f"{branch_prefix}/{dest_branch}" if branch_prefix else dest_branch

    async with utils.get_github_http_client(github_server, token) as client:
        # Always fetch base branch — needed for merge-base and for any eventual rebase.
        await utils.git("fetch", remote, base_branch)
        notes_ref_fetched = await fetch_notes_ref(remote)

        base_commit_sha = await utils.git(
            "merge-base",
            "--fork-point",
            f"{remote}/{base_branch}",
        )
        if not base_commit_sha:
            console_error(
                f"common commit between `{remote}/{base_branch}` and `{dest_branch}` branches not found",
            )
            sys.exit(ExitCode.STACK_NOT_FOUND)

        with console.status("Retrieving latest pushed stacks"):
            remote_changes = await changes.get_remote_changes(
                client,
                user,
                repo,
                stack_prefix,
                author,
            )

        with console.status("Preparing stacked branches..."):
            console.log("Stacked pull request plan:", style="green")
            planned_changes = await changes.get_changes(
                base_commit_sha=base_commit_sha,
                stack_prefix=stack_prefix,
                base_branch=base_branch,
                dest_branch=dest_branch,
                remote_changes=remote_changes,
                only_update_existing_pulls=only_update_existing_pulls,
                next_only=next_only,
            )

        rebase_decision = await approvals_mod.decide_rebase(
            client,
            user,
            repo,
            planned_changes=planned_changes,
            skip_rebase=skip_rebase,
            force_rebase=force_rebase,
        )

        if not dry_run:
            if rebase_decision.should_rebase:
                from mergify_cli.stack import sync as stack_sync_mod

                with console.status(
                    f"Rebasing branch `{dest_branch}` on `{remote}/{base_branch}`...",
                ):
                    sync_status = await stack_sync_mod.smart_rebase(
                        github_server,
                        token,
                        trunk=trunk,
                        branch_prefix=branch_prefix,
                        author=author,
                    )
                _log_rebase_performed(
                    dest_branch,
                    remote,
                    base_branch,
                    sync_status,
                    rebase_decision,
                )
                # Rebase changed local SHAs; recompute base and planned changes.
                base_commit_sha = await utils.git(
                    "merge-base",
                    "--fork-point",
                    f"{remote}/{base_branch}",
                )
                if not base_commit_sha:
                    console_error(
                        f"common commit between `{remote}/{base_branch}` and "
                        f"`{dest_branch}` branches not found after rebase",
                    )
                    sys.exit(ExitCode.STACK_NOT_FOUND)
                planned_changes = await changes.get_changes(
                    base_commit_sha=base_commit_sha,
                    stack_prefix=stack_prefix,
                    base_branch=base_branch,
                    dest_branch=dest_branch,
                    remote_changes=remote_changes,
                    only_update_existing_pulls=only_update_existing_pulls,
                    next_only=next_only,
                )
            else:
                _log_rebase_skipped(dest_branch, rebase_decision)
        else:
            # Dry-run: always compute commits_behind (cheap), then delegate display.
            commits_behind = int(
                await utils.git("rev-list", "--count", f"HEAD..{remote}/{base_branch}"),
            )
            _log_rebase_dry_run(
                dest_branch,
                remote,
                base_branch,
                commits_behind,
                rebase_decision,
            )
            if rebase_decision.should_rebase and commits_behind > 0:
                # If the branch is behind, we know for sure that all the existing
                # pull requests will need to be updated, so we can directly plan
                # them as "update" instead of "skip-up-to-date".
                planned_changes.replace_local_action(
                    old="skip-up-to-date",
                    new="update",
                )

        await read_reasons(planned_changes.locals)

        changes.display_plan(
            planned_changes,
            create_as_draft=create_as_draft,
        )

        if dry_run:
            console.log("[orange]Finished (dry-run mode).[/]")
            sys.exit(ExitCode.SUCCESS)

        if revision_history:
            # Fetch old PR heads for patch-id comparison before force-pushing
            updated_pr_numbers = [
                int(c.pull["number"])
                for c in planned_changes.locals
                if c.action == "update" and c.pull is not None
            ]
            with console.status("Fetching old PR heads for comparison..."):
                try:
                    await fetch_old_pr_heads(remote, updated_pr_numbers)
                except utils.CommandError as exc:
                    # Non-fatal: change type will be "unknown" — but surface
                    # the underlying error so the user can fix it. Escape the
                    # exception text since it can contain `[`/`]` that Rich
                    # would otherwise interpret as markup tags.
                    console.log(
                        f"[orange]Could not fetch old PR heads; revision-history "
                        f"change types will fall back to 'unknown': "
                        f"{rich.markup.escape(str(exc))}[/]",
                    )

            # Detect change types before force-push overwrites refs
            change_types: dict[str, str] = {}
            for change in planned_changes.locals:
                if change.action == "update" and change.pull is not None:
                    change_types[change.id] = await detect_change_type(
                        change.pull_head_sha,
                        change.commit_sha,
                    )

        with console.status("Pushing stacked branches..."):
            await push_branches(
                remote,
                planned_changes.locals,
                no_verify=no_verify,
                notes_ref_fetched=notes_ref_fetched,
            )

        console.log("Updating and/or creating stacked pull requests:", style="green")

        tasks = _build_change_tasks(planned_changes.locals)
        sem = asyncio.Semaphore(MAX_CONCURRENT_API_CALLS)

        await asyncio.gather(
            *(
                _process_change_task(
                    task,
                    tasks,
                    client,
                    user,
                    repo,
                    create_as_draft=create_as_draft,
                    keep_pull_request_title_and_body=keep_pull_request_title_and_body,
                    sem=sem,
                )
                for task in tasks
                if task.change.action in {"create", "update"}
            ),
        )

        for task in tasks:
            console.log(
                task.change.get_log_from_local_change(
                    dry_run=False,
                    create_as_draft=create_as_draft,
                ),
            )

        changes_to_comment = [
            task.change for task in tasks if task.change.pull is not None
        ]

        with console.status("Updating comments..."):
            await create_or_update_comments(
                client,
                user,
                repo,
                changes_to_comment,
                stack_id=dest_branch,
            )

        console.log("[green]Comments updated.[/]")

        if revision_history:
            updated_changes = [
                (task.change, change_types.get(task.change.id, "unknown"))
                for task in tasks
                if task.change.action == "update" and task.change.pull is not None
            ]
            with console.status("Updating revision history..."):
                await create_or_update_revision_comments(
                    client,
                    user,
                    repo,
                    github_server,
                    updated_changes,
                )
            console.log("[green]Revision history updated.[/]")

        with console.status("Deleting unused branches..."):
            if planned_changes.orphans:
                await asyncio.gather(
                    *(
                        delete_stack(client, user, repo, stack_prefix, change)
                        for change in planned_changes.orphans
                    ),
                )

        console.log("[green]Finished.[/]")


@dataclasses.dataclass
class StackComment:
    local_changes: list[changes.LocalChange]

    _STACK_COMMENT_OLD_HEADER: typing.ClassVar[str] = (
        "This pull request is part of a stack:\n"
    )

    STACK_COMMENT_HEADER: typing.ClassVar[str] = (
        "This pull request is part of a [Mergify stack](https://docs.mergify.com/stacks/):\n"
    )

    def _json_marker(
        self,
        current_pull: github_types.PullRequest,
        stack_id: str,
    ) -> str:
        current_number = int(current_pull["number"])
        payload = {
            "schema_version": 1,
            "stack_id": stack_id,
            "pulls": [
                {
                    "number": int(change.pull["number"]),
                    "change_id": change.id,
                    "head_sha": change.commit_sha,
                    "base_branch": change.base_branch,
                    "dest_branch": change.dest_branch,
                    "is_current": int(change.pull["number"]) == current_number,
                }
                for change in self.local_changes
                if change.pull is not None
            ],
        }
        return (
            "<!-- mergify-stack-data: "
            + json.dumps(payload, separators=(",", ":"))
            + " -->"
        )

    def body(
        self,
        current_pull: github_types.PullRequest,
        stack_id: str,
    ) -> str:
        body = self.STACK_COMMENT_HEADER
        body += "\n"
        body += "| # | Pull Request | Link | |\n"
        body += "|--:|---|---|---|\n"

        current_number = int(current_pull["number"])
        row = 0
        for change in self.local_changes:
            if change.pull is None:
                continue
            row += 1
            pull = change.pull
            title = pull["title"].replace("|", "\\|")
            link = f"[#{pull['number']}]({pull['html_url']})"
            status = "👈" if int(pull["number"]) == current_number else ""
            body += f"| {row} | {title} | {link} | {status} |\n"

        body += self._json_marker(current_pull, stack_id) + "\n"
        return body

    @staticmethod
    def is_stack_comment(comment: github_types.Comment) -> bool:
        return comment["body"].startswith(
            StackComment.STACK_COMMENT_HEADER,
        ) or comment["body"].startswith(StackComment._STACK_COMMENT_OLD_HEADER)


_MAX_REASON_LEN = 200


def _escape_reason(reason: str) -> str:
    """Escape a reason for a markdown table cell (no newlines, no pipes)."""
    if not reason:
        return ""
    s = reason.replace("\\", "\\\\").replace("|", "\\|").replace("\n", "<br>")
    if len(s) > _MAX_REASON_LEN:
        s = s[: _MAX_REASON_LEN - 1] + "…"
    return s


@dataclasses.dataclass
class _RevisionEntry:
    number: int
    change_type: str
    old_sha: str | None  # None for "initial"
    new_sha: str
    timestamp: datetime.datetime | None

    @property
    def timestamp_human(self) -> str:
        if self.timestamp is None:
            return ""
        return self.timestamp.strftime("%Y-%m-%d %H:%M UTC")

    @property
    def timestamp_iso(self) -> str | None:
        if self.timestamp is None:
            return None
        return self.timestamp.astimezone(datetime.UTC).strftime("%Y-%m-%dT%H:%M:%SZ")

    reason: str = ""


@dataclasses.dataclass
class RevisionHistoryComment:
    github_server: str
    user: str
    repo: str
    entries: list[_RevisionEntry]
    _raw_rows: list[str] = dataclasses.field(default_factory=list)

    REVISION_COMMENT_FIRST_LINE: typing.ClassVar[str] = "### Revision history\n"

    @staticmethod
    def is_revision_comment(comment: github_types.Comment) -> bool:
        return comment["body"].startswith(
            RevisionHistoryComment.REVISION_COMMENT_FIRST_LINE,
        )

    def _compare_url(self, old_sha: str, new_sha: str) -> str:
        api_url = self.github_server.rstrip("/")
        if "/api/v3" in api_url:
            html_url = api_url.replace("/api/v3", "")
        else:
            html_url = api_url.replace("api.github.com", "github.com")
        return f"{html_url}/{self.user}/{self.repo}/compare/{old_sha}...{new_sha}"

    @classmethod
    def create_initial(
        cls,
        *,
        github_server: str,
        user: str,
        repo: str,
        old_sha: str,
        new_sha: str,
        change_type: str,
        timestamp: datetime.datetime,
        reason: str = "",
    ) -> RevisionHistoryComment:
        entries = [
            _RevisionEntry(1, "initial", None, old_sha, timestamp),
            _RevisionEntry(2, change_type, old_sha, new_sha, timestamp, reason=reason),
        ]
        return cls(
            github_server=github_server,
            user=user,
            repo=repo,
            entries=entries,
        )

    def append(
        self,
        *,
        old_sha: str,
        new_sha: str,
        change_type: str,
        timestamp: datetime.datetime,
        reason: str = "",
    ) -> None:
        next_number = len(self.entries) + 1
        self.entries.append(
            _RevisionEntry(
                next_number,
                change_type,
                old_sha,
                new_sha,
                timestamp,
                reason=reason,
            ),
        )

    def _render_entry(self, entry: _RevisionEntry) -> str:
        if entry.old_sha is None:
            changes_cell = f"`{entry.new_sha[:7]}`"
        else:
            url = self._compare_url(entry.old_sha, entry.new_sha)
            changes_cell = f"[`{entry.old_sha[:7]} \u2192 {entry.new_sha[:7]}`]({url})"
        reason_cell = _escape_reason(entry.reason)
        return (
            f"| {entry.number} | {entry.change_type} | {changes_cell} | "
            f"{reason_cell} | {entry.timestamp_human} |"
        )

    def _json_marker(self, pull_number: int) -> str:
        payload = {
            "schema_version": 1,
            "pull_number": pull_number,
            "entries": [
                {
                    "number": e.number,
                    "change_type": e.change_type,
                    "old_sha": e.old_sha,
                    "new_sha": e.new_sha,
                    "timestamp_iso": e.timestamp_iso,
                    "reason": e.reason,
                    "compare_url": (
                        None
                        if e.old_sha is None
                        else self._compare_url(e.old_sha, e.new_sha)
                    ),
                }
                for e in self.entries
            ],
        }
        return (
            "<!-- mergify-revision-data: "
            + json.dumps(payload, separators=(",", ":"))
            + " -->"
        )

    def body(self, pull_number: int) -> str:
        lines = [
            self.REVISION_COMMENT_FIRST_LINE,
            "| # | Type | Changes | Reason | Date |",
            "|---|------|---------|--------|------|",
        ]
        for i, entry in enumerate(self.entries):
            if i < len(self._raw_rows):
                # Preserve original row verbatim from parsed comment
                lines.append(self._raw_rows[i])
            else:
                lines.append(self._render_entry(entry))
        return "\n".join(lines) + "\n" + self._json_marker(pull_number) + "\n"

    _JSON_MARKER_RE: typing.ClassVar[re.Pattern[str]] = re.compile(
        r"^<!-- mergify-revision-data: (?P<payload>\{.*\}) -->$",
    )

    _ROW_RE_4: typing.ClassVar[re.Pattern[str]] = re.compile(
        r"^\| (\d+) \| (\w+) \| .+ \| (.+) \|$",
    )
    _ROW_RE_5: typing.ClassVar[re.Pattern[str]] = re.compile(
        r"^\| (\d+) \| (\w+) \| .+ \| (.*) \| (.+) \|$",
    )

    @classmethod
    def parse(
        cls,
        body: str,
        *,
        github_server: str,
        user: str,
        repo: str,
    ) -> RevisionHistoryComment | None:
        if not body.startswith(cls.REVISION_COMMENT_FIRST_LINE):
            return None

        entries: list[_RevisionEntry] = []
        raw_rows: list[str] = []
        marker_entries: list[typing.Any] | None = None
        for line in body.splitlines():
            m5 = cls._ROW_RE_5.match(line)
            if m5:
                number = int(m5.group(1))
                change_type = m5.group(2)
                timestamp_str = m5.group(4).strip()
                try:
                    parsed_timestamp: datetime.datetime | None = (
                        datetime.datetime.strptime(
                            timestamp_str,
                            "%Y-%m-%d %H:%M UTC",
                        ).replace(tzinfo=datetime.UTC)
                    )
                except ValueError:
                    parsed_timestamp = None
                entries.append(
                    _RevisionEntry(number, change_type, None, "", parsed_timestamp),
                )
                raw_rows.append(line)
                continue
            m4 = cls._ROW_RE_4.match(line)
            if m4:
                number = int(m4.group(1))
                change_type = m4.group(2)
                timestamp_str = m4.group(3).strip()
                try:
                    parsed_timestamp = datetime.datetime.strptime(
                        timestamp_str,
                        "%Y-%m-%d %H:%M UTC",
                    ).replace(tzinfo=datetime.UTC)
                except ValueError:
                    parsed_timestamp = None
                entries.append(
                    _RevisionEntry(number, change_type, None, "", parsed_timestamp),
                )
                raw_rows.append(line)
                continue
            marker_match = cls._JSON_MARKER_RE.match(line)
            if marker_match:
                try:
                    payload = json.loads(marker_match.group("payload"))
                except json.JSONDecodeError:
                    continue
                if (
                    isinstance(payload, dict)
                    and payload.get("schema_version") == 1
                    and isinstance(payload.get("entries"), list)
                ):
                    marker_entries = payload["entries"]

        if not entries:
            return None

        if marker_entries is not None and len(marker_entries) == len(entries):
            for entry, data in zip(entries, marker_entries, strict=True):
                if not isinstance(data, dict):
                    continue
                if data.get("number") != entry.number:
                    continue
                old_sha = data.get("old_sha")
                new_sha = data.get("new_sha")
                timestamp_iso = data.get("timestamp_iso")
                if isinstance(old_sha, str) or old_sha is None:
                    entry.old_sha = old_sha
                if isinstance(new_sha, str):
                    entry.new_sha = new_sha
                if isinstance(timestamp_iso, str):
                    try:
                        entry.timestamp = datetime.datetime.strptime(
                            timestamp_iso,
                            "%Y-%m-%dT%H:%M:%SZ",
                        ).replace(tzinfo=datetime.UTC)
                    except ValueError:
                        pass
                elif timestamp_iso is None:
                    entry.timestamp = None
                reason = data.get("reason")
                if isinstance(reason, str):
                    entry.reason = reason

        return cls(
            github_server=github_server,
            user=user,
            repo=repo,
            entries=entries,
            _raw_rows=raw_rows,
        )


async def _update_comment_for_pull(
    client: httpx.AsyncClient,
    user: str,
    repo: str,
    pull: github_types.PullRequest,
    stack_comment: StackComment,
    stack_id: str,
    total_pulls: int,
    sem: asyncio.Semaphore,
) -> None:
    new_body = stack_comment.body(pull, stack_id)

    async with sem:
        r = await client.get(
            f"/repos/{user}/{repo}/issues/{pull['number']}/comments",
        )
        comments = typing.cast("list[github_types.Comment]", r.json())
        for comment in comments:
            if StackComment.is_stack_comment(comment):
                if comment["body"] != new_body:
                    await client.patch(comment["url"], json={"body": new_body})
                return

        # NOTE(charly): don't create a stack comment if there is only one
        # pull, it's not a stack
        if total_pulls == 1:
            return

        await client.post(
            f"/repos/{user}/{repo}/issues/{pull['number']}/comments",
            json={"body": new_body},
        )


async def create_or_update_comments(
    client: httpx.AsyncClient,
    user: str,
    repo: str,
    local_changes: list[changes.LocalChange],
    stack_id: str,
) -> None:
    stack_comment = StackComment(local_changes)
    pulls = [c.pull for c in local_changes if c.pull is not None]
    sem = asyncio.Semaphore(MAX_CONCURRENT_API_CALLS)

    await asyncio.gather(
        *(
            _update_comment_for_pull(
                client,
                user,
                repo,
                pull,
                stack_comment,
                stack_id,
                len(pulls),
                sem,
            )
            for pull in pulls
            if not pull["merged_at"]
        ),
    )


async def _update_revision_for_pull(
    client: httpx.AsyncClient,
    user: str,
    repo: str,
    github_server: str,
    change: changes.LocalChange,
    change_type: str,
    timestamp: datetime.datetime,
    sem: asyncio.Semaphore,
) -> None:
    if change.pull is None:
        return

    pull_number = int(change.pull["number"])
    old_sha = change.pull_head_sha
    new_sha = change.commit_sha

    async with sem:
        r = await client.get(
            f"/repos/{user}/{repo}/issues/{pull_number}/comments",
        )
        comments = typing.cast("list[github_types.Comment]", r.json())

        for comment in comments:
            if RevisionHistoryComment.is_revision_comment(comment):
                parsed = RevisionHistoryComment.parse(
                    comment["body"],
                    github_server=github_server,
                    user=user,
                    repo=repo,
                )
                if parsed is not None:
                    parsed.append(
                        old_sha=old_sha,
                        new_sha=new_sha,
                        change_type=change_type,
                        timestamp=timestamp,
                        reason=change.reason,
                    )
                    new_body = parsed.body(pull_number)
                    if comment["body"] != new_body:
                        await client.patch(
                            comment["url"],
                            json={"body": new_body},
                        )
                    return
                # Comment header matched but body corrupted — overwrite it
                revision = RevisionHistoryComment.create_initial(
                    github_server=github_server,
                    user=user,
                    repo=repo,
                    old_sha=old_sha,
                    new_sha=new_sha,
                    change_type=change_type,
                    timestamp=timestamp,
                    reason=change.reason,
                )
                await client.patch(
                    comment["url"],
                    json={"body": revision.body(pull_number)},
                )
                return

        # No existing revision comment — create one
        revision = RevisionHistoryComment.create_initial(
            github_server=github_server,
            user=user,
            repo=repo,
            old_sha=old_sha,
            new_sha=new_sha,
            change_type=change_type,
            timestamp=timestamp,
            reason=change.reason,
        )
        await client.post(
            f"/repos/{user}/{repo}/issues/{pull_number}/comments",
            json={"body": revision.body(pull_number)},
        )


async def create_or_update_revision_comments(
    client: httpx.AsyncClient,
    user: str,
    repo: str,
    github_server: str,
    updated_changes: list[tuple[changes.LocalChange, str]],
) -> None:
    if not updated_changes:
        return

    sem = asyncio.Semaphore(MAX_CONCURRENT_API_CALLS)
    now = datetime.datetime.now(datetime.UTC)

    await asyncio.gather(
        *(
            _update_revision_for_pull(
                client,
                user,
                repo,
                github_server,
                change,
                change_type,
                now,
                sem,
            )
            for change, change_type in updated_changes
        ),
    )


async def delete_stack(
    client: httpx.AsyncClient,
    user: str,
    repo: str,
    stack_prefix: str,
    change: changes.OrphanChange,
) -> None:
    if change.pull is not None:
        branch_ref = change.pull["head"]["ref"]
    else:
        branch_ref = f"{stack_prefix}/{change.id}"
    await client.delete(
        f"/repos/{user}/{repo}/git/refs/heads/{branch_ref}",
    )
    console.log(change.get_log_from_orphan_change(dry_run=False))


async def create_or_update_pr(
    client: httpx.AsyncClient,
    *,
    user: str,
    repo: str,
    change: changes.LocalChange,
    depends_on: github_types.PullRequest | None,
    create_as_draft: bool,
    keep_pull_request_title_and_body: bool,
) -> github_types.PullRequest:
    if change.action == "update":
        if change.pull is None:
            msg = "Can't update pull with change.pull unset"
            raise RuntimeError(msg)

        pull_changes: dict[str, typing.Any] = {
            "head": change.dest_branch,
            "base": change.base_branch,
        }
        if keep_pull_request_title_and_body:
            pull_changes["body"] = format_pull_description(
                change.pull["body"] or "",
                depends_on,
            )
        else:
            pull_changes["title"] = change.title
            pull_changes["body"] = format_pull_description(
                change.message,
                depends_on,
            )

        await client.patch(
            f"/repos/{user}/{repo}/pulls/{change.pull['number']}",
            json=pull_changes,
        )
        return change.pull

    if change.action == "create":
        r = await client.post(
            f"/repos/{user}/{repo}/pulls",
            json={
                "title": change.title,
                "body": format_pull_description(change.message, depends_on),
                "draft": create_as_draft,
                "head": change.dest_branch,
                "base": change.base_branch,
            },
        )
        return typing.cast("github_types.PullRequest", r.json())

    msg = f"Unhandled action: {change.action}"
    raise RuntimeError(msg)