# Copyright 2020 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.
"""Find dependencies on pending CLs."""

from __future__ import annotations

import base64
import json
import re
from typing import TYPE_CHECKING

import attrs
from PB.recipe_modules.pigweed.cq_deps.properties import InputProperties
from recipe_engine import recipe_api

if TYPE_CHECKING:  # pragma: no cover
    from typing import Any, Generator, Literal, Sequence

_PATCHES_FILE = 'patches.json'


@attrs.define
class Change:
    """Details of a matching change.

    This is meant to match the GerritChange message defined at
    https://source.chromium.org/chromium/infra/infra/+/HEAD:go/src/go.chromium.org/luci/buildbucket/proto/common.proto?q=GerritChange
    A few fields have been added for use by this module.
    """

    gerrit_name: str
    change: int = attrs.field(converter=int)
    project: str | None = None
    patchset: str | None = None
    status: str | None = None
    commit: str | None = None

    @property
    def host(self) -> str:
        return f'{self.gerrit_name}-review.googlesource.com'

    @property
    def link(self) -> str:
        return f'https://{self.host}/{self.change}'

    @property
    def name(self) -> str:
        return f'{self.gerrit_name}:{self.change}'

    @property
    def remote(self) -> str:
        return f'https://{self.gerrit_name}.googlesource.com/{self.project}'

    @property
    def gerrit_url(self) -> str:
        return f'https://{self.gerrit_name}-review.googlesource.com/c/{self.change}'

    def __str__(self) -> str:
        return self.name


@attrs.define
class CqDepsResult:
    """Result of resolve(), containing references that were found or not."""

    resolved: tuple[Change]
    unresolved: tuple[Change]


if TYPE_CHECKING:  # pragma: no cover
    ChangeState = Literal['NEW', 'MERGED', 'ABANDONED']


class CqDepsApi(recipe_api.RecipeApi):
    """Find dependencies on pending CLs."""

    if TYPE_CHECKING:  # pragma: no cover
        ChangeState = ChangeState
    Change = Change
    Result = CqDepsResult

    def __init__(self, props: InputProperties, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._details = {}
        self._patches_json_enabled: bool = not props.patches_json_disabled
        self._topics_enabled: bool = props.topics_enabled

    def _lookup_cl(
        self, gerrit_name: str, commit: str
    ) -> list[dict[str, Any]] | None:
        query_results = self.m.gerrit.change_query(
            f'number {commit}',
            f'commit:{commit}',
            host=f'{gerrit_name}-review.googlesource.com',
            max_attempts=2,
            timeout=30,
            # The strange value for _number is to ensure different hashes result
            # in different CL numbers by default. This way we don't accidentally
            # create a cycle by specifying a fixed default value.
            test_data=self.m.json.test_api.output(
                [
                    {
                        '_number': sum(ord(x) for x in commit),
                    }
                ]
            ),
        ).json.output
        if query_results:
            return query_results[0]
        return None

    def resolve(
        self,
        gerrit_name: str,
        number: int | str,
        topic: str | None = None,
    ) -> CqDepsResult:
        """Recursively resolve dependencies of a change.

        Resolve dependencies of a change. Depending on the module properties,
        this might be by parsing a 'patches.json' file in the change or by
        looking up other changes in the same topic.

        The two systems are evaluated independently. Topics of patches found via
        'patches.json' are not evaluated, and 'patches.json' files on other
        changes in the same topic are not evaluated.

        Args:
            gerrit_name: Name of Gerrit host of initial change.
            number: Gerrit number or git commit hash of initial change.
            topic: Gerrit topic containing this the change, if any.

        Returns resolved Change objects (excluding the given change) and
        unresolved Change objects (which are not completely populated).
        """

        with self.m.step.nest('resolve CL deps', status='last') as pres:
            with self.m.context(infra_steps=True):
                if isinstance(number, str):
                    cl = self._lookup_cl(gerrit_name, number)
                    if not cl:
                        return CqDepsResult((), ())
                    number: int = cl['_number']

                seen: set[str] = set()
                deps: dict[str, Change] = {}
                unresolved: list[str] = []
                initial_change = Change(gerrit_name, number)
                to_process: list[str] = [initial_change]

                # Want all initial details calls to come from this step level so
                # their step names are predictable. Further calls can be made
                # anywhere and will hit the cache.
                _ = self._change_details(initial_change)

                if self._patches_json_enabled:
                    while to_process:
                        current: Change = to_process.pop()
                        deps[current.name] = current
                        seen.add(current.name)

                        for dep in self._resolve(current):
                            if dep.name not in deps:
                                details = self._change_details(dep)
                                if not details:
                                    unresolved.append(dep)
                                    continue
                                if details['status'] != 'NEW':
                                    continue

                                if dep.name not in seen:
                                    to_process.append(dep)
                                deps[dep.name] = dep

                if not topic:
                    self.m.step.empty('no topic')

                else:
                    if self._topics_enabled:
                        assert '"' not in topic, f'invalid topic: {topic!r}'
                        changes = self.m.gerrit.change_query(
                            f'topic {topic}',
                            f'topic:"{topic}" status:new',
                            host=f'{gerrit_name}-review.googlesource.com',
                            max_attempts=2,
                            timeout=30,
                            # Always include the current change as well as the
                            # "next" one. We would only get here if there was a
                            # topic set, so always assume there's another change
                            # in the topic in testing.
                            test_data=self.m.json.test_api.output(
                                [
                                    {
                                        '_number': number,
                                    },
                                    {
                                        '_number': number + 1,
                                    },
                                ]
                            ),
                        ).json.output

                        for change in changes:
                            dep = Change(gerrit_name, change['_number'])
                            deps[dep.name] = dep

                    else:
                        self.m.step.empty('topic set but topics not enabled')

                if initial_change.name in deps:
                    del deps[initial_change.name]

                for dep in deps.values():
                    pres.links[dep.name] = dep.link

                # Make sure the last step passes because _change_details might
                # fail for an individual CL in ways we don't care about.
                with self.m.step.nest('pass'):
                    pass

                return CqDepsResult(deps.values(), unresolved)

    def _parse_deps(self, change: Change) -> Generator[Change, None, None]:
        """Parse dependencies from patches.json in a change.

        For a change with the following patches.json file,

            [
                {"gerrit_name": "pigweed", "number": 1001},
                {"gerrit_name": "pigweed", "number": 1002},
                {"gerrit_name": "other", "number": 1003}
            ]

        the return value would look like

            [
                Change("pigweed", 1001),
                Change("pigweed", 1002),
                Change("other", 1003),
            ]

        Args:
            change (Change): Change object from which to retrieve dependencies

        Returns a list of direct dependencies of the given change.
        """

        details = self._change_details(change)
        if not details:
            return []  # pragma: no cover
        host = self.m.gerrit.normalize_host(change.host)
        host = host.replace('-review.', '.')
        current_revision = details['revisions'][details['current_revision']]

        # If there is no _PATCHES_FILE in the CL, don't try to retrieve it.
        if _PATCHES_FILE not in current_revision.get('files', ()):
            return []

        # If _PATCHES_FILE is being deleted, don't try to retrieve it.
        if current_revision['files'][_PATCHES_FILE].get('status', 'M') == 'D':
            return []  # pragma: no cover

        cl = details['_number']
        ps = current_revision['_number']

        raw_data = self.m.gitiles.fetch(
            host,
            details['project'],
            _PATCHES_FILE,
            'refs/changes/{:02}/{}/{}'.format(cl % 100, cl, ps),
            step_name=f'fetch {_PATCHES_FILE}',
        )

        if not raw_data:
            raise self.m.step.InfraFailure(
                f'{_PATCHES_FILE} is in CL but failed to retrieve it'
            )

        deps = json.loads(raw_data)

        for dep in deps:
            yield Change(dep['gerrit_name'], dep['number'])

    def _resolve(self, change: Change) -> list[Change]:
        """Resolve dependencies of the given change.

        Retrieve and parse the patches.json file of the given change. Does not
        attempt to verify if these dependent changes are accessible or even
        valid.

        Args:
            change (Change): Change to resolve dependencies for.

        Returns a list of changes on which the given change depends.
        """

        with self.m.step.nest(f'resolve deps for {change}') as pres:
            deps = []
            for dep in self._parse_deps(change):
                deps.append(dep)
                pres.links[dep.name] = dep.link
            if not deps:
                pres.step_summary_text = 'no dependencies'
            return deps

    def _change_details(
        self, change: Change, *, test_status: ChangeState = 'NEW'
    ) -> dict[str, Any]:
        if change.name in self._details:
            return self._details[change.name]

        try:
            step = self.m.gerrit.change_details(
                f'details {change}',
                change_id=str(change.change),
                host=change.host,
                query_params=[
                    'CURRENT_COMMIT',
                    'CURRENT_FILES',
                    'CURRENT_REVISION',
                ],
                max_attempts=2,
                timeout=30,
                test_data=self.m.json.test_api.output(
                    {
                        '_number': change.change,
                        'current_revision': 'HASH',
                        'revisions': {
                            'HASH': {
                                '_number': 1,
                                'commit': {'message': ''},
                                'files': {
                                    'filename': {},
                                },
                            },
                        },
                        'project': 'project',
                        'status': test_status,
                    }
                ),
            )
            details: dict[str, Any] = step.json.output or {}

        except self.m.step.StepFailure:
            details: dict[str, Any] = {}

        self._details[change.name] = details
        if details:
            change.project = details['project']
            current_revision = details['revisions'][details['current_revision']]
            change.patchset = current_revision['_number']
            change.status = details['status']
            change.commit = details['current_revision']

        return self._details[change.name]