"""Observe a running build without parsing its output.

Arch has no machine-readable build event stream, so anything read out of paru's
or makepkg's human output is brittle by construction: it breaks on version,
locale, terminal width and colour settings.

We spawn paru ourselves, so we can observe the kernel instead:

    /proc/<pid>/cwd      -> which package's clone dir is being built (exact)
    /proc/<pid>/stat     -> starttime, hence an exact per-package elapsed timer
    /proc/<pid>/cmdline  -> which build system is actually running
    stat() on artefacts  -> corroborates completion

No cooperation from paru or makepkg is required, and nothing is scraped.
"""

from __future__ import annotations

import os
import time
from collections import deque
from dataclasses import dataclass
from pathlib import Path

PROC = Path("/proc")


def _clk_tck() -> int:
    try:
        return os.sysconf("SC_CLK_TCK")
    except (ValueError, OSError):  # pragma: no cover
        return 100


def boot_time() -> float:
    """Seconds since epoch at boot, from /proc/stat `btime`."""
    with open("/proc/stat", encoding="ascii", errors="replace") as fh:
        for line in fh:
            if line.startswith("btime "):
                return float(line.split()[1])
    return 0.0


@dataclass(frozen=True)
class Proc:
    pid: int
    ppid: int
    comm: str
    cmdline: tuple[str, ...]
    cwd: str | None
    started: float  # epoch seconds

    @property
    def elapsed(self) -> float:
        return max(0.0, time.time() - self.started)

    @property
    def argv0(self) -> str:
        return os.path.basename(self.cmdline[0]) if self.cmdline else self.comm


def read_proc(pid: int, btime: float, clk: int) -> Proc | None:
    """Read one process. Returns None the moment it dies, which is normal."""
    try:
        raw = (PROC / str(pid) / "stat").read_bytes()
    except OSError:
        return None
    # comm can contain spaces and parentheses, so parse from the last ')'
    end = raw.rfind(b")")
    if end < 0:
        return None
    comm = raw[raw.find(b"(") + 1 : end].decode(errors="replace")
    fields = raw[end + 2 :].split()
    if len(fields) < 20:
        return None
    try:
        ppid = int(fields[1])
        starttime = int(fields[19])
    except ValueError:
        return None

    try:
        cmdline = tuple(
            part.decode(errors="replace")
            for part in (PROC / str(pid) / "cmdline").read_bytes().split(b"\0")
            if part
        )
    except OSError:
        cmdline = ()

    try:
        cwd = os.readlink(PROC / str(pid) / "cwd")
    except OSError:
        cwd = None

    return Proc(
        pid=pid, ppid=ppid, comm=comm, cmdline=cmdline, cwd=cwd, started=btime + starttime / clk
    )


def snapshot() -> dict[int, Proc]:
    """Every readable process, keyed by pid."""
    btime, clk = boot_time(), _clk_tck()
    out: dict[int, Proc] = {}
    for entry in os.listdir(PROC):
        if not entry.isdigit():
            continue
        proc = read_proc(int(entry), btime, clk)
        if proc is not None:
            out[proc.pid] = proc
    return out


def descendants(root_pid: int, procs: dict[int, Proc], *, include_root: bool = False) -> list[Proc]:
    """Every process under `root_pid`, breadth-first.

    The root is excluded by default because the root is the launcher we spawned
    (paru), which does not itself build and whose working directory is wherever
    the user happened to run us. Pass `include_root=True` when the root *is* the
    builder, as it would be if we ever drive makepkg directly for parallel
    builds.
    """
    children: dict[int, list[Proc]] = {}
    for proc in procs.values():
        children.setdefault(proc.ppid, []).append(proc)

    out: list[Proc] = []
    if include_root and root_pid in procs:
        out.append(procs[root_pid])
    queue = deque(children.get(root_pid, []))
    while queue:
        proc = queue.popleft()
        out.append(proc)
        queue.extend(children.get(proc.pid, []))
    return out


# Build systems, most specific first. The first match on a process's argv0 wins.
BUILD_TOOLS: tuple[tuple[str, str], ...] = (
    ("cc1plus", "compile"),
    ("cc1", "compile"),
    ("clang", "compile"),
    ("gcc", "compile"),
    ("g++", "compile"),
    ("rustc", "compile"),
    ("cargo", "cargo"),
    ("go", "go build"),
    ("nim", "nim"),
    ("meson", "meson"),
    ("ninja", "ninja"),
    ("cmake", "cmake"),
    ("make", "make"),
    ("scons", "scons"),
    ("waf", "waf"),
    ("npm", "node"),
    ("pnpm", "node"),
    ("yarn", "node"),
    ("node", "node"),
    ("setup.py", "python"),
    ("pip", "python"),
    ("python", "python"),
    ("patch", "prepare"),
    ("git", "fetch"),
    ("curl", "fetch"),
    ("wget", "fetch"),
    ("fakeroot", "package"),
    ("makepkg", "makepkg"),
)

#: processes that only exist to drive the build; not worth reporting as a phase
DRIVERS = {"makepkg", "fakeroot", "sudo", "paru", "systemd"}


#: higher-level build systems first. When several are running at once the label
#: should describe the phase, and "cargo" is more useful than "compile".
PHASE_PRIORITY: tuple[str, ...] = (
    "cargo",
    "go build",
    "meson",
    "ninja",
    "cmake",
    "node",
    "python",
    "nim",
    "scons",
    "waf",
    "make",
    "compile",
    "package",
    "prepare",
    "fetch",
    "makepkg",
)


def _tool_names(proc: Proc) -> set[str]:
    """Candidate tool names for a process.

    argv0 always counts. Later arguments count only when they look like a path,
    so a shell wrapper invoked as `/bin/bash /usr/bin/cargo` is recognised while
    `cat make` is not mistaken for a make run.
    """
    names: set[str] = set()
    for i, arg in enumerate(proc.cmdline):
        if i == 0 or "/" in arg:
            names.add(os.path.basename(arg))
    return names or {proc.comm}


def _labels(proc: Proc) -> list[str]:
    names = _tool_names(proc)
    return [label for tool, label in BUILD_TOOLS if tool in names]


#: helper processes that are part of building but are not the work itself
_NOISE = frozenset(
    {
        "sh",
        "bash",
        "dash",
        "make",
        "ninja",
        "meson",
        "cmake",
        "python",
        "python3",
        "perl",
        "sed",
        "awk",
        "cat",
        "tee",
        "env",
        "time",
        "xargs",
    }
)


def summarise_workers(procs: list[Proc], *, limit: int = 4) -> str:
    """What the running jobs are, counted by program.

    "5 jobs" says something is happening; "rustc ×4, cc1 ×1" says what, which
    is the only thing that distinguishes a compiling build from a hung one.
    """
    counts: dict[str, int] = {}
    for proc in procs:
        name = proc.argv0 or proc.comm
        if not name or name in _NOISE:
            continue
        counts[name] = counts.get(name, 0) + 1
    if not counts:
        return ""
    ordered = sorted(counts.items(), key=lambda kv: (-kv[1], kv[0]))
    shown = [f"{name} ×{n}" if n > 1 else name for name, n in ordered[:limit]]
    if len(ordered) > limit:
        shown.append(f"+{len(ordered) - limit} more")
    return ", ".join(shown)


def classify(procs: list[Proc]) -> str:
    """Best available label for what a package's process tree is doing.

    Deterministic regardless of process ordering: the highest-priority label
    present wins, rather than whichever process happened to be listed first.
    """
    found = {label for proc in procs for label in _labels(proc)}
    for label in PHASE_PRIORITY:
        if label in found:
            return label
    return ""


def package_of(path: str | None, clone_dir: Path, known: set[str] | None = None) -> str | None:
    """Map a cwd back to the AUR package whose clone directory contains it.

    makepkg runs with cwd == <clone_dir>/<pkg>; compilers run deeper, in
    <clone_dir>/<pkg>/src/... . Either way the first path component is the answer.
    """
    if not path:
        return None
    # /proc/<pid>/cwd is a fully resolved path, so a symlinked or relative
    # clone directory would never match it and every lookup would return None.
    try:
        rel = Path(path).relative_to(Path(clone_dir).resolve())
    except ValueError:
        return None
    if not rel.parts:
        return None
    name = rel.parts[0]
    if name.startswith("."):
        return None
    if known is not None and name not in known:
        return None
    return name


@dataclass
class BuildState:
    package: str
    phase: str
    elapsed: float
    jobs: int = 0
    #: what the jobs actually are, e.g. "rustc ×4, cc1 ×1". A bare count says
    #: something is happening; this says what.
    workers: str = ""

    @property
    def clock(self) -> str:
        return format_duration(self.elapsed)


def format_duration(seconds: float) -> str:
    total = int(seconds)
    if total < 60:
        return f"{total}s"
    if total < 3600:
        return f"{total // 60}m{total % 60:02d}s"
    return f"{total // 3600}h{(total % 3600) // 60:02d}m"


class Watcher:
    """Polls the process tree under a build and reports per-package state.

    Completion is derived from observation rather than artefacts: paru's
    `CleanAfter` deletes the built package, so a package that was seen building
    and now has no processes is finished. Artefacts are used only to corroborate.
    """

    def __init__(
        self, clone_dir: Path, known: set[str] | None = None, *, include_root: bool = False
    ) -> None:
        self.clone_dir = clone_dir
        self.known = known
        self.include_root = include_root
        self.seen: set[str] = set()
        #: pending, drained by `new_finishes`
        self.finished: list[str] = []
        #: everything ever handed to the caller. `finished` alone cannot serve
        #: as both the queue and the record: clearing it to report means the
        #: next poll re-finishes the same package, which shows up as a duplicate
        #: row in the feed.
        self.reported: set[str] = set()

    def poll(self, root_pid: int | None) -> list[BuildState]:
        """Observe the tree. `root_pid=None` means it is gone, which drains.

        This is how a package that was still building when the launcher exited
        still reaches the feed, instead of being silently lost.
        """
        procs = snapshot()
        children = (
            descendants(root_pid, procs, include_root=self.include_root)
            if root_pid is not None
            else []
        )

        grouped: dict[str, list[Proc]] = {}
        for proc in children:
            pkg = package_of(proc.cwd, self.clone_dir, self.known)
            if pkg:
                grouped.setdefault(pkg, []).append(proc)

        # A package we have seen building and that no longer has processes is
        # done. `makepkg` holds the package's clone directory as its working
        # directory for the whole build, so a package only disappears from the
        # tree when it has genuinely finished.
        for pkg in sorted(self.seen - set(grouped)):
            if pkg in self.reported:
                continue
            self.reported.add(pkg)
            self.finished.append(pkg)
        self.seen |= set(grouped)

        states = []
        for pkg, group in sorted(grouped.items()):
            # elapsed is measured from the oldest process still alive for this
            # package, which is the honest lower bound on its build time
            oldest = min(p.started for p in group)
            compilers = sum(1 for p in group if classify([p]) == "compile")
            states.append(
                BuildState(
                    package=pkg,
                    phase=classify(group),
                    elapsed=time.time() - oldest,
                    jobs=compilers or len(group),
                    workers=summarise_workers(group),
                )
            )
        return states

    def new_finishes(self) -> list[str]:
        out = self.finished
        self.finished = []
        return out


# --------------------------------------------------------------------------
# repo-phase download progress (real bytes, no guessing)
# --------------------------------------------------------------------------


def download_progress(cache_dir: Path = Path("/var/cache/pacman/pkg")) -> tuple[int, int]:
    """(bytes currently staged, number of in-flight .part files).

    pacman writes `*.part` while downloading and renames on completion, so the
    sum of those sizes is a real progress figure rather than an animation.
    """
    total = 0
    count = 0
    try:
        for entry in os.scandir(cache_dir):
            if entry.name.endswith(".part"):
                try:
                    total += entry.stat().st_size
                    count += 1
                except OSError:
                    continue
    except OSError:
        return 0, 0
    return total, count