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

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

import asyncio
import os
import shutil
import subprocess
import sys
from pathlib import Path

# ── 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.

TASK_NAME = "MeshBay Node"   # the name the Electron client shows


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 _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, detached and 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")
    subprocess.Popen([exe], creationflags=0x00000008 | 0x08000000,  # DETACHED | NO_WINDOW
                     close_fds=True)


def autostart_end() -> None:
    """Stop any running daemon (hard: there is no CTRL_CLOSE handler yet)."""
    if autostart_supported():
        subprocess.run(["taskkill", "/IM", "meshbay-node.exe", "/F"],
                       capture_output=True)