# SPDX-FileCopyrightText: © 2026 Jeffrey C. Ollie <jeff@ocjtech.us>
# SPDX-License-Identifier: MIT

"""Make a submission to the JCOllie winget source for a program's release.

Run by the composite action in `action.yml`, with its inputs in
`WINGET_SUBMIT_*` environment variables. It clones jeff/winget-pkgs at main,
has `wingetsource new-version` write the new version there and
`wingetsource check` judge it, then pushes a branch and opens a pull request
as the submitter. The manifest itself is never written here (ADR 0004 in
jeff/winget-pkgs).
"""

import base64
import json
import os
import subprocess
import sys
import tempfile
import urllib.error
import urllib.request
from dataclasses import dataclass

FORGE = "https://git.jcollie.dev"
MANIFEST_REPOSITORY = "jeff/winget-pkgs"
BASE_BRANCH = "main"
USER_AGENT = "winget-submit (+https://git.jcollie.dev/jeff/winget-submit)"

# The submitter. Built in rather than read from the forge: its token is
# limited to jeff/winget-pkgs, and Forgejo lets such a token have only
# repository and issue scopes, so it cannot read its own user.
SUBMITTER = "winget-bot"
SUBMITTER_EMAIL = "winget-bot@jcollie.dev"


class SubmitError(Exception):
    pass


@dataclass
class Inputs:
    identifier: str
    version: str
    installers: dict[str, str]
    product_codes: dict[str, str]
    release_notes_url: str
    release_date: str
    token: str
    dry_run: bool


def parse_pairs(name: str, what: str, text: str) -> dict[str, str]:
    """`<architecture> <what>`, one per line; blank lines are ignored."""
    pairs = {}
    for number, line in enumerate(text.splitlines(), 1):
        fields = line.split()
        if not fields:
            continue
        if len(fields) != 2:
            raise SubmitError(f"{name} line {number}: expected '<architecture> <{what}>', got {line.strip()!r}")
        architecture, value = fields
        if architecture in pairs:
            raise SubmitError(f"{name} line {number}: {architecture} is given more than once")
        pairs[architecture] = value
    return pairs


def parse_installers(text: str) -> dict[str, str]:
    installers = parse_pairs("installers", "url", text)
    if not installers:
        raise SubmitError("installers: at least one '<architecture> <url>' line is needed")
    return installers


def parse_product_codes(text: str) -> dict[str, str]:
    """Optional: `wingetsource new-version` reads a new MSI's ProductCode
    from it, and one given here is written instead."""
    return parse_pairs("product-codes", "code", text)


def parse_bool(name: str, text: str) -> bool:
    value = text.strip().lower()
    if value in ("", "false"):
        return False
    if value == "true":
        return True
    raise SubmitError(f"{name}: expected true or false, got {text!r}")


def read_inputs(environ) -> Inputs:
    def get(name: str) -> str:
        return environ.get(f"WINGET_SUBMIT_{name}", "").strip()

    inputs = Inputs(
        identifier=get("IDENTIFIER"),
        version=get("VERSION"),
        installers=parse_installers(environ.get("WINGET_SUBMIT_INSTALLERS", "")),
        product_codes=parse_product_codes(environ.get("WINGET_SUBMIT_PRODUCT_CODES", "")),
        release_notes_url=get("RELEASE_NOTES_URL"),
        release_date=get("RELEASE_DATE"),
        token=get("TOKEN"),
        dry_run=parse_bool("dry-run", get("DRY_RUN")),
    )
    for name in ("identifier", "version"):
        if not getattr(inputs, name):
            raise SubmitError(f"{name} is required")
    if not inputs.token and not inputs.dry_run:
        raise SubmitError("token is required, unless this is a dry run")
    return inputs


def branch_name(identifier: str, version: str) -> str:
    return f"submit/{identifier}/{version}"


def title(identifier: str, version: str) -> str:
    return f"{identifier} {version}"


def release_url(inputs: Inputs, environ) -> str | None:
    """The program's release: its release notes if given, or else the release
    page of the tag the workflow is running for."""
    if inputs.release_notes_url:
        return inputs.release_notes_url
    server, repository, ref = (environ.get(k, "") for k in ("GITHUB_SERVER_URL", "GITHUB_REPOSITORY", "GITHUB_REF"))
    if server and repository and ref.startswith("refs/tags/"):
        return f"{server}/{repository}/releases/tag/{ref.removeprefix('refs/tags/')}"
    return None


def run_url(environ) -> str | None:
    server, repository, number = (environ.get(k, "") for k in ("GITHUB_SERVER_URL", "GITHUB_REPOSITORY", "GITHUB_RUN_NUMBER"))
    if server and repository and number:
        return f"{server}/{repository}/actions/runs/{number}"
    return None


def body(inputs: Inputs, copied_from: str, environ) -> str:
    lines = [f"Submits {inputs.identifier} {inputs.version}, written by `wingetsource new-version`.", ""]
    release = release_url(inputs, environ)
    if release:
        lines.append(f"- Release: {release}")
    run = run_url(environ)
    if run:
        lines.append(f"- Opened by: {run}")
    lines.append(f"- Copied from: {copied_from}")
    lines.append("")
    lines.append("Opened by [winget-submit](https://git.jcollie.dev/jeff/winget-submit).")
    return "\n".join(lines) + "\n"


def git_auth_environment(token: str, login: str) -> dict[str, str]:
    """Configuration that hands git the token as basic auth, through the
    environment rather than the command line or a remote URL, so that it is
    neither visible in the process list nor written into the clone."""
    if not token:
        return {}
    credentials = base64.b64encode(f"{login}:{token}".encode()).decode()
    return {
        "GIT_CONFIG_COUNT": "1",
        "GIT_CONFIG_KEY_0": f"http.{FORGE}/.extraHeader",
        "GIT_CONFIG_VALUE_0": f"Authorization: Basic {credentials}",
    }


def open_pull_request_for(pulls: list[dict], branch: str) -> dict | None:
    for pull in pulls:
        head = pull.get("head") or {}
        if head.get("ref") == branch and (head.get("repo") or {}).get("full_name") == MANIFEST_REPOSITORY:
            return pull
    return None


class Forge:
    def __init__(self, token: str):
        self.token = token

    def request(self, method: str, path: str, data: dict | None = None):
        request = urllib.request.Request(
            f"{FORGE}/api/v1{path}",
            method=method,
            data=json.dumps(data).encode() if data is not None else None,
            headers={
                "Accept": "application/json",
                "Content-Type": "application/json",
                "Authorization": f"token {self.token}",
                "User-Agent": USER_AGENT,
            },
        )
        try:
            with urllib.request.urlopen(request, timeout=60) as response:
                return json.load(response)
        except urllib.error.HTTPError as e:
            raise SubmitError(f"{method} {path}: {e.code} {e.reason}: {e.read().decode(errors='replace')}") from e

    def open_pulls(self) -> list[dict]:
        pulls, page = [], 1
        while True:
            batch = self.request("GET", f"/repos/{MANIFEST_REPOSITORY}/pulls?state=open&limit=50&page={page}")
            pulls.extend(batch)
            if len(batch) < 50:
                return pulls
            page += 1


def run(command: list[str], cwd: str, env: dict, capture: bool = False) -> str:
    print(f"+ {' '.join(command)}", file=sys.stderr, flush=True)
    result = subprocess.run(command, cwd=cwd, env=env, text=True, stdout=subprocess.PIPE if capture else None)
    if result.returncode != 0:
        raise SubmitError(f"{command[0]} {command[1]} failed with status {result.returncode}")
    return result.stdout if capture else ""


def submit(inputs: Inputs, environ) -> None:
    forge = Forge(inputs.token)
    env = {**environ, **git_auth_environment(inputs.token, SUBMITTER)}
    branch = branch_name(inputs.identifier, inputs.version)

    with tempfile.TemporaryDirectory(dir=environ.get("RUNNER_TEMP")) as work:
        clone = os.path.join(work, "winget-pkgs")
        run(["git", "clone", "--quiet", "--depth", "1", "--branch", BASE_BRANCH, f"{FORGE}/{MANIFEST_REPOSITORY}.git", clone], work, env)

        command = [
            "nix", "develop", "-c", "python3", "-m", "wingetsource", "new-version", "--json",
            "--manifests", "manifests",
            "--identifier", inputs.identifier,
            "--version", inputs.version,
        ]
        for architecture, url in inputs.installers.items():
            command += ["--installer", f"{architecture}={url}"]
        for architecture, code in inputs.product_codes.items():
            command += ["--product-code", f"{architecture}={code}"]
        if inputs.release_notes_url:
            command += ["--release-notes-url", inputs.release_notes_url]
        if inputs.release_date:
            command += ["--release-date", inputs.release_date]
        output = run(command, clone, env, capture=True)
        written = json.loads(output.strip().splitlines()[-1])

        run(["git", "add", "--", written["directory"]], clone, env)
        identity = ["-c", f"user.name={SUBMITTER}", "-c", f"user.email={SUBMITTER_EMAIL}"]
        run(["git", *identity, "commit", "--quiet", "-m", title(inputs.identifier, inputs.version)], clone, env)
        # `check` finds the versions to download installers for by what
        # differs from the base, so it runs on the commit.
        run(["nix", "develop", "-c", "python3", "-m", "wingetsource", "check", "--manifests", "manifests", "--changed-since", f"origin/{BASE_BRANCH}"], clone, env)

        if inputs.dry_run:
            run(["git", "--no-pager", "show", "--stat", "--patch", "HEAD"], clone, env)
            print(f"Dry run: would push {branch} and open '{title(inputs.identifier, inputs.version)}' with:\n")
            print(body(inputs, written["copied_from"], environ))
            return

        run(["git", "push", "--quiet", "--force", "origin", f"HEAD:refs/heads/{branch}"], clone, env)

    existing = open_pull_request_for(forge.open_pulls(), branch)
    if existing:
        print(f"Updated {branch}; its pull request is already open: {existing['html_url']}")
        return
    pull = forge.request(
        "POST",
        f"/repos/{MANIFEST_REPOSITORY}/pulls",
        {
            "base": BASE_BRANCH,
            "head": branch,
            "title": title(inputs.identifier, inputs.version),
            "body": body(inputs, written["copied_from"], environ),
        },
    )
    print(f"Opened {pull['html_url']}")


def main() -> int:
    try:
        submit(read_inputs(os.environ), os.environ)
    except SubmitError as e:
        print(f"winget-submit: {e}", file=sys.stderr)
        return 1
    return 0


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