insights/products/signals/backend/implementation_pr.py

"""Resolve implementation PR URLs linked to signal reports."""

from dataclasses import dataclass
from typing import Literal, cast

import structlog

from insights.models.github_integration_base import GitHubIntegrationBase
from insights.models.integration import GitHubIntegration

from products.signals.backend.models import SignalReport
from products.signals.backend.task_run_artefacts import SIGNALS_PRODUCT, TASK_RUN_TYPE_IMPLEMENTATION
from products.tasks.backend.facade import api as tasks_facade

logger = structlog.get_logger(__name__)


@dataclass(frozen=True)
class ImplementationPr:
    """The one implementation PR surfaced for a report, and whether the webhook saw it merge."""

    url: str
    merged: bool


def fetch_implementation_pr_state_for_reports(report_ids: list[str]) -> dict[str, ImplementationPr]:
    """The implementation PR surfaced for each report, with its merge state, when one exists.

    The task↔report association comes from `SignalReport.associated_task_runs_for_reports` (the
    unified view of the `task_run` artefact log + legacy gate rows, batched over the whole page);
    the facade then resolves the latest PR-bearing run for each task, so multiple runs of a task
    collapse to the newest PR.

    A report can be associated with several implementation tasks (retries), but only one PR is
    surfaced — the first associated task that has one. The merge flag is read from *that* task, so
    the URL and its state always describe the same PR. Reading them independently would let a
    retry's merged PR vouch for a different PR's URL.
    """
    if not report_ids:
        return {}

    # (report_id, task_id) for each report's implementation task(s); signals owns this mapping.
    # Batched across the whole page so association costs two queries, not two per report (N+1).
    runs_by_report = SignalReport.associated_task_runs_for_reports(
        report_ids=[str(report_id) for report_id in report_ids],
        product=SIGNALS_PRODUCT,
        type=TASK_RUN_TYPE_IMPLEMENTATION,
    )
    pairs: list[tuple[str, str]] = [
        (report_id, run.task_id) for report_id, runs in runs_by_report.items() for run in runs
    ]
    if not pairs:
        return {}

    task_ids = [task_id for _, task_id in pairs]
    pr_url_by_task = tasks_facade.get_latest_pr_url_by_task(task_ids)
    merged_task_ids = tasks_facade.get_merged_pr_task_ids(task_ids)

    result: dict[str, ImplementationPr] = {}
    for report_id, task_id in pairs:
        pr_url = pr_url_by_task.get(task_id)
        if pr_url and report_id not in result:
            result[report_id] = ImplementationPr(url=pr_url, merged=task_id in merged_task_ids)
    return result


def fetch_implementation_pr_urls_for_reports(report_ids: list[str]) -> dict[str, str]:
    """PR URL from the latest implementation task run for each report, when available."""
    return {report_id: pr.url for report_id, pr in fetch_implementation_pr_state_for_reports(report_ids).items()}


PrCloseReason = Literal["suppressed", "snoozed"]

# Left on the PR before it's closed, so anyone looking at the PR sees why it was closed and how to undo it.
_PR_CLOSE_COMMENT_TEMPLATE = (
    "🔕 Closing this PR because the linked Insights report was {action}.\n\n"
    "If that wasn't intended, restore the report in Insights and reopen this PR."
)
_PR_CLOSE_COMMENTS: dict[PrCloseReason, str] = {
    reason: _PR_CLOSE_COMMENT_TEMPLATE.format(action=reason)
    for reason in cast(tuple[PrCloseReason, ...], ("suppressed", "snoozed"))
}


def close_implementation_pr_for_report(
    team_id: int,
    report_id: str,
    *,
    reason: PrCloseReason = "suppressed",
) -> bool:
    """Best-effort: comment on and close the GitHub PR opened for this report's implementation task.

    Called when a report is suppressed or snoozed — the open PR shouldn't linger. Only acts on a PR
    that is still open: an already-closed or merged PR is left untouched (no comment, no close), so
    we never leave a confusing "closing this PR" note on a PR that shipped months ago. Leaves an
    explanatory comment, then closes the PR. Returns True when the PR was closed, False when there
    was nothing to close or the close couldn't be completed. Never raises: the state transition
    must succeed regardless.
    """
    try:
        pr_url = fetch_implementation_pr_urls_for_reports([str(report_id)]).get(str(report_id))
        if not pr_url:
            return False

        parsed = GitHubIntegrationBase.parse_pull_request_url(pr_url)
        if parsed is None:
            logger.warning("close_implementation_pr_unparseable_url", report_id=str(report_id), pr_url=pr_url)
            return False

        github = GitHubIntegration.first_for_team_repository(team_id, parsed.repository)
        if github is None:
            logger.info(
                "close_implementation_pr_no_integration", report_id=str(report_id), repository=parsed.repository
            )
            return False

        # Only comment on and close a PR that's still open. A merged PR reports state "closed" with
        # merged=True, and an already-closed PR reports state "closed" — in either case there's
        # nothing to close, and leaving a comment would just be noise. If the state can't be
        # confirmed, skip rather than risk commenting on a PR that already shipped.
        pr_status = github.get_pull_request(parsed.repository, parsed.number)
        if not pr_status.get("success"):
            logger.warning(
                "close_implementation_pr_status_fetch_failed",
                report_id=str(report_id),
                pr_url=pr_url,
                error=pr_status.get("error"),
                status_code=pr_status.get("status_code"),
            )
            return False
        if pr_status.get("state") != "open" or pr_status.get("merged"):
            logger.info(
                "close_implementation_pr_not_open",
                report_id=str(report_id),
                pr_url=pr_url,
                state=pr_status.get("state"),
                merged=pr_status.get("merged"),
            )
            return False

        # Explain first, close second — a failed comment shouldn't stop the close.
        comment_outcome = github.comment_on_pull_request(parsed.repository, parsed.number, _PR_CLOSE_COMMENTS[reason])
        if not comment_outcome.get("success"):
            logger.warning(
                "close_implementation_pr_comment_failed",
                report_id=str(report_id),
                pr_url=pr_url,
                error=comment_outcome.get("error"),
                status_code=comment_outcome.get("status_code"),
            )

        outcome = github.close_pull_request(parsed.repository, parsed.number)
        if not outcome.get("success"):
            logger.warning(
                "close_implementation_pr_failed",
                report_id=str(report_id),
                pr_url=pr_url,
                error=outcome.get("error"),
                status_code=outcome.get("status_code"),
            )
            return False
        return True
    except Exception:
        logger.exception("close_implementation_pr_unexpected_error", report_id=str(report_id))
        return False