path: root/packages/meshbay-node/src/meshbay_node/platform.py

"""Platform-specific paths and tool resolution for meshbay-node."""

import asyncio
import logging
import os
import shutil
import signal
import subprocess
import sys
import threading
import time
from pathlib import Path

log = logging.getLogger(__name__)

# ── Console ──────────────────────────────────────────────────────────────────


def force_utf8_stdio() -> None:
    """
    Make stdout/stderr UTF-8. A Windows console is cp1252 by default, so any
    ``print()`` carrying a character outside it — the ``->`` arrows and em
    dashes the CLI help and messages are full of — raises UnicodeEncodeError
    and takes the command down with it. No effect where the streams are
    already UTF-8 or cannot be reconfigured.
    """
    for stream in (sys.stdout, sys.stderr):
        try:
            stream.reconfigure(encoding="utf-8")
        except (AttributeError, ValueError, OSError):
            pass


# ── Event loop ───────────────────────────────────────────────────────────────


def configure_event_loop() -> None:
    """
    The daemon runs on Windows' default ProactorEventLoop: verified end to end
    (a live browser peer connecting, an index sync, a file download and an
    ffmpeg-transcoded video stream). aiortc only ever hangs on it in the
    *same-process loopback* the tests use, which the test suite handles on its
    own (repo-root conftest).

    Escape hatch, opt-in only: MESHBAY_NODE_EVENT_LOOP=selector switches to the
    SelectorEventLoop. That fixes aiortc-in-one-process but breaks ffmpeg
    (SelectorEventLoop cannot spawn subprocesses on Windows), so it is not the
    default and probably never should be.
    """
    if sys.platform != "win32":
        return
    if os.environ.get("MESHBAY_NODE_EVENT_LOOP", "").lower() == "selector":
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())


# ── Directories ──────────────────────────────────────────────────────────────


def config_dir() -> Path:
    if sys.platform == "win32":
        return Path(os.environ.get("LOCALAPPDATA") or Path.home()) / "meshbay"
    return Path.home() / ".config" / "meshbay"


def data_dir() -> Path:
    if sys.platform == "win32":
        return Path(os.environ.get("LOCALAPPDATA") or Path.home()) / "meshbay" / "data"
    return Path.home() / ".local" / "share" / "meshbay"


def state_dir() -> Path:
    if sys.platform == "win32":
        return Path(os.environ.get("LOCALAPPDATA") or Path.home()) / "meshbay" / "state"
    return Path.home() / ".local" / "state" / "meshbay"


# ── Packaged defaults ────────────────────────────────────────────────────────


def packaged_default_env() -> Path | None:
    """
    The `default.env` shipped with the package: build-time defaults, currently
    the shared read-only TMDB token. `init` copies it to config_dir()/node.env
    and nothing reads it in place, so an operator's edits to their own copy
    survive an upgrade.

    Frozen (PyInstaller/Windows): beside the executable, where
    build-node-runtime.ps1 puts it -- the same placement it uses for ffmpeg.
    Packaged (Linux): /opt/meshbay-node/share/default.env, from build-node.sh.
    None in a source checkout, where no package wrote one.
    """
    candidates = []
    if getattr(sys, "frozen", False):
        candidates.append(Path(sys.executable).parent / "default.env")
    candidates.append(Path("/opt/meshbay-node/share/default.env"))
    for path in candidates:
        try:
            if path.is_file():
                return path
        except OSError:
            continue
    return None


def install_node_env(target_dir: Path) -> Path | None:
    """
    Copy the packaged default.env to <target_dir>/node.env, once, at init.

    Never overwrites: an existing node.env holds the operator's own values, and
    silently replacing a configured token with the packaged one would be worse
    than doing nothing. Returns the path when written, None when there was
    nothing to copy or a file was already there.
    """
    src = packaged_default_env()
    if src is None:
        return None
    dest = target_dir / "node.env"
    if dest.exists():
        return None
    dest.write_bytes(src.read_bytes())
    chmod_private(dest)
    return dest


def load_node_env(source_dir: Path) -> int:
    """
    Read <source_dir>/node.env into os.environ, returning how many names were
    set.

    systemd does this on Linux through `EnvironmentFile=`, but the Windows
    autostart is a Startup-folder .vbs with no equivalent, so the daemon reads
    the file itself and both platforms behave the same. An existing environment
    variable always wins -- an operator exporting a value, or systemd having
    already loaded the same file, overrides the packaged default rather than
    being overridden by it.
    """
    path = source_dir / "node.env"
    try:
        text = path.read_text(encoding="utf-8")
    except (OSError, UnicodeDecodeError):
        return 0
    count = 0
    for line in text.splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        name, _, value = line.partition("=")
        name = name.strip()
        value = value.strip().strip('"').strip("'")
        if not name or name in os.environ:
            continue
        os.environ[name] = value
        count += 1
    return count


# ── File permissions ─────────────────────────────────────────────────────────


def chmod_private(path: Path, *, mode: int = 0o600) -> None:
    """Set restrictive permissions on a file. No-op on Windows (NTFS ignores mode bits)."""
    if sys.platform != "win32":
        path.chmod(mode)


# ── Media tools ──────────────────────────────────────────────────────────────

_ffmpeg_path: str = "ffmpeg"
_ffprobe_path: str = "ffprobe"


def check_media_tools(
    ffmpeg: str = "ffmpeg", ffprobe: str = "ffprobe",
) -> None:
    """Resolve ffmpeg/ffprobe at daemon startup. Raises RuntimeError if not found."""
    global _ffmpeg_path, _ffprobe_path
    resolved = shutil.which(ffmpeg)
    if not resolved:
        raise RuntimeError(
            f"{ffmpeg!r} not found in PATH. "
            "Install ffmpeg or set [node] ffmpeg_path in node.toml."
        )
    _ffmpeg_path = resolved
    resolved = shutil.which(ffprobe)
    if not resolved:
        raise RuntimeError(
            f"{ffprobe!r} not found in PATH. "
            "Install ffmpeg or set [node] ffprobe_path in node.toml."
        )
    _ffprobe_path = resolved


def ffmpeg_cmd() -> str:
    return _ffmpeg_path


def ffprobe_cmd() -> str:
    return _ffprobe_path


# ── Autostart (Windows) ──────────────────────────────────────────────────────
#
# The Windows stand-in for the Linux `systemctl --user` unit. Task Scheduler
# would be nicer (retry semantics), but a logon-triggered task needs elevation
# to create — and this must work for an ordinary user with no admin rights.
# So: a `.vbs` launcher in the per-user Startup folder. wscript runs it hidden
# (Run(..., 0, ...)) at every sign-in; no console window, no admin, no
# third-party dependency. See the Service mode section below for the
# boot-capable, admin-once alternative built on top of Task Scheduler instead.


def autostart_supported() -> bool:
    return sys.platform == "win32"


def _startup_vbs() -> Path:
    base = os.environ.get("APPDATA") or str(Path.home() / "AppData" / "Roaming")
    return (Path(base) / "Microsoft" / "Windows" / "Start Menu" / "Programs"
            / "Startup" / "MeshBay Node.vbs")


def _pid_file() -> Path:
    """Where autostart_run() records the pid it spawned, for autostart_end()
    to signal later -- possibly from a different process (a new Electron
    session, or a fresh CLI invocation), so this cannot be an in-memory
    handle."""
    return state_dir() / "node.pid"


def _pid_is_meshbay_node(pid: int) -> bool:
    """True if `pid` is currently running *and* is meshbay-node.exe. Guards
    against a stale pidfile whose pid Windows has since handed to an
    unrelated process -- autostart_end() would otherwise send CTRL_BREAK_EVENT
    to whatever that is instead."""
    r = subprocess.run(["tasklist", "/FI", f"PID eq {pid}", "/NH"],
                       capture_output=True, text=True)
    return "meshbay-node.exe" in r.stdout.lower()


def _node_exe() -> str | None:
    """Best guess at the meshbay-node launcher: PATH first, then next to the
    interpreter (a venv's Scripts/ dir, or a bundled runtime), then argv[0]."""
    found = shutil.which("meshbay-node")
    if found:
        return found
    for cand in (Path(sys.executable).parent / "meshbay-node.exe",
                 Path(sys.argv[0])):
        if cand.name.lower().startswith("meshbay-node") and cand.exists():
            return str(cand.resolve())
    return None


def autostart_status() -> dict:
    """{'installed': bool, 'state': str}. 'state' is left empty — there is no
    Task Scheduler to ask 'is it running'; the Node page probes the daemon."""
    if not autostart_supported():
        return {"installed": False, "state": ""}
    return {"installed": _startup_vbs().exists(), "state": ""}


def autostart_install(exe: str | None = None) -> None:
    """Write the Startup-folder launcher. Raises RuntimeError on failure."""
    if not autostart_supported():
        raise RuntimeError("autostart is Windows-only")
    exe = exe or _node_exe()
    if not exe:
        raise RuntimeError(
            "cannot locate the meshbay-node launcher — pass its path, or run "
            "this from where meshbay-node is on PATH")
    vbs = _startup_vbs()
    vbs.parent.mkdir(parents=True, exist_ok=True)
    # Chr(34) is a literal " — wraps the path so a space in it doesn't split the
    # command. 0 = hidden window, False = don't wait. (A Windows path cannot
    # itself contain ", so no further escaping is needed.)
    vbs.write_text(
        f'CreateObject("WScript.Shell").Run Chr(34) & "{exe}" & Chr(34), 0, False\n',
        encoding="utf-8", newline="\r\n")


def autostart_remove() -> None:
    """Delete the Startup-folder launcher if present."""
    if autostart_supported():
        _startup_vbs().unlink(missing_ok=True)


def autostart_run() -> None:
    """Start the daemon now, windowless. Raises RuntimeError if the launcher
    cannot be located."""
    if not autostart_supported():
        raise RuntimeError("autostart is Windows-only")
    exe = _node_exe()
    if not exe:
        raise RuntimeError("cannot locate the meshbay-node launcher")
    # CREATE_NEW_PROCESS_GROUP, not DETACHED_PROCESS: still no visible window
    # (CREATE_NO_WINDOW), but the child keeps a console object of its own and
    # becomes the root of its own process group -- what autostart_end() needs
    # to target it with CTRL_BREAK_EVENT instead of only ever a hard taskkill.
    # DETACHED_PROCESS has no console at all, so nothing could be signalled.
    proc = subprocess.Popen([exe], creationflags=0x00000200 | 0x08000000,
                            close_fds=True)
    try:
        pid_file = _pid_file()
        pid_file.parent.mkdir(parents=True, exist_ok=True)
        pid_file.write_text(str(proc.pid), encoding="utf-8")
    except OSError:
        pass  # best effort -- autostart_end() falls back to taskkill by image name


# How long autostart_end() waits for a graceful CTRL_BREAK_EVENT stop before
# giving up and force-killing. A chosen grace period, not an OS-enforced one
# (unlike the ~5 s Windows itself allows a CTRL_CLOSE/LOGOFF/SHUTDOWN handler,
# see install_console_close_handler below -- CTRL_BREAK carries no such ceiling).
_GRACEFUL_STOP_TIMEOUT_SECS = 5.0


def autostart_end() -> None:
    """
    Stop the running daemon.

    Tries a graceful stop first: CTRL_BREAK_EVENT to the pid autostart_run()
    recorded. Because that process is the root of its own group
    (CREATE_NEW_PROCESS_GROUP), daemon.py's own SIGBREAK handler turns this
    into the same stop_event.set() SIGINT/SIGTERM already use, running the
    real _shutdown() -- closes WebRTC sessions, kills any in-flight ffmpeg
    transcode. Falls back to a hard `taskkill /F`, by image name, when there
    is no pidfile, the recorded process is already gone, or it does not exit
    within the grace period -- same as before this existed, just no longer
    the only path. `taskkill /F` itself is TerminateProcess and cannot be made
    graceful; nothing can catch it, on any OS.
    """
    if not autostart_supported():
        return
    pid_file = _pid_file()
    try:
        pid = int(pid_file.read_text(encoding="utf-8").strip())
    except (OSError, ValueError):
        pid = None
    if pid is not None and not _pid_is_meshbay_node(pid):
        pid = None  # stale pidfile -- Windows may have reused the pid since
    if pid is not None:
        try:
            os.kill(pid, signal.CTRL_BREAK_EVENT)
        except OSError:
            pid = None  # already gone, or never existed
        else:
            deadline = time.monotonic() + _GRACEFUL_STOP_TIMEOUT_SECS
            while time.monotonic() < deadline:
                if not _pid_is_meshbay_node(pid):
                    pid_file.unlink(missing_ok=True)
                    return
                time.sleep(0.2)
            log.warning("pid %d did not exit within %.1fs of CTRL_BREAK_EVENT, "
                        "falling back to taskkill /F", pid, _GRACEFUL_STOP_TIMEOUT_SECS)
    pid_file.unlink(missing_ok=True)
    subprocess.run(["taskkill", "/IM", "meshbay-node.exe", "/F"],
                   capture_output=True)


# ── Service mode (Windows, opt-in at install time) ───────────────────────────
#
# The Startup-folder .vbs above only ever runs after *this* user signs in. A
# real Windows Service would run before anyone signs in, but under
# LocalSystem/NetworkService — accounts with no normal user profile, so
# %LOCALAPPDATA%\meshbay\ (config, keystore, data) would not exist for it.
# Relocating storage to make that work is real surgery (Phase 2, deliberately
# not this).
#
# The middle ground: a Scheduled Task, created once with admin rights, that
# runs *as this user* at system boot without needing them to sign in first.
# That requires an S4U (Service For User) logon: no password stored anywhere,
# and unlike LocalSystem it loads this account's own profile, so
# config_dir()/data_dir() need no special-casing at all. The cost: S4U
# carries no *network* credential (no reaching a domain share as this user),
# which the node never needed anyway — everything it touches is local disk
# plus outbound internet.
#
# Getting S4U out of raw `schtasks.exe /create` means inferring it from
# whether `/rp` is present and what it holds -- undocumented, and it went
# wrong twice on this exact machine's blank-password account (common on a
# personal PC — confirmed via `net user`, "Password required: No"):
# `/rp ""` routes through credential validation, which Windows' default
# policy blocks for a blank password ("WARNING: When the run-as password is
# empty..." then "ERROR: The user name or password is incorrect.", 2026-09-05
# repro); omitting `/rp` entirely does get past that, but registers
# `Logon Mode: Interactive only` instead of S4U — confirmed live the same
# day: the task never ran at boot, and manually running it while signed in
# still failed (`Last Result: -2147024894`, no process ever launched).
# `Register-ScheduledTask` from the `ScheduledTasks` PowerShell module takes
# `-LogonType S4U` as a named, explicit value — no inference, no ambiguity —
# so that is what actually creates the task, shelling out to `powershell.exe`
# instead of `schtasks.exe` for this one call. Still unverified end-to-end
# past "the syntax runs and denies access when not elevated" (2026-09-05) —
# needs a real elevated install + reboot to confirm S4U registers as such and
# the task actually launches the process.
#
# Creating the task needs admin (a boot-trigger touches system-wide scheduler
# state, the same reason /sc onlogon did — see the autostart section above).
# Querying, running and ending an *already-created* task, as the same user it
# was registered for, does not — Task Scheduler grants the owner that much by
# default, which is what lets the Node page drive it with no further prompts.

TASK_NAME = "MeshBay Node"   # the Scheduled Task's own name


def service_supported() -> bool:
    return sys.platform == "win32"


def _current_user() -> str:
    domain = os.environ.get("USERDOMAIN") or os.environ.get("COMPUTERNAME") or "."
    user = os.environ.get("USERNAME") or ""
    return f"{domain}\\{user}" if user else ""


def _schtasks(*args: str) -> subprocess.CompletedProcess:
    return subprocess.run(["schtasks", *args], capture_output=True, text=True)


def service_status() -> dict:
    """{'installed': bool, 'state': str}. 'state' is Task Scheduler's own word
    ('Ready', 'Running', 'Disabled', ...), '' when not installed."""
    if not service_supported():
        return {"installed": False, "state": ""}
    r = _schtasks("/query", "/tn", TASK_NAME, "/fo", "list")
    if r.returncode != 0:
        return {"installed": False, "state": ""}
    state = ""
    for line in r.stdout.splitlines():
        if line.lower().startswith("status:"):
            state = line.split(":", 1)[1].strip()
            break
    return {"installed": True, "state": state}


def service_install(exe: str | None = None) -> None:
    """
    Register the boot-time Scheduled Task with an S4U logon. Needs admin —
    raises RuntimeError with Register-ScheduledTask's own message on failure,
    which is "Access is denied." when not elevated.

    Removes the per-user Startup launcher first, if present: the two
    mechanisms are mutually exclusive by design (both installed would start
    the daemon twice, once at boot and again at sign-in), and this is a
    separate front door from the Node page's own startup-mode selector (which
    enforces the same thing on its side) -- the CLI (`meshbay-node service
    install`) must not be able to leave that invariant broken.
    """
    if not service_supported():
        raise RuntimeError("service mode is Windows-only")
    autostart_remove()
    exe = exe or _node_exe()
    if not exe:
        raise RuntimeError(
            "cannot locate the meshbay-node launcher — pass its path, or run "
            "this from where meshbay-node is on PATH")
    user = _current_user()
    if not user:
        raise RuntimeError("could not determine the current user (USERNAME unset)")
    # Passed via the environment, not interpolated into the -Command string,
    # so a path or username containing a quote or $ can't break the script.
    env = {**os.environ, "MESHBAY_SVC_EXE": exe, "MESHBAY_SVC_USER": user,
           "MESHBAY_SVC_TASK": TASK_NAME}
    ps_script = (
        "$ErrorActionPreference = 'Stop'; "
        "$a = New-ScheduledTaskAction -Execute $env:MESHBAY_SVC_EXE; "
        "$t = New-ScheduledTaskTrigger -AtStartup; "
        "$p = New-ScheduledTaskPrincipal -UserId $env:MESHBAY_SVC_USER "
        "-LogonType S4U -RunLevel Limited; "
        "Register-ScheduledTask -TaskName $env:MESHBAY_SVC_TASK "
        "-Action $a -Trigger $t -Principal $p -Force | Out-Null"
    )
    r = subprocess.run(
        ["powershell", "-NoProfile", "-NonInteractive", "-Command", ps_script],
        capture_output=True, text=True, env=env)
    if r.returncode != 0:
        raise RuntimeError(
            f"Register-ScheduledTask failed: {r.stderr.strip() or r.stdout.strip()}")


def service_remove() -> None:
    """Delete the Scheduled Task if present. Needs admin; silent otherwise
    (mirrors autostart_remove — nothing to report if it was never installed)."""
    if service_supported():
        _schtasks("/delete", "/tn", TASK_NAME, "/f")


def service_run() -> None:
    """Start the task now. No admin needed for an already-registered task."""
    if service_supported():
        _schtasks("/run", "/tn", TASK_NAME)


def service_end() -> None:
    """Stop the running instance, if any. No admin needed."""
    if service_supported():
        _schtasks("/end", "/tn", TASK_NAME)


# ── Console close / logoff / shutdown handler (Windows) ──────────────────────
#
# CPython's own console handler claims CTRL_C_EVENT and CTRL_BREAK_EVENT --
# delivered as SIGINT/SIGBREAK, handled in daemon.py's win32 signal block --
# but returns "not handled" for CTRL_CLOSE_EVENT, CTRL_LOGOFF_EVENT and
# CTRL_SHUTDOWN_EVENT: there is no Python signal for any of the three. Without
# a handler of our own, Windows just ends the process for these -- no
# _shutdown(), no closed WebRTC sessions, no killed ffmpeg. Covers: closing
# the console window of an interactively-run `meshbay-node run`, user logoff,
# system shutdown. Does NOT cover `taskkill /F` -- TerminateProcess is
# uncatchable on any OS, the same as SIGKILL; see autostart_end() for how the
# Node page's Stop button avoids relying on it instead.

_CONSOLE_HANDLER_REFS: list = []  # ctypes callbacks must be kept referenced or they may be freed

CTRL_CLOSE_EVENT = 2
CTRL_LOGOFF_EVENT = 5
CTRL_SHUTDOWN_EVENT = 6


def install_console_close_handler(
    loop: asyncio.AbstractEventLoop, stop_event: asyncio.Event,
) -> "threading.Event | None":
    """
    Register the handler. Returns a threading.Event the caller must set once
    its own graceful shutdown has actually finished -- daemon.py does this
    right after `await self._shutdown()` -- or None off-Windows, or if
    registration itself failed (logged, not raised: losing this is a
    regression, refusing to start the daemon over it would not be).

    MSDN: for these three events the process is ended "after the process
    returns from the handler function, or after 5 seconds, whichever occurs
    first" -- so the handler, which Windows runs on a thread of its own and
    never the main one, blocks here instead of returning immediately, and
    nudges the asyncio loop the thread-safe way since it is not the loop's
    own thread. The wait is capped just under that ceiling so the process
    still exits by itself if cleanup runs long, rather than the OS treating an
    unresponsive handler as a hang.
    """
    if sys.platform != "win32":
        return None
    import ctypes
    from ctypes import wintypes

    handled = {CTRL_CLOSE_EVENT, CTRL_LOGOFF_EVENT, CTRL_SHUTDOWN_EVENT}
    shutdown_done = threading.Event()
    handler_type = ctypes.WINFUNCTYPE(wintypes.BOOL, wintypes.DWORD)

    def _handler(ctrl_type: int) -> bool:
        if ctrl_type not in handled:
            return False  # not ours -- let Python's own handler or the default action take it
        log.info("Console control event %d (close/logoff/shutdown) -- shutting down", ctrl_type)
        loop.call_soon_threadsafe(stop_event.set)
        shutdown_done.wait(4.5)
        return True

    handler_ref = handler_type(_handler)
    if not ctypes.windll.kernel32.SetConsoleCtrlHandler(handler_ref, True):
        log.warning("SetConsoleCtrlHandler failed (%s) -- closing the console window, "
                    "logging off or shutting down will not run a clean shutdown; "
                    "SIGINT/SIGTERM/SIGBREAK are unaffected",
                    ctypes.WinError())
        return None
    _CONSOLE_HANDLER_REFS.append(handler_ref)
    return shutdown_done