"""Run the update, emitting newline-delimited JSON on stdout.

The front end is a separate process, so the two halves need a protocol. This is
it: one JSON object per line, nothing else.

**stdout is JSON only.** Anything a human would want to read goes to stderr or
into the per-phase log file. A stray print on stdout is a parse failure in the Go
side, and this codebase has already been bitten once by a tool that printed an
error and still exited zero (see flatpak.py).

Events:

    {"event":"start",      "phases":[...], "total":N}
    {"event":"phase_start","phase":"repo","command":"pacman -Syu","total":N}
    {"event":"build",      "packages":[{"name":..,"phase":..,"elapsed":..,"jobs":..}],
                           "download_bytes":N,"download_rate":F,"done":N}
    {"event":"finished",   "name":..,"elapsed":F,"state":"done","version":..,
                           "shared":bool}
    {"event":"failed",     "name":..,"reason":..}
    {"event":"recovery",   "label":..,"argv":[..],"note":..}
    {"event":"phase_end",  "phase":"repo","returncode":0,"elapsed":F}
    {"event":"error",      "message":.., "details":[..]}
    {"event":"done",       "returncode":0,"elapsed":F}
"""

from __future__ import annotations

import contextlib
import json
import os
import signal
import subprocess
import sys
import time
from pathlib import Path

from .live import BuildObserver, build_env, spawn_kwargs, terminate

POLL_HZ = 5.0

#: a child that survived SIGKILL: the phase did not finish and did not stop
EXIT_STILL_RUNNING = 125


#: How much of the progress bar a package that actually builds is worth,
#: against one that only downloads and installs. A build runs for minutes and a
#: download for about a second, so without this the bar raced to nearly full
#: during the repository phase -- every package there finishes in no time --
#: and the estimate that follows from it was meaningless.
BUILD_WEIGHT = 60
PLAIN_WEIGHT = 1


class SkipRequest:
    """A request to abandon the package in progress and go on to the next.

    Set from a signal handler, which may only touch simple state, and taken by
    the supervise loop between polls. One package per invocation is what makes
    this possible at all: inside a single transaction there is nothing to skip
    without unpicking what pacman has already done.
    """

    def __init__(self) -> None:
        self._asked = False

    def ask(self) -> None:
        self._asked = True

    def take(self) -> bool:
        asked, self._asked = self._asked, False
        return asked


#: the process-wide request; a signal handler has nowhere else to put it
SKIP = SkipRequest()


def request_skip(*_args: object) -> None:
    """Handler for the signal the dashboard sends when `s` is pressed."""
    SKIP.ask()


def listen_for_skip() -> None:
    """Route the skip signal to the request.

    Installed only where something can act on it, so an ordinary CLI run keeps
    the default disposition and an unexpected SIGUSR1 still stops it.
    """
    with contextlib.suppress(ValueError, OSError):
        signal.signal(signal.SIGUSR1, request_skip)


class Emitter:
    """Writes one JSON object per line and flushes, so the reader sees it live.

    A dead reader must never interrupt a running transaction. If the front end
    exits, this process's stdout pipe breaks; letting that exception escape would
    unwind the phase loop into the `finally` that terminates pacman, killing an
    upgrade halfway through. That is precisely how a system ends up in a
    half-upgraded state, so a broken pipe silences the reporting and nothing else.
    """

    def __init__(self, stream=None) -> None:
        self.stream = stream or sys.stdout
        self.alive = True

    def _die(self) -> None:
        """Stop reporting and make the stream safe to write to again.

        Redirecting the descriptor to /dev/null is the documented way to handle
        this: without it the interpreter prints its own BrokenPipeError noise on
        exit, from a place no handler of ours can reach.
        """
        self.alive = False
        try:
            fileno = self.stream.fileno()
        except (AttributeError, OSError, ValueError):
            return
        try:
            devnull = os.open(os.devnull, os.O_WRONLY)
            os.dup2(devnull, fileno)
            os.close(devnull)
        except OSError:
            pass

    def __call__(self, event: str, **fields) -> None:
        if not self.alive:
            return
        payload = {"event": event}
        payload.update(fields)
        try:
            self.stream.write(json.dumps(payload, separators=(",", ":")) + "\n")
            self.stream.flush()
        except (BrokenPipeError, ValueError, OSError):
            self._die()

    def human(self, message: str) -> None:
        """Anything a person should read. Never stdout."""
        try:
            print(message, file=sys.stderr, flush=True)
        except (BrokenPipeError, ValueError, OSError):
            pass


class StreamingRunner:
    """Runs one phase while emitting build state, without rendering anything."""

    #: most lines of a command's output to send in one tick. A cap rather than
    #: a window: the interface keeps what it has been sent, so nothing is lost
    #: except a burst larger than this between two polls.
    TAIL_LINES = 200

    def __init__(
        self,
        observer: BuildObserver,
        emit: Emitter,
        *,
        poll_hz: float = POLL_HZ,
        workers: int = 1,
        nproc: int | None = None,
    ) -> None:
        self.observer = observer
        self.emit = emit
        self.poll_hz = poll_hz
        self.workers = workers
        self.nproc = nproc
        self._pid: int | None = None
        self._log: Path | None = None
        self._log_seen: Path | None = None
        self._log_offset = 0
        #: set when the run was asked to stop. The interrupt is caught around
        #: the child so that the child can be shut down cleanly, which means
        #: the caller has to be told: catching it here otherwise looked like a
        #: phase that had simply finished, and a phase made of many
        #: invocations carried on to the next one.
        self.interrupted = False
        self.skipped = False

    def run(
        self,
        argv: list[str],
        phase: str,
        log_path: Path,
        *,
        total: int | None = None,
        since: float | None = None,
    ) -> int:
        """`since` is when the whole run began, for a cumulative elapsed.

        A phase's own duration is meaningless once several invocations make up
        one phase: reporting it made the clock read 19s while a package had
        been building for nearly two minutes.
        """
        log_path.parent.mkdir(parents=True, exist_ok=True)
        self._log = log_path
        # Per package, not per runner. One runner makes every invocation in a
        # phase, so a flag left set made a single skip look like a skip of
        # every package that followed it. `interrupted` is deliberately not
        # reset: once the run has been asked to stop, it has stopped.
        self.skipped = False

        started = time.monotonic()
        began = since if since is not None else started
        count = total if total is not None else self.observer.total
        self.emit(
            "phase_start", phase=phase, command=" ".join(argv), total=count, log=str(log_path)
        )

        with open(log_path, "wb") as log:
            proc = subprocess.Popen(  # noqa: S603 - argv is built by the caller
                argv, **spawn_kwargs(build_env(self.workers, self.nproc), log)
            )
            self._pid = proc.pid
            try:
                while proc.poll() is None:
                    if SKIP.take():
                        # the caller decides what a skip means; the runner
                        # only stops this child and says that it did
                        self.skipped = True
                        terminate(proc)
                        break
                    self._tick(proc.pid, phase)
                    time.sleep(1.0 / self.poll_hz)
            except KeyboardInterrupt:
                self.interrupted = True
                # No human line here: the caller reports the stop as a proper
                # event, and a human line is echoed back into the interface as a
                # second, identical warning.
                terminate(proc)
            finally:
                if proc.poll() is None:
                    terminate(proc)
                self._pid = None

        # drain whatever landed between the last tick and exit, so a package
        # that finished at the very end still reaches the feed
        self._tick(None, phase)

        elapsed = time.monotonic() - began
        # A child that outlived SIGKILL is not a success. `returncode or 0`
        # turned None into 0, so the run reported the phase as done and started
        # the next one beside a live pacman transaction.
        code = proc.returncode
        if code is None:
            self.interrupted = True
            code = EXIT_STILL_RUNNING
        self.emit("phase_end", phase=phase, returncode=code, elapsed=round(elapsed, 2))
        return code

    def _new_log_lines(self) -> list[str]:
        """Lines the running command printed since the last tick.

        Sending the *last* few lines each time lost anything a burst produced
        between two polls, and meant the interface could never show more than a
        window of the output. Tracking the read offset instead makes the stream
        whole, so it can be read back in full.
        """
        if self._log is None:
            return []
        if self._log != self._log_seen:
            # a new package has its own log; start from the beginning of it
            self._log_seen = self._log
            self._log_offset = 0
        try:
            if self._log.stat().st_size < self._log_offset:
                self._log_offset = 0  # truncated or replaced
            with open(self._log, "rb") as handle:
                handle.seek(self._log_offset)
                chunk = handle.read(1 << 20)
        except OSError:
            return []

        # Work in bytes: a partial final line has to be left for the next tick,
        # and re-decoding with replacement would not tell us how many there were.
        cut = chunk.rfind(b"\n")
        if cut < 0:
            return []
        text = chunk[: cut + 1].decode("utf-8", errors="replace")
        self._log_offset += cut + 1
        lines = [line.rstrip() for line in text.splitlines() if line.strip()]
        return [line[:400] for line in lines[-self.TAIL_LINES :]]

    def _tick(self, pid: int | None, phase: str) -> None:
        finished = self.observer.poll(pid)
        for entry in finished:
            # the weight travels with the result, so the consumer can weigh
            # progress by work rather than by package
            self.emit(
                "finished",
                name=entry.name,
                elapsed=round(entry.elapsed, 2),
                state=entry.state,
                version=entry.version,
                shared=entry.shared,
            )
        states = [
            {
                "name": state.package,
                "phase": state.phase,
                "elapsed": round(state.elapsed, 2),
                "jobs": state.jobs,
                "workers": getattr(state, "workers", ""),
            }
            for state in self.observer.states.values()
        ]
        done = sum(1 for e in self.observer.feed if e.state != "fail")
        self.emit(
            "build",
            phase=phase,
            packages=states,
            download_bytes=self.observer.download_bytes,
            download_rate=round(self.observer.download_rate, 1),
            done=done,
            total=self.observer.total,
            log_lines=self._new_log_lines(),
        )


# --------------------------------------------------------------------------
# phase planning
# --------------------------------------------------------------------------


def repo_command(held: list[str], *, refresh: bool = True) -> list[str]:
    """`-Syu` with `--ignore`, which is exactly the IgnorePkg mechanism.

    Holding a package back makes the run a partial upgrade, which Arch does not
    support. It is still offered because it is occasionally what you want, but
    the caller is expected to have said so out loud first.
    """
    # `-y` refreshes the databases first. The plan is built from the databases
    # on disk, so refreshing here installs from a set the reader never saw --
    # the same option has to decide it in both places.
    argv = ["sudo", "pacman", "-Syu" if refresh else "-Su", "--noconfirm"]
    if held:
        argv.append("--ignore=" + ",".join(sorted(held)))
    return argv


def aur_command(targets: list[str]) -> list[str]:
    """An explicit target list rather than `-Sua`.

    `-Sua` updates everything and offers no way to leave one package out, so
    skipping a package requires naming the ones to include.

    `--nobatchinstall` is deliberate, and is not paru's default. By default paru
    builds every package and then hands the whole set to pacman as a *single*
    transaction, which makes one bad package poison the entire update: a single
    unresolved conflict meant none of twenty-seven packages installed. paru's
    documented alternative is to install each package as soon as it is built, so
    a failure costs one package instead of all of them.

    `--failfast` is deliberately absent too: paru's default is to carry on after
    a build fails, which is the behaviour wanted here.
    """
    return ["paru", "-S", "--needed", "--noconfirm", "--nobatchinstall", *sorted(targets)]


def aur_command_interactive(targets: list[str]) -> list[str]:
    """The same target list, but with the prompts left switched on.

    Some questions have no correct automatic answer -- "replace the installed
    scenefx0.5 with scenefx?" is a decision, not a default. Those can only be
    put to a human, which needs the real terminal the interface normally keeps
    to itself.
    """
    return ["paru", "-S", "--needed", "--nobatchinstall", *sorted(targets)]


def flatpak_command() -> list[str]:
    return ["flatpak", "update", "--noninteractive", "-y"]


def installed_progress(updates) -> int:
    """How many planned updates have actually landed.

    The repo phase has no build graph, so progress has to be measured: a package
    is done when its new version directory exists in the local database.
    """
    local = Path("/var/lib/pacman/local")
    done = 0
    for update in updates:
        if (local / f"{update.name}-{update.new}").is_dir():
            done += 1
    return done