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 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"


def log_file() -> Path | None:
    """Where the daemon logs on Windows; None elsewhere, where journald has it.

    A daemon started by the service task or the Startup launcher has no console:
    its stderr goes nowhere, and a node that fails there fails in silence.
    """
    if sys.platform == "win32":
        return state_dir() / "node.log"
    return None


# ── 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. The daemon reads it in place, beneath
    <config>/node.env, so it reaches every node however that node was set up --
    `meshbay-node init` and the desktop client's onboarding alike -- and an
    operator's own node.env still wins.

    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 _load_env_file(path: Path) -> int:
    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


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

    systemd does the first 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, and node.env wins over the packaged default -- an
    operator exporting a value, or systemd having already loaded the same file,
    overrides the packaged default rather than being overridden by it.

    The packaged default used to reach a node only as a copy made by
    `meshbay-node init`; a node onboarded by the desktop client never ran it and
    ran with no TMDB token at all.
    """
    count = _load_env_file(source_dir / "node.env")
    packaged = packaged_default_env()
    if packaged is not None:
        count += _load_env_file(packaged)
    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_alive(pid: int) -> bool:
    """Whether a process with this pid exists, in any session (tasklist lists
    session 0 too, where a service-mode daemon runs; opening it would not)."""
    if sys.platform != "win32":
        # A zombie counts as gone: it has exited, and only its parent has not
        # reaped it yet -- os.kill(pid, 0) would still find it.
        try:
            stat = Path(f"/proc/{pid}/stat").read_text(encoding="ascii")
        except OSError:
            return False
        return stat.rsplit(")", 1)[1].split()[0] != "Z"
    r = subprocess.run(["tasklist", "/FI", f"PID eq {pid}", "/NH", "/FO", "CSV"],
                       capture_output=True, text=True)
    return f'"{pid}"' in r.stdout


def _node_exe() -> str | None:
    """The meshbay-node to launch: this very one when frozen, then the launcher
    next to the interpreter (a venv's Scripts/), then argv[0], then PATH.

    PATH was first, and a machine with two copies on it -- an MSIX and an NSIS
    install, an older one left behind -- had `autostart start` from one launch
    the other, a different version, found by testing a real install.
    """
    if getattr(sys, "frozen", False):
        return sys.executable
    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 shutil.which("meshbay-node")


def request_graceful_stop(data_dir: Path, ui_port: int, timeout: float = 15.0) -> bool:
    """Ask the running daemon to stop through its control API, and wait until
    its process is gone. True once it is; False when nothing answered or it
    did not exit in time -- the caller then forces it.

    The one stop that reaches a daemon in any session with no elevation, and
    runs its _shutdown() (WebRTC sessions closed, transcodes stopped). CTRL_BREAK
    did neither across sessions, and aimed at a process on another console it
    reached every process on the caller's own -- the CLI killed itself.
    """
    import json
    import urllib.request

    try:
        token = (data_dir / "ui-token").read_text(encoding="utf-8").strip()
    except OSError:
        return False
    base = f"http://127.0.0.1:{ui_port}"
    try:
        with urllib.request.urlopen(f"{base}/api/status?t={token}", timeout=3) as r:
            pid = json.loads(r.read()).get("pid")
        req = urllib.request.Request(f"{base}/api/shutdown?t={token}", method="POST", data=b"")
        with urllib.request.urlopen(req, timeout=3):
            pass
    except (OSError, ValueError):
        return False
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        if pid is None:
            try:
                urllib.request.urlopen(f"{base}/api/status?t={token}", timeout=1)
            except OSError:
                return True
        elif not _pid_alive(int(pid)):
            return True
        time.sleep(0.3)
    log.warning("the node did not exit within %.0fs of being asked to stop", timeout)
    return False


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_NO_WINDOW: a hidden console of its own, which still delivers
    # CTRL_LOGOFF/SHUTDOWN to install_console_close_handler, so signing out
    # stops the node properly. close_fds: nothing of the caller's is inherited
    # -- the desktop app starts its node through here for that reason, since a
    # child of Electron inherited Electron's sockets and held them after it quit.
    subprocess.Popen([exe], creationflags=0x00000200 | 0x08000000, close_fds=True)


NODE_IMAGE = "meshbay-node.exe"


def autostart_end() -> None:
    """
    Force-stop a daemon running in this session: the fallback after
    request_graceful_stop(), which callers try first.

    It used to send CTRL_BREAK_EVENT to the pid autostart_run() recorded. That
    process has a console of its own (CREATE_NO_WINDOW), so it is not a group
    on the caller's console -- and GenerateConsoleCtrlEvent aimed at such a pid
    reaches every process on the caller's console instead: `autostart stop`
    and `restart-daemon` killed themselves, the latter before restarting
    anything (found by running both against a real install).
    `taskkill /F` is TerminateProcess: cannot be caught, cannot reach session 0.

    Never this process or its parent: the CLI is meshbay-node.exe too (and a
    venv's launcher is its parent), and `taskkill /IM meshbay-node.exe` killed
    the very command that ran it -- `autostart stop` and `restart-daemon`
    exited 1 in silence, the latter with the node down.
    `/T` because a venv's meshbay-node.exe is a launcher whose python.exe child
    is the daemon, and a frozen daemon's children are its ffmpeg transcodes.
    """
    if not autostart_supported():
        return
    argv = ["taskkill", "/F", "/T", "/IM", NODE_IMAGE]
    for pid in {os.getpid(), os.getppid()}:
        argv += ["/FI", f"PID ne {pid}"]
    subprocess.run(argv, 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

# Task Scheduler's defaults end a task after 72 hours, never start it on
# battery and stop it when the power cable comes out — a node on a laptop was
# down for any of the three. Kept identical to packaging/win/service.ps1.
SERVICE_TASK_SETTINGS = (
    "New-ScheduledTaskSettingsSet -ExecutionTimeLimit ([TimeSpan]::Zero) "
    "-AllowStartIfOnBatteries -DontStopIfGoingOnBatteries "
    "-MultipleInstances IgnoreNew -StartWhenAvailable"
)


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_state() -> str:
    """Task Scheduler's word for the task ('Running', 'Ready', ...), '' if absent."""
    return service_status()["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; "
        f"$s = {SERVICE_TASK_SETTINGS}; "
        "Register-ScheduledTask -TaskName $env:MESHBAY_SVC_TASK "
        "-Action $a -Trigger $t -Principal $p -Settings $s -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("/end", "/tn", TASK_NAME)   # deleting does not end the instance
        _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; request_graceful_stop() is how
# every Stop avoids relying on it.

_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