1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
|
"""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"
# ── 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)
|