#!/usr/bin/env python3
"""Fetch exact Forgejo source archives declared by the policy source inventory."""

from __future__ import annotations

import argparse
import hashlib
import json
import os
from pathlib import Path, PurePosixPath
import tarfile
import tempfile
import urllib.parse
import urllib.request


def _url_origin(url: str) -> str:
    parsed = urllib.parse.urlparse(url)
    if parsed.scheme != "https" or not parsed.netloc or parsed.username or parsed.password:
        raise ValueError(f"authenticated source URL must use an HTTPS origin: {url!r}")
    return f"{parsed.scheme}://{parsed.netloc.lower()}"


def _credential_origin(value: str) -> str:
    parsed = urllib.parse.urlparse(value)
    if parsed.path not in ("", "/") or parsed.params or parsed.query or parsed.fragment:
        raise ValueError("credential origin must contain only an HTTPS scheme and authority")
    return _url_origin(value)


def _request(
    url: str, *, token: str | None = None, token_origin: str | None = None
) -> urllib.request.Request:
    if bool(token) != bool(token_origin):
        raise ValueError("source token and token origin must be provided together")
    request = urllib.request.Request(url)
    if token:
        expected_origin = _credential_origin(token_origin or "")
        if _url_origin(url) != expected_origin:
            raise ValueError(
                f"refusing to send source credential outside {expected_origin}"
            )
        request.add_header("Authorization", f"token {token}")
    return request


def _revision(
    remote: str,
    branch: str,
    *,
    token: str | None = None,
    token_origin: str | None = None,
) -> str:
    parsed = urllib.parse.urlparse(remote)
    parts = parsed.path.removesuffix(".git").strip("/").split("/")
    if parsed.scheme != "https" or len(parts) != 2:
        raise ValueError(f"unsupported Forgejo remote {remote!r}")
    owner, repo = (urllib.parse.quote(part, safe="") for part in parts)
    branch_name = urllib.parse.quote(branch, safe="")
    url = f"{parsed.scheme}://{parsed.netloc}/api/v1/repos/{owner}/{repo}/branches/{branch_name}"
    with urllib.request.urlopen(
        _request(url, token=token, token_origin=token_origin), timeout=30
    ) as response:
        revision = json.load(response).get("commit", {}).get("id", "")
    if len(revision) != 40 or any(char not in "0123456789abcdef" for char in revision):
        raise ValueError(f"{remote}: could not resolve a clean 40-hex {branch} revision")
    return revision


def _archive_url(remote: str, revision: str) -> str:
    if not remote.startswith("https://") or not remote.endswith(".git"):
        raise ValueError(f"archive source must be an HTTPS .git URL, got {remote!r}")
    return f"{remote[:-4]}/archive/{revision}.tar.gz"


def _extract(archive: Path, target: Path) -> None:
    target.mkdir(parents=True, exist_ok=False)
    with tarfile.open(archive, "r:gz") as bundle:
        members = bundle.getmembers()
        roots = {
            PurePosixPath(member.name).parts[0]
            for member in members
            if PurePosixPath(member.name).parts
        }
        if len(roots) != 1:
            raise ValueError(f"{archive}: expected exactly one archive root")
        root = next(iter(roots))
        for member in members:
            path = PurePosixPath(member.name)
            if not path.parts or path.parts[0] != root:
                raise ValueError(f"{archive}: inconsistent archive root")
            relative = PurePosixPath(*path.parts[1:])
            if not relative.parts:
                continue
            if relative.is_absolute() or ".." in relative.parts:
                raise ValueError(f"{archive}: unsafe member {member.name!r}")
            destination = target.joinpath(*relative.parts)
            if member.isdir():
                destination.mkdir(parents=True, exist_ok=True)
                continue
            if not member.isfile():
                # Links and special files are not part of the publishable source
                # corpus. Skipping them avoids materializing archive links; if a
                # selector ever names one, the inventory audit fails on absence.
                continue
            destination.parent.mkdir(parents=True, exist_ok=True)
            source = bundle.extractfile(member)
            if source is None:
                raise ValueError(f"{archive}: could not read {member.name!r}")
            with destination.open("wb") as output:
                while chunk := source.read(1024 * 1024):
                    output.write(chunk)


def fetch(
    config_path: Path,
    destination: Path,
    policy_revision: str,
    *,
    token: str | None = None,
    token_origin: str | None = None,
) -> dict[str, object]:
    if len(policy_revision) != 40 or any(
        char not in "0123456789abcdef" for char in policy_revision
    ):
        raise ValueError("--policy-revision must be a clean 40-hex Git commit")
    config = json.loads(config_path.read_text(encoding="utf-8"))
    if config.get("schema_version") != 1:
        raise ValueError("source inventory config schema_version must be 1")
    destination.mkdir(parents=True, exist_ok=True)
    if any(destination.iterdir()):
        raise ValueError(f"destination must be empty: {destination}")

    revisions: dict[str, str] = {}
    repositories: dict[str, dict[str, str]] = {}
    for name, repository in sorted(config["repositories"].items()):
        if repository.get("local"):
            revision = policy_revision
            repositories[name] = {"revision": revision, "source": "workflow-checkout"}
            revisions[name] = revision
            continue
        remote = repository["remote"]
        branch = repository.get("branch", "main")
        revision = _revision(
            remote, branch, token=token, token_origin=token_origin
        )
        url = _archive_url(remote, revision)
        with tempfile.NamedTemporaryFile(suffix=".tar.gz") as archive:
            with urllib.request.urlopen(
                _request(url, token=token, token_origin=token_origin), timeout=60
            ) as response:
                while chunk := response.read(1024 * 1024):
                    archive.write(chunk)
            archive.flush()
            _extract(Path(archive.name), destination / name)
        revisions[name] = revision
        repositories[name] = {
            "revision": revision,
            "remote": remote,
            "branch": branch,
        }

    canonical = json.dumps(revisions, sort_keys=True, separators=(",", ":")).encode()
    lock: dict[str, object] = {
        "schema_version": 1,
        "source_set_digest": hashlib.sha256(canonical).hexdigest(),
        "repositories": repositories,
    }
    (destination / "source-lock.json").write_text(
        json.dumps(lock, indent=2, sort_keys=True) + "\n", encoding="utf-8"
    )
    return lock


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--config", type=Path, default=Path("source-inventory.config.json"))
    parser.add_argument("--destination", type=Path, required=True)
    parser.add_argument("--policy-revision", required=True)
    parser.add_argument(
        "--token-env",
        help="environment variable containing a Forgejo repository-read token",
    )
    parser.add_argument(
        "--token-origin",
        help="sole HTTPS origin to which the source token may be sent",
    )
    args = parser.parse_args()
    token = None
    if args.token_env:
        token = os.environ.get(args.token_env, "").strip()
        if not token:
            parser.error(f"source token environment variable {args.token_env!r} is empty")
    if bool(token) != bool(args.token_origin):
        parser.error("--token-env and --token-origin must be supplied together")
    lock = fetch(
        args.config.resolve(),
        args.destination.resolve(),
        args.policy_revision,
        token=token,
        token_origin=args.token_origin,
    )
    print(
        json.dumps(
            {
                "source_set_digest": lock["source_set_digest"],
                "repositories": len(lock["repositories"]),
            },
            sort_keys=True,
        )
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())
