# Copyright 2025 The Fuchsia Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

import abc
import argparse
import configparser
import shlex
import subprocess
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Callable, Dict, List, Optional, TextIO, Tuple, Type


class Context:
    """Holds the context for the git polyfill execution."""

    def __init__(self, args: "ArgsCollection"):
        self.args = args
        self.git_subcommand_args: Optional[argparse.Namespace] = None

        # Polyfill arguments
        polyfill_parser = _create_polyfill_parser()
        self.polyfill_options = polyfill_parser.parse_args(args.polyfill_args)

        self.real_git: str = self.polyfill_options.real_git
        self.invoker_cwd: str = self.polyfill_options.invoker_cwd
        self.repository_root: Path = Path(self.polyfill_options.repository_root)
        self.log_file: Optional[str] = self.polyfill_options.log_file

        self._log_file_path = Path(self.log_file) if self.log_file else None

        # Global git arguments
        global_git_parser = _create_global_git_args_parser()
        self.global_git_options, _ = global_git_parser.parse_known_args(
            args.global_git_args
        )

    def write_log(self, level: str, message: str) -> None:
        if self._log_file_path:
            with self._log_file_path.open("a") as f:
                f.write(f"[{level}] {message}\n")

    def print(self, message: str, file: Optional[TextIO] = sys.stdout) -> None:
        """Prints a message and logs it."""
        print(message, file=file)
        self.write_log("PRINT", message)

    def error(self, message: str) -> None:
        """Prints an error message to stderr and logs it."""
        print(message, file=sys.stderr)
        self.write_log("ERROR", message)

    def fatal(self, message: str) -> None:
        """Prints a fatal error message to stderr and logs it."""
        print(f"fatal: {message}", file=sys.stderr)
        self.write_log("FATAL", message)

    def log_info(self, message: str) -> None:
        """Logs an info message to the log."""
        self.write_log("INFO", message)

    def output(self, message: str, end: str = "") -> None:
        """Prints output to stdout and logs it."""
        sys.stdout.write(message)
        if end:
            sys.stdout.write(end)
        self.write_log("OUTPUT", message.strip())

    def get_repository_root(self) -> Path:
        """Returns the repository root."""
        return self.repository_root

    def get_relative_path(self) -> str:
        """Returns the relative path to the repository root that we should be using based on the args."""
        invoker_cwd = Path(self.invoker_cwd)
        args = self.global_git_options

        if args.C and args.git_dir:
            raise ValueError("Cannot use both -C and --git-dir")

        path: Path
        if args.C:
            path = Path(args.C)
        elif args.git_dir:
            path = Path(args.git_dir)
            if path.name != ".git":
                raise ValueError(f"git_dir must end in .git, got {path}")
            path = path.parent
        else:
            path = invoker_cwd

        if not path.is_absolute():
            path = invoker_cwd / path

        return str(path.relative_to(self.repository_root))

    def run_real_git(self, args: List[str], cwd: Optional[str] = None) -> str:
        """Runs the real git command with the given arguments."""
        full_command = [self.real_git] + args
        self.write_log("EXEC", shlex.join(full_command))

        result = subprocess.run(
            full_command,
            capture_output=True,
            text=True,
            cwd=cwd,
            check=True,
        )

        return result.stdout


def get_workspace_id_and_snapshot_version(
    context: Context,
) -> Tuple[str, int]:
    """Returns the workspace id and snapshot version for the given repository root.

    Args:
        context: The execution context.

    Returns:
        A tuple of the workspace id and snapshot version.
    """
    citc_dir = context.repository_root.parent / ".citc"
    try:
        workspace_id = (citc_dir / "workspace_id").read_text().strip()
        snapshot_version = (citc_dir / "snapshot_version").read_text().strip()
    except Exception as e:
        context.fatal(f"could not read citc metadata: {e}")
        return "", 0
    return workspace_id, int(snapshot_version)


def get_repo_root_for_repo(repo_path: str) -> str:
    """Returns the repo root string expected by the backend.

    Args:
        repo_path: The relative path of the repo (e.g. "" or "sub/module").

    Returns:
        The full repo root string (e.g. "fuchsia" or "fuchsia/sub/module").
    """
    if not repo_path:
        return "fuchsia"
    return f"fuchsia/{repo_path}"


def get_submodule_paths(repository_path: Path) -> List[str]:
    """Returns a list of submodule paths from the .gitmodules file.

    This function returns a list of submodule paths from the .gitmodules file.

    Args:
        repository_path: The path to the repository. This path should be an
        absolute path.

    Returns:
        A list of submodule paths.
    """

    gitmodules_path = repository_path / ".gitmodules"
    paths: List[str] = []
    if not gitmodules_path.exists():
        return paths

    config = configparser.ConfigParser()
    try:
        config.read(gitmodules_path)
        for section in config.sections():
            if "path" in config[section]:
                paths.append(config[section]["path"])
    except configparser.Error:
        # If we can't parse the file, we assume there are no submodules.
        pass
    return paths


def get_target_repository_at_path(
    relative_path: str, repository_root: Path
) -> str:
    """Returns the relative path to the repository root for the target path.

    Args:
        relative_path: The relative path to the root directory.
        repository_root: The root of the repository.

    Returns:
        The relative path to the submodule from repository_root, or "" if it is the main repository.
    """
    current_repo_root = repository_root
    target_full_path = repository_root / relative_path

    # Ensure target_full_path is inside repository_root
    try:
        target_full_path.relative_to(repository_root)
    except ValueError:
        # If the path is outside the repo root, assume main repo.
        return ""

    while True:
        # Check for submodules in the current repo root
        submodule_paths = get_submodule_paths(current_repo_root)

        found_submodule = False
        for submodule_path_str in submodule_paths:
            submodule_path = current_repo_root / submodule_path_str

            # Check if target_full_path is inside this submodule
            try:
                target_full_path.relative_to(submodule_path)
                # If it is inside, update current_repo_root and break to outer loop to continue searching deeper
                current_repo_root = submodule_path
                found_submodule = True
                break
            except ValueError:
                continue

        if not found_submodule:
            # We are in the deepest submodule containing the target path
            break

    if current_repo_root == repository_root:
        return ""

    return str(current_repo_root.relative_to(repository_root))


class GitSubCommand(abc.ABC):
    """Abstract base class for git subcommands."""

    # The command name which gets printed in help messages.
    _command_name: str

    def add_arguments(self, parser: argparse.ArgumentParser) -> None:
        """Override to add subcommand-specific arguments."""

    @abc.abstractmethod
    def execute(self, context: Context) -> int:
        """Executes the command.

        Args:
            context: The execution context.

        Returns:
            The exit code (0 for success, non-zero for failure).
        """

    def run(self, context: Context) -> int:
        """Parses arguments and executes the command."""
        parser = argparse.ArgumentParser(
            prog=f"git {getattr(self, '_command_name')}"
        )
        self.add_arguments(parser)
        context.git_subcommand_args = parser.parse_args(
            context.args.remaining_args
        )
        return self.execute(context)


_COMMANDS: Dict[str, Type[GitSubCommand]] = {}


def register_command(
    name: str,
) -> Callable[[Type[GitSubCommand]], Type[GitSubCommand]]:
    """Decorator to register a GitSubCommand implementation."""

    def decorator(cls: Type[GitSubCommand]) -> Type[GitSubCommand]:
        _COMMANDS[name] = cls
        cls._command_name = name
        return cls

    return decorator


@register_command("rev-parse")
class RevParseCommand(GitSubCommand):
    def add_arguments(self, parser: argparse.ArgumentParser) -> None:
        parser.add_argument("rev", nargs="*", help="The revision to parse")
        parser.add_argument(
            "--show-toplevel",
            action="store_true",
            help="Show the absolute path of the top-level directory",
        )

    def execute(self, context: Context) -> int:
        repository_root = context.get_repository_root()

        rev_cache: Dict[str, str] = {}

        target_repo = get_target_repository_at_path(
            context.get_relative_path(), repository_root
        )

        def _get_rev(rev: str) -> Optional[str]:
            if rev in rev_cache:
                return rev_cache[rev]

            (
                workspace_id,
                snapshot_version,
            ) = get_workspace_id_and_snapshot_version(context)
            if not workspace_id or not snapshot_version:
                context.error(
                    "Cannot determine workspace id or snapshot version"
                )
                return None

            # Determine repo_root
            repo_root = get_repo_root_for_repo(target_repo)

            request = f'request_base {{ workspace_id: "{workspace_id}" base_snapshot_version: {snapshot_version}}} repo_root: "{repo_root}"'

            try:
                args = ["citc", "api.call", "GetDrafts", request]
                stdout = context.run_real_git(args)

                for line in stdout.splitlines():
                    if "commit_hash:" in line:
                        parts = line.split('"')
                        if len(parts) >= 2:
                            fetched_rev = parts[1]
                            rev_cache[rev] = fetched_rev
                            return fetched_rev
                context.error("Could not find commit hash in citc output")
            except subprocess.CalledProcessError as e:
                context.error(e.stderr)
            except Exception as e:
                context.fatal(f"{e}")
            return None

        # Iterate over arguments in the order they were provided. This is important because
        # the order of arguments determines the order of the output. For example, if the
        # arguments are ["--show-toplevel", "HEAD"], the output should be the repository root
        # followed by the head revision.
        for arg in context.args.remaining_args:
            if arg == "--show-toplevel":
                context.print(str(repository_root / target_repo))
            elif arg.startswith("-"):
                pass
            else:
                if arg != "HEAD":
                    context.error(
                        "cog workspaces only support 'HEAD' revisions at this time"
                    )
                    return 1

                if rev := _get_rev(arg):
                    context.print(rev)
                else:
                    # Do not print an error here since the error was already printed by
                    # _get_rev failing.
                    return 1

        return 0


@register_command("status")
class StatusCommand(GitSubCommand):
    def execute(self, context: Context) -> int:
        context.log_info(str(context.args.global_git_args))
        context.log_info(str(context.git_subcommand_args))
        context.print("not implemented yet")
        return 0


@register_command("ls-files")
class LsFilesCommand(GitSubCommand):
    def add_arguments(self, parser: argparse.ArgumentParser) -> None:
        parser.add_argument(
            "-z",
            action="store_true",
            help="\\0 line termination on output and do not quote filenames.",
        )
        parser.add_argument(
            "-c",
            "--cached",
            action="store_true",
            help="Show all files cached in Git’s index, i.e. all tracked files.",
        )
        parser.add_argument(
            "-d",
            "--deleted",
            action="store_true",
            help="Show files with an unstaged deletion.",
        )
        parser.add_argument(
            "-o",
            "--others",
            action="store_true",
            help="Show other (i.e. untracked) files in the output.",
        )
        parser.add_argument(
            "-i",
            "--ignored",
            action="store_true",
            help="Show only ignored files in the output.",
        )
        parser.add_argument(
            "-s",
            "--stage",
            action="store_true",
            help="Show staged contents' mode bits, object name and stage number in the output.",
        )
        parser.add_argument(
            "-u",
            "--unmerged",
            action="store_true",
            help="Show information about unmerged files in the output.",
        )
        parser.add_argument(
            "-m",
            "--modified",
            action="store_true",
            help="Show files with an unstaged modification.",
        )
        parser.add_argument(
            "-x",
            "--exclude",
            type=str,
            metavar="<pattern>",
            nargs="*",
            help="Skip untracked files matching pattern.",
        )
        parser.add_argument(
            "--format",
            type=str,
            metavar="<format>",
            help="A string that interpolates %(fieldname) from the result being shown.",
        )
        parser.add_argument(
            "--exclude-standard",
            action="store_true",
            help="Exclude the standard git files.",
        )
        parser.add_argument("file", nargs="*", help="Files to show.")

    def execute(self, context: Context) -> int:
        args = context.git_subcommand_args
        try:
            git_args = []
            git_args.extend(context.args.global_git_args)
            git_args.append("ls-files")
            git_args.extend(context.args.remaining_args)

            stdout = context.run_real_git(git_args, cwd=context.invoker_cwd)

            end = "\0" if args and args.z else "\n"
            context.output(stdout, end=end)
            return 0
        except subprocess.CalledProcessError as e:
            context.error(e.stderr)
            return e.returncode
        except Exception as e:
            context.fatal(f"{e}")
            return 10


def _find_command_name_and_position(
    args: List[str],
) -> Tuple[Optional[str], int]:
    for i, arg in enumerate(args):
        if arg in _COMMANDS:
            return arg, i
    return None, -1


def _split_args(args: List[str]) -> Tuple[List[str], List[str]]:
    """Splits args into two lists, everything before '--' and everything after."""
    i = 0
    while i < len(args):
        if args[i] == "--":
            return args[:i], args[i + 1 :]
        i += 1
    return args, []


def _create_polyfill_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(prog="git-polyfill", add_help=False)
    parser.add_argument(
        "--real-git",
        type=str,
        help="Path to the real git binary",
        required=True,
    )
    parser.add_argument(
        "--invoker-cwd",
        type=str,
        help="Path that git was invoked from",
        required=True,
    )
    parser.add_argument(
        "--repository-root",
        type=str,
        help="Path to the repository root. This is the directory that would contain the .git directory for the root repository.",
        required=True,
    )
    parser.add_argument(
        "--log-file",
        type=str,
        help="Path to a file to append logs to.",
    )
    return parser


def _create_global_git_args_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(prog="git", add_help=False)
    parser.add_argument(
        "-C",
        type=str,
        metavar="path",
        help="Run as if git was started in <path>",
    )
    parser.add_argument(
        "--no-optional-locks",
        action="store_true",
        help="Do not perform optional operations that require locks",
    )
    parser.add_argument(
        "--git-dir",
        type=str,
        metavar="path",
        help="Use <path> as the path to the .git directory",
    )
    parser.add_argument(
        "--version", action="version", version="git version 2.x (fuchsia-cog)"
    )
    return parser


@dataclass
class ArgsCollection:
    polyfill_args: List[str]
    global_git_args: List[str]
    command_name: str
    remaining_args: List[str]

    def __init__(self, args: List[str]):
        if "--" not in args:
            raise ValueError(
                "Arguments must contain '--' to separate polyfill args from git args."
            )

        # We split on the first '--' found. This is important because git commands
        # can also use '--' to separate flags from positional arguments (e.g. file paths).
        # We want to ensure that the first '--' is used to separate the polyfill arguments
        # from the git command and its arguments.
        #
        # Example: git.py <polyfill-args> -- <global-git-args> <command> <command-args> -- <files>
        self.polyfill_args, git_args = _split_args(args)

        command_name, command_index = _find_command_name_and_position(git_args)

        if not command_name:
            raise ValueError("No git command found.")

        self.global_git_args = git_args[:command_index]
        self.command_name = command_name
        self.remaining_args = git_args[command_index + 1 :]


def verify_repository_root_is_cog(repository_root: Path) -> bool:
    return (repository_root.parent / ".citc").is_dir()


def main() -> int:
    if len(sys.argv) < 2:
        print("usage: git <command> [<args>]", file=sys.stderr)
        return 1

    provided_args = sys.argv[1:]
    try:
        args_collection = ArgsCollection(provided_args)
    except ValueError as e:
        print(f"error: {e}", file=sys.stderr)
        return 1

    context = Context(args_collection)
    context.write_log("START", f"{shlex.join(sys.argv)}\n")

    if not verify_repository_root_is_cog(context.repository_root):
        context.error("Not in a cog workspace.")
        return 1

    command_class = _COMMANDS[args_collection.command_name]
    command = command_class()
    return command.run(context)


if __name__ == "__main__":
    sys.exit(main())