# 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."""

import base64
import json
import re
from typing import Any, Generator, Literal, Sequence

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

_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


ChangeState = Literal['NEW', 'MERGED', 'ABANDONED']


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

    ChangeState = ChangeState
    Change = Change

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

    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,
    ) -> tuple[Sequence[Change], Sequence[Change]]:
        """Recursively resolve dependencies of a CL.

        Resolve dependencies of a CL by parsing its patches.json file.

        Args:
            gerrit_name (str): Name of Gerrit host of initial change.
            number (int|str): Gerrit number or git commit hash of initial
                change.

        Returns a tuple of 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:
            if not self._patches_json_enabled:
                with self.m.step.nest('patches.json processing disabled'):
                    return ((), ())

            with self.m.context(infra_steps=True):
                if isinstance(number, str):
                    cl = self._lookup_cl(gerrit_name, number)
                    if not cl:
                        return ((), ())
                    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)

                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

                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 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]