GKI_KernelSU_SUSFS/.github/actions/download-kernel/action.yml

name: 'Download Kernel Repository'
description: 'Initialize and sync Android kernel source repository'

inputs:
  android_version:
    description: 'Android version (e.g., android14)'
    required: true
  kernel_version:
    description: 'Kernel version (e.g., 5.15, 6.1)'
    required: true
  os_patch_level:
    description: 'OS patch level (e.g., 2024-01)'
    required: true

outputs:
  deprecated_branch:
    description: 'Whether the branch is deprecated'
    value: ${{ steps.sync.outputs.deprecated }}

runs:
  using: composite
  steps:
    # - name: Initialize and Sync Kernel Repository
    #   id: sync
    #   shell: bash
    #   working-directory: ${{ github.workspace }}/kernel
    #   run: |
    #     FORMATTED_BRANCH="${{ inputs.android_version }}-${{ inputs.kernel_version }}-${{ inputs.os_patch_level }}"
    #
    #     repo init -u https://android.googlesource.com/kernel/manifest -b common-${FORMATTED_BRANCH} --depth=1 --partial-clone --clone-filter=blob:limit=10M
    #
    #     REMOTE_BRANCH=$(git ls-remote https://android.googlesource.com/kernel/common ${FORMATTED_BRANCH})
    #     DEFAULT_MANIFEST_PATH=.repo/manifests/default.xml
    #     DEPRECATED=false
    #
    #     if grep -q deprecated <<< $REMOTE_BRANCH; then
    #       sed -i "s/\"${FORMATTED_BRANCH}\"/\"deprecated\/${FORMATTED_BRANCH}\"/g" $DEFAULT_MANIFEST_PATH
    #       echo "⚠ Note: Branch ${FORMATTED_BRANCH} is considered deprecated."
    #       DEPRECATED=true
    #     fi
    #
    #     echo "deprecated=$DEPRECATED" >> $GITHUB_OUTPUT
    #
    #     repo sync -c -j$(nproc --all) --fail-fast --no-tags --force-sync --no-clone-bundle

    - name: Ensure Kernel Directory
      shell: bash
      run: |
        mkdir -p "${{ github.workspace }}/kernel"

    - name: Download Kernel Repository (Manifest Archives)
      id: sync
      shell: python
      working-directory: ${{ github.workspace }}/kernel
      env:
        PYTHONUNBUFFERED: "1"
        GITHUB_TOKEN: ${{ github.token }}
      run: |
        import base64
        import os
        import shutil
        import subprocess
        import tempfile
        import time
        import threading
        import urllib.parse
        import urllib.request
        import xml.etree.ElementTree as ET
        from concurrent.futures import ThreadPoolExecutor

        android_version = "${{ inputs.android_version }}"
        kernel_version = "${{ inputs.kernel_version }}"
        os_patch_level = "${{ inputs.os_patch_level }}"

        formatted_branch = f"{android_version}-{kernel_version}-{os_patch_level}"
        manifest_branch = f"common-{formatted_branch}"
        manifest_ref = f"refs/heads/{manifest_branch}"
        manifest_base = f"https://android.googlesource.com/kernel/manifest/+/{manifest_ref}/"

        def fetch_gitiles_file_text(file_name: str) -> str:
            url = f"{manifest_base}{file_name}?format=TEXT"
            req = urllib.request.Request(url, headers={"User-Agent": "actions-download-kernel"})
            with urllib.request.urlopen(req, timeout=60) as resp:
                data = resp.read()
            decoded = base64.b64decode(data)
            return decoded.decode("utf-8", errors="replace")

        def is_deprecated_branch(branch: str) -> bool:
            try:
                out = subprocess.check_output(
                    ["git", "ls-remote", "https://android.googlesource.com/kernel/common", branch],
                    text=True,
                    stderr=subprocess.STDOUT,
                    timeout=60,
                )
                return "deprecated" in out
            except Exception:
                return False

        deprecated = is_deprecated_branch(formatted_branch)
        if os.environ.get("GITHUB_OUTPUT"):
            with open(os.environ["GITHUB_OUTPUT"], "a", encoding="utf-8") as f:
                f.write(f"deprecated={str(deprecated).lower()}\n")

        print(f"Manifest branch: {manifest_branch}")
        print(f"Deprecated: {deprecated}")

        def load_manifest_with_includes(entry_file: str) -> tuple[ET.Element, list[tuple[str, ET.Element]]]:
            seen_files: set[str] = set()
            to_process = [entry_file]

            combined = ET.Element("manifest")
            combined_default = None
            combined_remotes: dict[str, ET.Element] = {}

            projects: list[ET.Element] = []
            link_copy: list[tuple[str, ET.Element]] = []

            while to_process:
                file_name = to_process.pop()
                if file_name in seen_files:
                    continue
                seen_files.add(file_name)

                xml_text = fetch_gitiles_file_text(file_name)
                if deprecated:
                    xml_text = xml_text.replace(f"\"{formatted_branch}\"", f"\"deprecated/{formatted_branch}\"")

                root = ET.fromstring(xml_text)

                for include in root.findall("include"):
                    inc_name = include.get("name")
                    if inc_name:
                        to_process.append(inc_name)

                for remote in root.findall("remote"):
                    remote_name = remote.get("name")
                    if remote_name and remote_name not in combined_remotes:
                        combined_remotes[remote_name] = remote

                if combined_default is None:
                    d = root.find("default")
                    if d is not None:
                        combined_default = d

                for project in root.findall("project"):
                    projects.append(project)
                    path = project.get("path", project.get("name") or "")
                    for child in list(project):
                        if child.tag in ("linkfile", "copyfile"):
                            link_copy.append((path, child))

            for r in combined_remotes.values():
                combined.append(r)
            if combined_default is not None:
                combined.append(combined_default)
            for p in projects:
                combined.append(p)

            return combined, link_copy

        manifest_root, link_copy = load_manifest_with_includes("default.xml")

        remotes = {r.get("name"): (r.get("fetch") or "").rstrip("/") for r in manifest_root.findall("remote")}
        default = manifest_root.find("default")
        def_remote = default.get("remote") if default is not None else None
        def_rev = default.get("revision") if default is not None else None

        target_repo = os.environ.get("GITHUB_REPOSITORY", "")
        github_token = os.environ.get("GITHUB_TOKEN", "")

        toolchain_map = {
            "clang/host/linux-x86": "clang",
            "prebuilts/rust": "rust",
            "prebuilts/clang-tools": "clang-tools",
            "prebuilts/build-tools": "build-tools",
        }

        _release_assets_lock = threading.Lock()
        _release_assets_cache: list[dict] | None = None
        _release_info_lock = threading.Lock()
        _release_info_cache: dict | None = None
        _upload_locks: dict[tuple[str, str], threading.Lock] = {}
        _upload_locks_guard = threading.Lock()

        def github_api_get_json(url: str) -> object:
            headers = {
                "Accept": "application/vnd.github.v3+json",
                "User-Agent": "actions-download-kernel",
            }
            if github_token:
                headers["Authorization"] = f"token {github_token}"
            req = urllib.request.Request(url, headers=headers)
            with urllib.request.urlopen(req, timeout=60) as resp:
                data = resp.read()
            import json

            return json.loads(data.decode("utf-8", errors="replace"))

        def github_api_post_json(url: str, payload: dict) -> object:
            headers = {
                "Accept": "application/vnd.github.v3+json",
                "User-Agent": "actions-download-kernel",
                "Content-Type": "application/json",
            }
            if github_token:
                headers["Authorization"] = f"token {github_token}"

            import json

            data = json.dumps(payload).encode("utf-8")
            req = urllib.request.Request(url, headers=headers, data=data, method="POST")
            with urllib.request.urlopen(req, timeout=60) as resp:
                body = resp.read()
            return json.loads(body.decode("utf-8", errors="replace"))

        def get_toolchain_label(project_name: str) -> str | None:
            for key, value in toolchain_map.items():
                if key in project_name:
                    return value
            return None

        def get_toolchain_release_assets() -> list[dict]:
            nonlocal _release_assets_cache
            with _release_assets_lock:
                if _release_assets_cache is not None:
                    return _release_assets_cache
                if not target_repo:
                    _release_assets_cache = []
                    return _release_assets_cache
                url = f"https://api.github.com/repos/{target_repo}/releases?per_page=100"
                try:
                    releases = github_api_get_json(url)
                    if not isinstance(releases, list):
                        _release_assets_cache = []
                        return _release_assets_cache
                    release = next(
                        (
                            r
                            for r in releases
                            if isinstance(r, dict)
                            and (r.get("name") == "Toolchains Mirror Cache" or r.get("tag_name") == "toolchain-cache")
                        ),
                        None,
                    )
                    if not release:
                        _release_assets_cache = []
                        return _release_assets_cache
                    assets = release.get("assets", [])
                    _release_assets_cache = assets if isinstance(assets, list) else []
                    return _release_assets_cache
                except Exception:
                    _release_assets_cache = []
                    return _release_assets_cache

        def get_or_create_toolchain_release() -> dict | None:
            nonlocal _release_info_cache, _release_assets_cache
            with _release_info_lock:
                if _release_info_cache is not None:
                    return _release_info_cache
                if not target_repo or not github_token:
                    _release_info_cache = None
                    return _release_info_cache

                try:
                    url = f"https://api.github.com/repos/{target_repo}/releases/tags/toolchain-cache"
                    release = github_api_get_json(url)
                    if isinstance(release, dict):
                        _release_info_cache = release
                        assets = release.get("assets", [])
                        with _release_assets_lock:
                            _release_assets_cache = assets if isinstance(assets, list) else []
                        return _release_info_cache
                except Exception:
                    pass

                try:
                    create_url = f"https://api.github.com/repos/{target_repo}/releases"
                    release = github_api_post_json(
                        create_url,
                        {
                            "tag_name": "toolchain-cache",
                            "name": "Toolchains Mirror Cache",
                            "draft": False,
                            "prerelease": False,
                        },
                    )
                    if isinstance(release, dict):
                        _release_info_cache = release
                        assets = release.get("assets", [])
                        with _release_assets_lock:
                            _release_assets_cache = assets if isinstance(assets, list) else []
                        return _release_info_cache
                except Exception:
                    try:
                        url = f"https://api.github.com/repos/{target_repo}/releases/tags/toolchain-cache"
                        release = github_api_get_json(url)
                        if isinstance(release, dict):
                            _release_info_cache = release
                            assets = release.get("assets", [])
                            with _release_assets_lock:
                                _release_assets_cache = assets if isinstance(assets, list) else []
                            return _release_info_cache
                    except Exception:
                        _release_info_cache = None
                        return _release_info_cache

                _release_info_cache = None
                return _release_info_cache

        def download_github_release_asset(api_url: str, out_path: str) -> None:
            headers = {
                "Accept": "application/octet-stream",
                "User-Agent": "actions-download-kernel",
            }
            if github_token:
                headers["Authorization"] = f"token {github_token}"
            req = urllib.request.Request(api_url, headers=headers)
            with urllib.request.urlopen(req, timeout=120) as resp, open(out_path, "wb") as f:
                shutil.copyfileobj(resp, f, length=1024 * 1024)

        def try_extract_toolchain_cache(project_name: str, rev: str, dest_dir: str, strip_components: int) -> bool:
            label = get_toolchain_label(project_name)
            if not label:
                return False

            assets = get_toolchain_release_assets()
            if not assets:
                return False

            base_filename = f"{label}-{rev}.tar.gz"
            prefix = f"{base_filename}"
            matching = []
            for a in assets:
                if not isinstance(a, dict):
                    continue
                asset_name = a.get("name")
                asset_url = a.get("url")
                if isinstance(asset_name, str) and isinstance(asset_url, str) and asset_name.startswith(prefix):
                    matching.append((asset_name, asset_url))

            if not matching:
                return False

            matching.sort(key=lambda x: x[0])
            part_paths: list[str] = []
            try:
                for asset_name, asset_url in matching:
                    part_path = os.path.abspath(asset_name)
                    download_github_release_asset(asset_url, part_path)
                    part_paths.append(part_path)

                if len(part_paths) == 1 and part_paths[0].endswith(".tar.gz"):
                    extract_tar_gz(part_paths[0], dest_dir, strip_components)
                else:
                    tmp_fd, tmp_path = tempfile.mkstemp(prefix="toolchain-", suffix=".tar.gz")
                    os.close(tmp_fd)
                    try:
                        with open(tmp_path, "wb") as out_f:
                            for p in part_paths:
                                with open(p, "rb") as in_f:
                                    shutil.copyfileobj(in_f, out_f, length=1024 * 1024)
                        extract_tar_gz(tmp_path, dest_dir, strip_components)
                    finally:
                        if os.path.exists(tmp_path):
                            os.remove(tmp_path)

                return True
            except Exception:
                return False
            finally:
                for p in part_paths:
                    if os.path.exists(p):
                        os.remove(p)

        def upload_release_asset(upload_url_template: str, file_path: str, asset_name: str) -> None:
            upload_url = upload_url_template.split("{", 1)[0]
            url = f"{upload_url}?name={urllib.parse.quote(asset_name, safe='')}"
            tmp_fd, tmp_resp = tempfile.mkstemp(prefix="gh-release-upload-", suffix=".json")
            os.close(tmp_fd)
            cmd = [
                "curl",
                "-LsS",
                "-X",
                "POST",
                "-H",
                "Accept: application/vnd.github.v3+json",
                "-H",
                "Content-Type: application/octet-stream",
                "-H",
                "User-Agent: actions-download-kernel",
                "-o",
                tmp_resp,
                "-w",
                "%{http_code}",
            ]
            if github_token:
                cmd.extend(["-H", f"Authorization: token {github_token}"])
            cmd.extend(["--data-binary", f"@{file_path}", url])
            try:
                res = subprocess.run(cmd, check=True, text=True, capture_output=True)
                code = (res.stdout or "").strip()
                if code in {"200", "201"}:
                    return
                if code == "422":
                    return
                body = ""
                try:
                    with open(tmp_resp, "r", encoding="utf-8", errors="replace") as f:
                        body = f.read()
                except Exception:
                    body = ""
                raise RuntimeError(f"Release asset upload failed (HTTP {code}): {body}")
            finally:
                if os.path.exists(tmp_resp):
                    os.remove(tmp_resp)

        def ensure_toolchain_cached(label: str, rev: str, src_dir: str, strip_components: int) -> None:
            if not github_token or not target_repo:
                return

            cache_key = (label, rev)
            with _upload_locks_guard:
                lock = _upload_locks.get(cache_key)
                if lock is None:
                    lock = threading.Lock()
                    _upload_locks[cache_key] = lock

            with lock:
                try:
                    base_filename = f"{label}-{rev}.tar.gz"
                    assets = get_toolchain_release_assets()
                    for a in assets:
                        if isinstance(a, dict):
                            asset_name = a.get("name")
                            if isinstance(asset_name, str) and asset_name.startswith(base_filename):
                                return

                    release = get_or_create_toolchain_release()
                    if not release:
                        return
                    upload_url_template = release.get("upload_url")
                    if not isinstance(upload_url_template, str) or not upload_url_template:
                        return

                    tmp_dir = tempfile.mkdtemp(prefix="toolchain-cache-")
                    try:
                        archive_path = os.path.join(tmp_dir, base_filename)
                        subprocess.run(
                            ["tar", "-I", "gzip -1", "-cf", archive_path, "-C", src_dir, "."],
                            check=True,
                            text=True,
                        )

                        size = os.path.getsize(archive_path)
                        max_part = 1900 * 1024 * 1024

                        if size <= max_part:
                            upload_release_asset(upload_url_template, archive_path, base_filename)
                        else:
                            subprocess.run(
                                ["split", "-b", str(max_part), "-d", "-a", "2", archive_path, f"{archive_path}.part"],
                                check=True,
                                text=True,
                            )
                            for name in sorted(os.listdir(tmp_dir)):
                                if name.startswith(f"{base_filename}.part"):
                                    part_path = os.path.join(tmp_dir, name)
                                    upload_release_asset(upload_url_template, part_path, name)

                        with _release_assets_lock:
                            _release_assets_cache = None
                    finally:
                        shutil.rmtree(tmp_dir, ignore_errors=True)
                except Exception:
                    return

        def url_quote_rev(rev: str) -> str:
            return urllib.parse.quote(rev, safe="")

        def run_curl_download(url: str, out_path: str) -> None:
            subprocess.run(
                ["curl", "-LfsS", "--retry", "5", "--connect-timeout", "30", "-o", out_path, url],
                check=True,
                text=True,
            )

        def extract_tar_gz(archive_path: str, dest_dir: str, strip_components: int) -> None:
            cmd = ["tar", "-xzf", archive_path, "-C", dest_dir]
            if strip_components > 0:
                cmd.extend(["--strip-components", str(strip_components)])
            subprocess.run(cmd, check=True, text=True)

        def github_archive_urls(repo_url: str, rev: str) -> list[str]:
            if rev.startswith("refs/"):
                return [f"{repo_url}/archive/{rev}.tar.gz"]
            return [
                f"{repo_url}/archive/refs/heads/{rev}.tar.gz",
                f"{repo_url}/archive/refs/tags/{rev}.tar.gz",
                f"{repo_url}/archive/{rev}.tar.gz",
            ]

        def gitlab_archive_urls(repo_url: str, rev: str, repo_basename: str) -> list[str]:
            return [f"{repo_url}/-/archive/{urllib.parse.quote(rev, safe='')}/{repo_basename}-{urllib.parse.quote(rev, safe='')}.tar.gz"]

        def build_project_urls(base_url: str, name: str, rev: str) -> tuple[list[str], int]:
            repo_url = f"{base_url}/{name}"

            if "github.com" in base_url:
                return github_archive_urls(repo_url, rev), 1

            if "googlesource.com" in base_url or "android.googlesource.com" in base_url:
                return [f"{repo_url}/+archive/{url_quote_rev(rev)}.tar.gz"], 0

            if "git.codelinaro.org" in base_url or "gitlab" in base_url:
                repo_basename = name.rstrip("/").split("/")[-1]
                return gitlab_archive_urls(repo_url, rev, repo_basename), 1

            return [], 0

        def sync_project(task: tuple[str, str, str, str]) -> bool:
            name, path, rev, base_url = task
            start_time = time.time()
            dest_dir = os.path.abspath(path if path not in ("./", ".") else ".")
            os.makedirs(dest_dir, exist_ok=True)

            try:
                urls, strip_components = build_project_urls(base_url, name, rev)
                if not urls:
                    return False

                if try_extract_toolchain_cache(name, rev, dest_dir, strip_components):
                    duration = time.time() - start_time
                    print(f"Synced {name} -> {path} ({duration:.2f}s)")
                    return True

                tmp_fd, tmp_path = tempfile.mkstemp(prefix="kernel-src-", suffix=".tar.gz")
                os.close(tmp_fd)
                try:
                    last_err = None
                    for url in urls:
                        try:
                            run_curl_download(url, tmp_path)
                            extract_tar_gz(tmp_path, dest_dir, strip_components)
                            label = get_toolchain_label(name)
                            if label:
                                ensure_toolchain_cached(label, rev, dest_dir, strip_components)
                            duration = time.time() - start_time
                            print(f"Synced {name} -> {path} ({duration:.2f}s)")
                            return True
                        except subprocess.CalledProcessError as e:
                            last_err = e
                            continue
                    if last_err:
                        raise last_err
                    return False
                finally:
                    if os.path.exists(tmp_path):
                        os.remove(tmp_path)
            except Exception:
                return False

        sync_tasks: list[tuple[str, str, str, str]] = []

        for project in manifest_root.findall("project"):
            name = project.get("name")
            if not name:
                continue
            path = project.get("path", name)
            remote_name = project.get("remote", def_remote)
            if not remote_name:
                continue
            base_url = remotes.get(remote_name)
            if not base_url:
                continue
            rev = project.get("revision", def_rev)
            if not rev:
                continue
            sync_tasks.append((name, path, rev, base_url))

        max_workers = min(32, (os.cpu_count() or 2) * 2)
        results = []
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            results = list(executor.map(sync_project, sync_tasks))

        if not all(results):
            raise SystemExit("One or more projects failed to sync")

        for project_path, child in link_copy:
            src_rel = child.get("src")
            dest_rel = child.get("dest")
            if not src_rel or not dest_rel:
                continue

            src_path = os.path.join(os.getcwd(), project_path, src_rel)
            dest_path = os.path.join(os.getcwd(), dest_rel)
            os.makedirs(os.path.dirname(dest_path), exist_ok=True)

            if child.tag == "linkfile":
                if os.path.lexists(dest_path):
                    os.remove(dest_path)
                rel_target = os.path.relpath(src_path, os.path.dirname(dest_path))
                os.symlink(rel_target, dest_path)
            elif child.tag == "copyfile":
                shutil.copy2(src_path, dest_path)