1"""
  2Oxc standalone binary management for plain-code.
  3
  4Downloads and manages oxlint (linter) and oxfmt (formatter) binaries
  5from the oxc-project/oxc GitHub releases.
  6"""
  7
  8from __future__ import annotations
  9
 10import io
 11import os
 12import platform
 13import subprocess
 14import tarfile
 15import zipfile
 16
 17import click
 18import requests
 19import tomlkit
 20
 21from plain.runtime import PLAIN_TEMP_PATH
 22
 23TAG_PREFIX = "apps_v"
 24
 25
 26class OxcTool:
 27    """Download, install, and invoke an Oxc CLI binary (oxlint or oxfmt)."""
 28
 29    def __init__(self, name: str) -> None:
 30        if name not in ("oxlint", "oxfmt"):
 31            raise ValueError(f"Unknown Oxc tool: {name}")
 32        self.name = name
 33
 34    @property
 35    def target_directory(self) -> str:
 36        return str(PLAIN_TEMP_PATH)
 37
 38    @property
 39    def standalone_path(self) -> str:
 40        exe = ".exe" if platform.system() == "Windows" else ""
 41        return os.path.join(self.target_directory, f"{self.name}{exe}")
 42
 43    @property
 44    def version_lockfile_path(self) -> str:
 45        return os.path.join(self.target_directory, "oxc.version")
 46
 47    def is_installed(self) -> bool:
 48        td = self.target_directory
 49        if not os.path.isdir(td):
 50            os.makedirs(td, exist_ok=True)
 51        return os.path.exists(self.standalone_path)
 52
 53    def needs_update(self) -> bool:
 54        if not self.is_installed():
 55            return True
 56        if not os.path.exists(self.version_lockfile_path):
 57            return True
 58        with open(self.version_lockfile_path) as f:
 59            locked = f.read().strip()
 60        return locked != self.get_version_from_config()
 61
 62    @staticmethod
 63    def get_version_from_config() -> str:
 64        project_root = os.path.dirname(str(PLAIN_TEMP_PATH))
 65        pyproject = os.path.join(project_root, "pyproject.toml")
 66        if not os.path.exists(pyproject):
 67            return ""
 68        doc = tomlkit.loads(open(pyproject, "rb").read().decode())
 69        return (
 70            doc.get("tool", {})
 71            .get("plain", {})
 72            .get("code", {})
 73            .get("oxc", {})
 74            .get("version", "")
 75        )
 76
 77    @staticmethod
 78    def set_version_in_config(version: str) -> None:
 79        project_root = os.path.dirname(str(PLAIN_TEMP_PATH))
 80        pyproject = os.path.join(project_root, "pyproject.toml")
 81        if not os.path.exists(pyproject):
 82            return
 83        doc = tomlkit.loads(open(pyproject, "rb").read().decode())
 84        doc.setdefault("tool", {}).setdefault("plain", {}).setdefault(
 85            "code", {}
 86        ).setdefault("oxc", {})["version"] = version
 87        open(pyproject, "w").write(tomlkit.dumps(doc))
 88
 89    def detect_platform_slug(self) -> str:
 90        system = platform.system()
 91        arch = platform.machine()
 92        if system == "Windows":
 93            if arch.lower() in ("arm64", "aarch64"):
 94                return "aarch64-pc-windows-msvc"
 95            return "x86_64-pc-windows-msvc"
 96        if system == "Linux":
 97            if arch == "aarch64":
 98                return "aarch64-unknown-linux-gnu"
 99            return "x86_64-unknown-linux-gnu"
100        if system == "Darwin":
101            if arch == "arm64":
102                return "aarch64-apple-darwin"
103            return "x86_64-apple-darwin"
104        raise RuntimeError(f"Unsupported platform for Oxc: {system}/{arch}")
105
106    @staticmethod
107    def get_latest_version() -> str:
108        """Find the latest apps_v release tag via the GitHub API."""
109        resp = requests.get(
110            "https://api.github.com/repos/oxc-project/oxc/releases",
111            params={"per_page": 20},
112            headers={"Accept": "application/vnd.github+json"},
113        )
114        resp.raise_for_status()
115        for release in resp.json():
116            tag = release["tag_name"]
117            if tag.startswith(TAG_PREFIX):
118                return tag[len(TAG_PREFIX) :]
119        raise RuntimeError("No apps_v release found on GitHub")
120
121    def download(self, version: str = "") -> str:
122        if not version:
123            version = self.get_latest_version()
124
125        slug = self.detect_platform_slug()
126        is_windows = platform.system() == "Windows"
127        ext = "zip" if is_windows else "tar.gz"
128        asset = f"{self.name}-{slug}.{ext}"
129        url = f"https://github.com/oxc-project/oxc/releases/download/{TAG_PREFIX}{version}/{asset}"
130
131        resp = requests.get(url, stream=True)
132        resp.raise_for_status()
133
134        td = self.target_directory
135        if not os.path.isdir(td):
136            os.makedirs(td, exist_ok=True)
137
138        # Download into memory for extraction
139        data = io.BytesIO()
140        total = int(resp.headers.get("Content-Length", 0))
141        if total:
142            with click.progressbar(
143                length=total,
144                label=f"Downloading {self.name}",
145                width=0,
146            ) as bar:
147                for chunk in resp.iter_content(chunk_size=1024 * 1024):
148                    data.write(chunk)
149                    bar.update(len(chunk))
150        else:
151            for chunk in resp.iter_content(chunk_size=1024 * 1024):
152                data.write(chunk)
153
154        data.seek(0)
155
156        # Extract the binary from the archive
157        if is_windows:
158            with zipfile.ZipFile(data) as zf:
159                # Find the binary inside the archive
160                members = zf.namelist()
161                binary_name = next(m for m in members if m.startswith(self.name))
162                with (
163                    zf.open(binary_name) as src,
164                    open(self.standalone_path, "wb") as dst,
165                ):
166                    dst.write(src.read())
167        else:
168            with tarfile.open(fileobj=data, mode="r:gz") as tf:
169                members = tf.getnames()
170                binary_name = next(m for m in members if m.startswith(self.name))
171                extracted = tf.extractfile(binary_name)
172                if extracted is None:
173                    raise RuntimeError(f"Failed to extract {binary_name} from archive")
174                with open(self.standalone_path, "wb") as dst:
175                    dst.write(extracted.read())
176
177        os.chmod(self.standalone_path, 0o755)
178        return version.lstrip("v")
179
180    def invoke(self, *args: str, cwd: str | None = None) -> subprocess.CompletedProcess:
181        config_path = os.path.join(
182            os.path.dirname(__file__), f"{self.name}_defaults.json"
183        )
184        extra_args = ["-c", config_path]
185        return subprocess.run([self.standalone_path, *extra_args, *args], cwd=cwd)
186
187
188def install_oxc(version: str = "") -> str:
189    """Install both oxlint and oxfmt, return the resolved version."""
190    oxlint = OxcTool("oxlint")
191    oxfmt = OxcTool("oxfmt")
192
193    resolved = oxlint.download(version)
194    oxfmt.download(resolved)
195
196    # Write version lockfile once (shared by both tools)
197    with open(oxlint.version_lockfile_path, "w") as f:
198        f.write(resolved)
199
200    OxcTool.set_version_in_config(resolved)
201    return resolved