"""Start and stop TTS backend servers from the TUI hub.

Each backend's ``detect()`` returns a list of ``ServerSpec`` — the exact argv
(absolute binaries in the managed venv, no shell activation needed), the
working directory to spawn it in, and the URL to probe for readiness. This
module turns those specs into running processes: ``start`` spawns the server
(streaming its output to ``app/logs/<name>-server.log``, in the spec's cwd
when it has one — audio.cpp resolves model_specs/ relative to its process
working directory), records its pid, and polls the URL until it accepts
connections (model loads are slow, so the timeout is generous). With a spec
``identity`` the poll also verifies the server answers HTTP as that backend,
so readiness means "serving", not just "listening".

Progress reporting goes through an optional ``progress`` callback (see
``_console_progress`` for the event shapes); the default callback prints the
same lines as before, so the plain-console flow is unchanged. ``stop``
terminates the process group the hub started.

Everything here can run in the plain console tail after the curses TUI
returns (matching the wizards' build/pip streaming) or behind the run view's
boot screen, which renders the same events. Pid/log files live under
``app/logs/`` which is already gitignored.
"""

import os
import signal
import subprocess
import sys
import time
from datetime import datetime
from pathlib import Path
from typing import Callable, List, Optional

from backends import common, probe
from backends.common import APP_DIR

LOG_DIR = APP_DIR / "logs"

# How long to wait for a server to accept connections on its URL. First-time
# model loads (especially qwen-tts / faster-qwen3-tts pulling weights into
# VRAM) can take minutes, so this is deliberately generous.
SERVER_START_TIMEOUT = 600

# Grace period after SIGTERM before escalating to SIGKILL (POSIX).
STOP_GRACE_SECONDS = 10

# How often the start poll re-checks readiness (seconds).
POLL_INTERVAL = 1

# Progress callback: called with an event dict. KIND is one of:
#   "starting"  {name, argv, cwd, log_path, pid}  spawned, waiting for boot
#   "elapsed"   {name, seconds}                   heartbeat while waiting
#   "ready"     {name, url}                       server is up and answering
#   "exited"    {name, returncode, log_tail}      process exited while booting
#   "timeout"   {name, seconds, log_tail}         readiness deadline elapsed
#   "running"   {name, url}                       already up (no spawn)
#   "cancelled" {name}                            boot aborted via cancel
#   "error"     {message}                         could not spawn the executable
ProgressCallback = Optional[Callable[[dict], None]]


def _console_progress(event: dict) -> None:
    """Print PROGRESS events as the plain-console output (the old behavior)."""
    kind = event.get("kind")
    if kind == "starting":
        print(f"[INFO] starting {event['name']} server: {event['argv']}")
        if event.get("cwd"):
            print(f"[INFO] working directory: {event['cwd']}")
        print(f"[INFO] pid {event['pid']}; logs: {event['log_path']}")
    elif kind == "elapsed":
        print(f"[INFO] still waiting for the server ({int(event['seconds'])}s)...")
    elif kind == "ready":
        print(f"[OK] {event['name']} server is up on {event['url']}")
    elif kind == "running":
        print(f"[INFO] {event['name']} server already running on {event['url']}")
    elif kind == "cancelled":
        print(f"[INFO] {event['name']} server start cancelled")
    elif kind == "exited":
        print(f"[ERROR] {event['name']} server exited with code "
              f"{event['returncode']}")
        _print_tail(event.get("log_tail"))
    elif kind == "timeout":
        print(f"[ERROR] {event['name']} server did not start within "
              f"{int(event['seconds'])}s")
        _print_tail(event.get("log_tail"))
    elif kind == "error":
        print(f"[ERROR] {event['message']}")


def _log_path(name: str) -> Path:
    return LOG_DIR / f"{name}-server.log"


def _pid_path(name: str) -> Path:
    return LOG_DIR / f"{name}-server.pid"


def _read_log_tail(name: str, lines: int = 20) -> List[str]:
    """Return the last LINES of the server's log (best-effort)."""
    try:
        text = _log_path(name).read_text(encoding="utf-8", errors="replace")
    except OSError:
        return []
    return text.splitlines()[-lines:]


def _print_tail(tail: List[str]) -> None:
    """Print a log-tail event payload (used by the console callback)."""
    if tail:
        print(f"--- last {len(tail)} lines of the server log ---")
        print("\n".join(tail))
        print("---")


def _pid_alive(pid: int) -> bool:
    """True when a process with PID is still running (POSIX signal-0 probe)."""
    if sys.platform == "win32":
        try:
            import ctypes
            kernel32 = ctypes.windll.kernel32  # type: ignore[attr-defined]
            PROCESS_QUERY_LIMITED_INFORMATION = 0x1000
            handle = kernel32.OpenProcess(
                PROCESS_QUERY_LIMITED_INFORMATION, False, pid)
            if not handle:
                return False
            kernel32.CloseHandle(handle)
            return True
        except OSError:
            return False
    try:
        os.kill(pid, 0)
    except ProcessLookupError:
        return False
    except PermissionError:
        return True
    return True


def _reap_exited(pid: int) -> bool:
    """Reap PID when it is our exited child; True when confirmed dead.

    An exited child stays visible to signal-0 probes until its parent waits
    for it, and the hub never reaps between ``start`` and ``stop`` — so a
    server that honored SIGTERM would still count as alive and every stop
    would burn the whole grace period before escalating to SIGKILL. Returns
    False when the process may still be running or was not our child (the
    caller then falls back to its own liveness probes).
    """
    if not hasattr(os, "waitpid") or not hasattr(os, "WNOHANG"):
        return False
    try:
        waited, _status = os.waitpid(pid, os.WNOHANG)
    except ChildProcessError:
        # Not our child (or someone reaped it already): not ours to judge.
        return False
    except OSError:
        return False
    return waited == pid


def _kill_pid(pid: int) -> bool:
    """Terminate PID (and its process group on POSIX). Returns True when dead."""
    if sys.platform == "win32":
        try:
            os.kill(pid, signal.SIGTERM)
        except (ProcessLookupError, PermissionError, OSError):
            return not _pid_alive(pid)
        for _ in range(int(STOP_GRACE_SECONDS * 10)):
            if not _pid_alive(pid):
                return True
            time.sleep(0.1)
        try:
            os.kill(pid, signal.SIGTERM)
        except OSError:
            pass
        return not _pid_alive(pid)
    # POSIX: kill the whole process group (started with start_new_session=True).
    try:
        pgid = os.getpgid(pid)
    except ProcessLookupError:
        return True
    try:
        os.killpg(pgid, signal.SIGTERM)
    except ProcessLookupError:
        return True
    except PermissionError:
        return False
    for _ in range(int(STOP_GRACE_SECONDS * 10)):
        # Reap first so an exited (zombie) child ends the wait immediately
        # instead of keeping the killpg(0) probe "alive" until SIGKILL.
        if _reap_exited(pid):
            return True
        try:
            os.killpg(pgid, 0)
        except ProcessLookupError:
            return True
        except PermissionError:
            return False
        time.sleep(0.1)
    try:
        os.killpg(pgid, signal.SIGKILL)
    except (ProcessLookupError, PermissionError):
        pass
    return True


def _server_ready(spec) -> bool:
    """True when the server described by SPEC is usable, not just listening.

    Without an IDENTITY this is the plain TCP-connect check. With one, the
    server must also answer HTTP as that backend (``probe.identify_server``);
    for the faster backend (whose model loads after the port opens) the
    ``/health`` model_loaded flag must additionally be true.
    """
    if not common.server_running(spec.url):
        return False
    identity = getattr(spec, "identity", None)
    if identity is None:
        return True
    if probe.identify_server(spec.url) != identity:
        return False
    if identity == probe.IDENTITY_FASTER:
        return probe.faster_model_loaded(spec.url)
    return True


def start(spec, progress: ProgressCallback = None,
          cancel=None) -> bool:
    """Start the server described by SPEC (a ``backends.ServerSpec``).

    Spawns its argv with stdout/stderr to ``logs/<name>-server.log``, in the
    spec's CWD when it has one (audio.cpp discovers model_specs/ from its
    process working directory), records the pid, and polls readiness —
    ``_server_ready``, so an IDENTITY spec must actually answer HTTP — until
    it is up or ``SERVER_START_TIMEOUT`` elapses. Returns True when the
    server is up; on timeout or early exit reports the log tail and returns
    False. A no-op (True) when the server is already running.

    PROGRESS, when given, receives each boot event (see ProgressCallback);
    the default ``_console_progress`` prints them, preserving the old
    console output. CANCEL (a threading.Event) aborts the boot: the spawned
    process is terminated and False is reported (event kind "cancelled").
    """
    report = progress if progress is not None else _console_progress
    argv: List[str] = list(spec.argv)
    exe = Path(argv[0])
    if not exe.exists():
        report({"kind": "error",
                "message": f"server executable not found: {exe}. Run 'Set "
                           "up a backend' first."})
        return False
    if _server_ready(spec):
        report({"kind": "running", "name": spec.name, "url": spec.url})
        return True

    LOG_DIR.mkdir(parents=True, exist_ok=True)
    # Refuse to double-start: a live pid file means a previous start is
    # still booting (or its process is wedged). Spawning a second server
    # on the same port would orphan the first with no pid record left.
    if alive(spec.name):
        report({"kind": "error",
                "message": f"a {spec.name} server (pid "
                           f"{pid_for(spec.name)}) is already starting or "
                           "running; stop it first"})
        return False
    pid_file = _pid_path(spec.name)
    if pid_file.exists():
        try:
            pid_file.unlink()
        except OSError:
            pass

    cwd = getattr(spec, "cwd", None)
    # Append so an earlier boot's output survives (crash-loop debugging);
    # the child inherits the handle and the parent's copy is closed right
    # after the spawn, so nothing leaks here.
    log_handle = _log_path(spec.name).open("a", encoding="utf-8")
    popen_kwargs = {"stdout": log_handle, "stderr": subprocess.STDOUT}
    if cwd is not None:
        popen_kwargs["cwd"] = str(cwd)
    if sys.platform == "win32":
        popen_kwargs["creationflags"] = \
            subprocess.CREATE_NEW_PROCESS_GROUP  # type: ignore[attr-defined]
    else:
        popen_kwargs["start_new_session"] = True
    try:
        proc = subprocess.Popen(argv, **popen_kwargs)
    except OSError as exc:
        report({"kind": "error",
                "message": f"could not start server: {exc}"})
        log_handle.close()
        return False
    log_handle.write(f"\n=== boot {datetime.now():%Y-%m-%d %H:%M:%S} "
                     f"(pid {proc.pid}) ===\n")
    log_handle.flush()
    log_handle.close()

    pid_file.write_text(str(proc.pid), encoding="utf-8")
    report({"kind": "starting", "name": spec.name,
            "argv": " ".join(str(a) for a in argv),
            "cwd": str(cwd) if cwd is not None else None,
            "log_path": str(_log_path(spec.name)), "pid": proc.pid})

    started = time.time()
    next_heartbeat = started + 15
    deadline = started + SERVER_START_TIMEOUT
    while time.time() < deadline:
        if cancel is not None and cancel.is_set():
            # User cancelled while booting: kill what we spawned (the
            # server we started is not left loading in the background).
            _kill_pid(proc.pid)
            try:
                pid_file.unlink()
            except OSError:
                pass
            report({"kind": "cancelled", "name": spec.name})
            return False
        if proc.poll() is not None:
            report({"kind": "exited", "name": spec.name,
                    "returncode": proc.returncode,
                    "log_tail": _read_log_tail(spec.name)})
            try:
                pid_file.unlink()
            except OSError:
                pass
            return False
        if _server_ready(spec):
            report({"kind": "ready", "name": spec.name, "url": spec.url})
            return True
        if time.time() >= next_heartbeat:
            report({"kind": "elapsed",
                    "seconds": time.time() - started})
            next_heartbeat += 15
        time.sleep(POLL_INTERVAL)
    report({"kind": "timeout", "name": spec.name,
            "seconds": SERVER_START_TIMEOUT,
            "log_tail": _read_log_tail(spec.name)})
    # Leave the pid file in place so stop() can kill it (it may still load).
    return False


def stop(name: str) -> bool:
    """Stop a server previously started by ``start`` (identified by pid file).

    Returns True when the process was terminated (or already gone). Returns
    False when there is no pid file — the server was not started by this tool,
    so the user must stop it manually (e.g. close its terminal).
    """
    pid_file = _pid_path(name)
    if not pid_file.exists():
        print(f"[INFO] no pid file for '{name}' "
              "(not started by this tool — stop it manually)")
        return False
    try:
        pid = int(pid_file.read_text(encoding="utf-8").strip())
    except (OSError, ValueError):
        print(f"[WARNING] could not read pid file {pid_file}; removing it")
        try:
            pid_file.unlink()
        except OSError:
            pass
        return False
    if not _pid_alive(pid):
        print(f"[INFO] {name} server (pid {pid}) already stopped")
        try:
            pid_file.unlink()
        except OSError:
            pass
        return True
    print(f"[INFO] stopping {name} server (pid {pid})...")
    killed = _kill_pid(pid)
    if killed:
        print(f"[OK] {name} server stopped")
    else:
        print(f"[WARNING] could not stop pid {pid}; stop it manually")
    try:
        pid_file.unlink()
    except OSError:
        pass
    return killed


def manages(specs) -> bool:
    """True when any SPEC in the list was started (and is kept alive) by us.

    A server counts as ours when ``start`` recorded a pid file for it and
    that pid is still alive — the same ownership rule ``stop`` applies
    before refusing ("not started by this tool"). Used by the backends'
    ``detect()`` so the hub's status table can tag an up server as
    "[remote]" when it was launched outside this tool.
    """
    for spec in specs:
        pid = pid_for(spec.name)
        if pid is not None and _pid_alive(pid):
            return True
    return False


def pid_for(name: str):
    """Return the recorded pid for NAME, or None when no pid file exists."""
    pid_file = _pid_path(name)
    if not pid_file.exists():
        return None
    try:
        return int(pid_file.read_text(encoding="utf-8").strip())
    except (OSError, ValueError):
        return None


def alive(name: str) -> bool:
    """True when the server named NAME was started by us and is still alive."""
    pid = pid_for(name)
    return pid is not None and _pid_alive(pid)