# Copyright 2026 The Pigweed Authors
#
# 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
#
#     https://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.
"""Check if the current tryjob just passed on the commit being rolled."""

from __future__ import annotations

import contextlib
import dataclasses
import datetime
import re
from collections.abc import Sequence

from recipe_engine import engine_types, recipe_api

from PB.go.chromium.org.luci.buildbucket.proto import (
    builder_common as builder_common_pb,
)
from PB.go.chromium.org.luci.buildbucket.proto import (
    builds_service as builds_service_pb,
)
from PB.go.chromium.org.luci.buildbucket.proto import common as common_pb
from PB.recipe_engine import result as result_pb

LUCI_PROJECT_ACCOUNTS = 'luci-project-accounts.iam.gserviceaccount.com'


@dataclasses.dataclass
class Change:
    gerrit_name: str
    number: int
    commit_message: str
    uploader: str


class RollTryjobReuseApi(recipe_api.RecipeApi):
    """Check if the current tryjob just passed on the commit being rolled."""

    Change = Change

    def __init__(self, props, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._dry_run = props.dry_run
        self._enabled = props.enabled
        self._max_age = datetime.timedelta(hours=props.max_age_hours)
        self._verbose = props.verbose

    def exit_early_if_applicable(
        self,
        changes: Sequence[Change],
        *,
        verbose: bool = False,
    ) -> result_pb.RawResult | None:
        try:
            return self._exit_early_if_applicable(
                changes=changes,
                verbose=verbose,
            )

        except Exception as e:  # pragma: no cover
            # If there's an error in this module we should just ignore it and
            # run the full build.
            pres = self.m.step.empty('roll tryjob reuse exception').presentation
            pres.step_summary_text = repr(e)
            pass

    def _exit_early_if_applicable(
        self,
        changes: Sequence[Change],
        *,
        verbose: bool = False,
    ) -> result_pb.RawResult | None:
        nest = lambda: self.m.step.nest('checking roll tryjob status')
        verbose = verbose or self._verbose

        with contextlib.ExitStack() as stack:
            checking_pres: engine_types.StepPresentation | None = None
            if verbose:
                checking_pres = stack.enter_context(nest())

            if not self._enabled:
                if verbose:
                    self.m.step.empty('disabled')
                    checking_pres.step_summary_text = 'disabled'
                return

            if not self.m.buildbucket_util.is_tryjob:
                if verbose:
                    self.m.step.empty('not a tryjob')
                    checking_pres.step_summary_text = 'not a tryjob'
                return

            if self.m.recipe_testing.enabled:
                if verbose:
                    self.m.step.empty('in recipe testing')
                    checking_pres.step_summary_text = 'in recipe testing'
                return

            if len(changes) != 1:
                if verbose:
                    num_changes = f'num changes ({len(changes)}) != 1'
                    self.m.step.empty(num_changes)
                    checking_pres.step_summary_text = num_changes
                return

            roll_change = changes[0]
            if '@' not in (roll_change.uploader or ''):
                if verbose:
                    self.m.step.empty(f'bad uploader: {roll_change.uploader!r}')
                return
            domain = roll_change.uploader.split('@', 1)[1]
            if domain not in (
                'pigweed-service-accounts.iam.gserviceaccount.com',
                'pw-internal-service-accounts.iam.gserviceaccount.com',
            ):
                if verbose:
                    not_roller = (
                        f'{roll_change.uploader} is not roller service account'
                    )
                    self.m.step.empty(not_roller)
                    checking_pres.step_summary_text = not_roller
                return

            if not verbose:
                checking_pres = stack.enter_context(nest())

            original_gerrit_change_url: str | None = None
            original_gerrit_change: common_pb.GerritChange | None = None
            for line in roll_change.commit_message.splitlines():
                if match := re.match(r'^Original-Reviewed-on: (.*)\s*$', line):
                    if verbose:
                        pres = self.m.step.empty(
                            'found Original-Reviewed-on line'
                        ).presentation
                        pres.step_summary_text = repr(line)

                    original_gerrit_change_url = match.group(1)
                    original_gerrit_change = self.m.gerrit.parse_change_url(
                        original_gerrit_change_url,
                    )
                    original_gerrit_change.host = (
                        original_gerrit_change.host.replace(
                            '.git.corp.google.com',
                            '.googlesource.com',
                        )
                    )
                    break

            # If we don't break out of the previous for loop then we don't have
            # tryjobs to compare with so we return.
            if not original_gerrit_change:
                could_not_find = 'could not find "Original-Reviewed-on:" line'
                self.m.step.empty(could_not_find)
                checking_pres.step_summary_text = could_not_find
                return

            details = self.m.gerrit.change_details(
                name=str(original_gerrit_change.change),
                change_id=original_gerrit_change.change,
                host=original_gerrit_change.host,
            ).json.output
            submitter = details['submitter']['email']
            submitted_patch = details['current_revision_number']

            if submitter.split('@', 1)[1] != LUCI_PROJECT_ACCOUNTS:
                not_scoped = (
                    'submitter is not a LUCI project-scoped service account'
                )
                self.m.step.empty(not_scoped)
                checking_pres.step_summary_text = not_scoped
                return

            # Since we got an "Original-Reviewed-on:" line, the original change
            # had a "Reviewed-on:" line. This is added by Gerrit on submission.
            # There should be no tryjobs on this last patchset since it didn't
            # exist until submission. Look at the prior patchset, but in case
            # something weird happened make sure we don't go below 1.
            original_gerrit_change.patchset = max(submitted_patch - 1, 0)
            builder_id = builder_common_pb.BuilderID(
                project=self.m.buildbucket.build.builder.project,
                bucket=self.m.buildbucket.build.builder.bucket.removesuffix(
                    '.shadow',
                ),
                builder=self.m.buildbucket.build.builder.builder,
            )
            predicate = builds_service_pb.BuildPredicate(
                builder=builder_id,
                status=common_pb.SUCCESS,
                gerrit_changes=[original_gerrit_change],
                experiments=['-pigweed.non_production'],
            )
            predicate.create_time.start_time.FromDatetime(
                self.m.time.utcnow() - self._max_age,
            )
            builds = self.m.buildbucket.search(predicate=predicate)

            if not builds:
                self.m.step.empty('no recent passing builds')
                checking_pres.step_summary_text = 'no recent passing builds'
                return

            full_name = '/'.join(
                (builder_id.project, builder_id.bucket, builder_id.builder),
            )

            pres = self.m.step.empty(
                'found recent passing builds',
            ).presentation
            for build in builds:
                pres.links[str(build.id)] = self.m.buildbucket.build_url(
                    build=build,
                )

            builds = [
                b
                for b in builds
                if 'early_exit_reason' not in b.output.properties
            ]
            if not builds:
                self.m.step.empty('all builds exited early')
                checking_pres.step_summary_text = 'all builds exited early'
                return

            if self._dry_run:
                self.m.step.empty('dry-run, continuing build')
                checking_pres.step_summary_text = 'dry-run, continuing build'
                return

            assert checking_pres
            checking_pres.step_summary_text = (
                'Exiting early since this tryjob passed on the rolling change '
                'immediately before the roll.'
            )
            prior_tryjob_link = self.m.buildbucket.build_url(build=builds[0])
            checking_pres.links[full_name] = prior_tryjob_link
            checking_pres.properties['early_exit_reason'] = 'roll tryjob reuse'
            checking_pres.properties['rolling change'] = (
                original_gerrit_change_url
            )

            return result_pb.RawResult(
                summary_markdown=(
                    f'Exiting early since [{full_name}]({prior_tryjob_link}) '
                    'passed on the '
                    f'[rolling change]({original_gerrit_change_url}) '
                    'immediately before the roll.'
                ),
                status=common_pb.SUCCESS,
            )