""" MeshBay Node daemon — main process. Startup sequence: 1. Load config (~/.config/meshbay/node.toml) 2. Load or create keystore (Argon2id unlock) 3. Connect to hub: register → login → announce node 4. Fetch GEK bundle from hub (if group configured) 5. Start directory indexer (watchdog) 6. Create chat stores (one SQLite DB per group) 7. Create WebRTC transport (browser + native clients via DataChannel) 8. Start QUIC chunk server (LAN / port-forwarded / hub-less direct access) 9. (Phase 11.5: the unauthenticated HTTP file API and the TCP+TLS server were removed) 10. Start hub WebSocket (signaling, revocations, WebRTC offers) 11. Start local control API on node.ui_port (loopback only, token-gated) 12. Run until SIGINT/SIGTERM Usage: meshbay-node # interactive password prompt meshbay-node --config /path # custom config meshbay-node status # node state + public key (works while stopped) meshbay-node gek-init # initialise the group key (no browser needed) meshbay-node init # write example config + create keystore meshbay-node --calibrate-argon2 # benchmark Argon2id, suggest parameters """ import asyncio import base64 import logging import os import signal import sys import time from dataclasses import asdict, replace from pathlib import Path import uvicorn from meshbay_common import MNP_VERSION from meshbay_common.background import spawn from meshbay_common.paths import fold from meshbay_common.protocol import MNP from meshbay_node import uploads as uploads_mod from meshbay_node.audit import RETENTION_DAYS as AUDIT_RETENTION_DAYS from meshbay_node.audit import AuditStore from meshbay_node.bundle_store import BundleStore from meshbay_node.chat.store import ChatStore from meshbay_node.cli.dispatch import run, start from meshbay_node.config import DEFAULT_CONFIG_PATH, Config, load_config from meshbay_node.hub_client import HubClient, HubConfig from meshbay_node.indexer import DirectoryIndexer, GroupIndex, IndexCache from meshbay_node.indexer.enrich import Enricher from meshbay_node.indexer.enrich_audio import AudioEnricher from meshbay_node.indexer.enrich_photo import PhotoEnricher from meshbay_node.keystore import load_or_create_keystore from meshbay_node.media_cache import MediaCache from meshbay_node.musicbrainz import MusicBrainzClient from meshbay_node.platform import chmod_private from meshbay_node.roots import RootError, RootSet, entry_abs_path, off_disk from meshbay_node.roster import Roster from meshbay_node.tmdb import TmdbClient from meshbay_node.transport import ( QUIC_AVAILABLE, WEBRTC_AVAILABLE, Denylist, ) from meshbay_node.transport.wire import index_delta_message, index_sync_message if QUIC_AVAILABLE: from meshbay_node.transport import QuicChunkServer if WEBRTC_AVAILABLE: from meshbay_node.transport import WebRTCTransport log = logging.getLogger(__name__) def _under_any_directory(path: str, directories: list[str]) -> bool: """ Whether an entry's folder is one of an app's directories, or inside one. Mirrors `underAnyDirectory` in the SPA's app modules. One helper for every app since they all take a list: Videos and Music used to take a single folder and had a function each saying the same thing, which is how the two came to differ in what they did with a trailing slash. """ path = path or "" return any(path == d or path.startswith(d + "/") for d in directories) def _owning_directory(path: str, directories: list[str]) -> str | None: """ Which of an app's directories an entry belongs to — the deepest match. Deepest, because directories may nest: with both `Media` and `Media/Albums` configured, a file under the second belongs to the second. Taking the first match instead would measure it against a boundary one level too shallow, which for Music is the difference between reading a folder as an artist and reading it as a release. """ path = path or "" best: str | None = None for d in directories: if path == d or path.startswith(d + "/"): if best is None or len(d) > len(best): best = d return best # ── Hub WS sender bridge ───────────────────────────────────────────────────── class _WsSender: """Thin bridge so WebRTC context can call hub_ws.send() for chat_notify.""" def __init__(self, hub_client: HubClient): self._hub = hub_client async def send(self, data: str) -> None: await self._hub.send_ws(data) # ── Daemon ──────────────────────────────────────────────────────────────────── def _root_shape(roots) -> set[tuple]: """What has to match for a group's roots to count as unchanged on reload.""" return {(r.name, str(r.path), r.writable, r.removable) for r in roots} class NodeDaemon: def __init__(self, config: Config, config_path: Path = DEFAULT_CONFIG_PATH): self._config = config self._config_path = config_path self._state: dict = { "status": "starting", "hub_url": config.hub.url, "username": config.hub.username, "groups": [g.name for g in config.groups], "quic_port": config.node.quic_port, "endpoint_hint": None, "indexes": {}, "indexers": {}, } self._quic_server = None self._webrtc = None # Persisted so a restart does not silently un-revoke everyone (H4) self._denylist = ( Denylist(path=config.data_dir / "denylist.json") if Denylist else None) self._chat_stores: dict[str, ChatStore] = {} # One instance, shared by every group's DirectoryIndexer — see # indexer/cache.py's docstring for why this stopped being per-group. self._index_cache: IndexCache | None = None # Coalesces a burst of index changes (one per debounced watchdog # event) into a single broadcast — see _on_index_change. 0.5s is # short enough nobody notices the wait, long enough that dropping a # few hundred files into a watched folder produces one push instead # of one per file. self._broadcast_coalesce_secs = 0.5 self._pending_broadcasts: dict[str, asyncio.TimerHandle] = {} # group_id -> (version, {id: entry}) as of the last thing actually # broadcast — the comparison point for the next delta. self._last_broadcast_snapshot: dict[str, tuple] = {} self._audit_store: AuditStore | None = None self._bundle_store: BundleStore | None = None self._media_cache: MediaCache | None = None self._enricher: Enricher | None = None self._tmdb_client: TmdbClient | None = None self._audio_enricher: AudioEnricher | None = None self._musicbrainz_client: MusicBrainzClient | None = None self._photo_enricher: PhotoEnricher | None = None # A file id attempted at most once per daemon run, success or # failure — a persistently unprobeable file (corrupt, still being # written) does not get re-queued on every coalesced broadcast. A # restart retries everything, matching the "disposable, rebuildable" # stance the rest of this cache takes (docs/MESHBAY_DESIGN.md §6.5). # Shared across the video and audio enrichment paths — content- # addressed ids never collide between the two. Keyed by # (group_id, entry.id), not entry.id alone: the id is a content # hash, so the same physical file shared into two different groups # (found live — overlapping test libraries across several demo # groups) produces the same id in both. A bare-id set marked the # second group's copy "already attempted" the moment the first # group's enrichment ran, even though nothing had ever populated # *that* group's own index — every file in the second group stayed # at duration 0 with no artist/album, permanently, since nothing # ever revisits an id already in this set. self._enriched_attempted: set[tuple[str, str]] = set() self._roster: Roster | None = None self._indexers: list[DirectoryIndexer] = [] self._tasks: list[asyncio.Task] = [] self._hub: HubClient | None = None self._reload_lock = asyncio.Lock() async def run(self) -> None: log.info("MeshBay Node starting up") # 1. Keystore keys = load_or_create_keystore( path=self._config.keystore.path, unlock_file=self._config.keystore.unlock_file, ) log.info("Keys loaded: %s", keys.pk_ed25519_b64[:16]) # 2. Start the local control API early (so the operator can read the # node key before hub login). It is JSON-only, loopback-only, and both # the CLI and the desktop client's Node page are its clients. self._state["pk_node_ed25519"] = keys.pk_ed25519_b64 self._state["config"] = self._config # Where it came from, so `group add` appends to the file this process # actually read rather than guessing at the default. self._state["config_path"] = str(self._config_path) # Per-run token for the control API (11.5.3). Not a password: it keeps # other local processes and rebound browser pages out of an API that can # re-initialise group keys. ui_token = base64.urlsafe_b64encode(os.urandom(18)).decode().rstrip("=") self._state["ui_token"] = ui_token # Persisted so the CLI and the desktop client can read it — nobody # should ever copy a token out of a log or a terminal. self._config.data_dir.mkdir(parents=True, exist_ok=True) self._ui_token_file = self._config.data_dir / "ui-token" self._ui_token_file.write_text(ui_token, encoding="utf-8", newline="\n") chmod_private(self._ui_token_file) from meshbay_node.ui import create_ui_app ui_app = create_ui_app(self._state) ui_cfg = uvicorn.Config( ui_app, host="127.0.0.1", port=self._config.node.ui_port, log_level="warning", ) ui_server = uvicorn.Server(ui_cfg) self._tasks.append(asyncio.create_task(ui_server.serve())) log.info("Control API on 127.0.0.1:%d", self._config.node.ui_port) self._tasks.append(asyncio.create_task(self._reap_partial_uploads())) # 3. Hub connection (Ed25519 auth — retries until node key is linked) hub_cfg = HubConfig( hub_url=self._config.hub.url, username=self._config.hub.username, ) async with HubClient(hub_cfg, keys) as hub: self._hub = hub session = await self._login_with_retry(hub) self._state["endpoint_hint"] = session.node_id try: session.email = await hub.fetch_owner_email() except Exception as e: log.warning("Could not fetch owner email: %s", e) # 4. Bundle store (P2P GEK bundles) data_dir = self._config.data_dir data_dir.mkdir(parents=True, exist_ok=True) self._bundle_store = BundleStore(db_path=data_dir / "bundles.db") await self._bundle_store.open() log.info("Bundle store opened: %s", data_dir / "bundles.db") # 4a. Path->hash cache, node-wide — opened once, shared by every # group's DirectoryIndexer below (indexer/cache.py). self._index_cache = IndexCache(db_path=data_dir / "index_cache.db") await self._index_cache.open() self._state["index_cache"] = self._index_cache log.info("Index cache opened: %s", data_dir / "index_cache.db") # 4b. Roster — who this node recognises and which keys are theirs. # Node authority is established here, locally, and never learned from # the hub: a hub that could name the operator's key could install # itself as node administrator. self._roster = Roster(db_path=data_dir / "roster.db") await self._roster.open() await self._roster.purge_expired() # Apply any roster overrides to node config (panel-edited values # take precedence over node.toml defaults). from meshbay_node.config import node_settings_defaults nd = self._config.node effective = await self._roster.node_settings( node_settings_defaults(nd)) for k, v in effective.items(): setattr(nd, k, v) # X25519 key material for GEK unwrapping from cryptography.hazmat.primitives import serialization sk_x_raw = keys.sk_x25519.private_bytes( serialization.Encoding.Raw, serialization.PrivateFormat.Raw, serialization.NoEncryption()) pk_x_raw = base64.b64decode(keys.pk_x25519_b64) # 4. Build per-group contexts groups_ctx: dict[str, dict] = {} for group_cfg in self._config.groups: if not group_cfg.id or not group_cfg.roots: log.warning("Group %r has no id or no shared directory — " "skipping", group_cfg.name) continue try: roots = await self._build_roots(group_cfg) except RootError as e: # Configuration the operator has to fix; guessing would put # a member's file on the wrong disk or index one twice. log.error("Group %r: %s — skipping", group_cfg.name, e) continue await off_disk(roots, roots.refresh_availability) if not any(r.available for r in roots): # Not skipped for being empty: a group whose only drive is # unplugged still exists, and its index is frozen rather # than lost. But there is nothing to serve until it returns. log.warning( "Group %r: none of its %d root(s) are readable right now " "(%s) — serving nothing until one returns", group_cfg.name, len(roots), ", ".join(str(r.path) for r in roots)) gek = None gek = await self._load_gek( group_cfg.id, session.user_id, sk_x_raw, pk_x_raw) if gek: log.info("GEK loaded for group %s", group_cfg.id[:8]) else: log.info("No GEK yet for group %s — will accept first setup", group_cfg.name) # Read once at load, like enabled_apps below — # kept current in place afterwards by set_scan_settings # (ops.py), which updates both this indexer object directly # and roster.db, so a restart picks up the same values. scan_settings = ( await self._roster.scan_settings(group_cfg.id) if self._roster else { "reconcile_interval_secs": DirectoryIndexer.DEFAULT_RECONCILE_SECS, "debounce_secs": DirectoryIndexer.DEFAULT_DEBOUNCE_SECS, }) indexer = DirectoryIndexer( roots=roots, group_id=group_cfg.id, sk_node=keys.sk_ed25519, gek=gek, on_change=self._on_index_change, on_root_ejected=self._eject_persister(group_cfg.id), cache=self._index_cache, reconcile_secs=scan_settings["reconcile_interval_secs"], debounce_secs=scan_settings["debounce_secs"], ) await indexer.start(defer_scan=True) self._indexers.append(indexer) self._state["indexes"][group_cfg.id] = indexer.index self._state["indexers"][group_cfg.id] = indexer log.info("Group %s configured: %s (scan deferred)", group_cfg.name, ", ".join(f"{r.name}={r.path}" for r in roots)) groups_ctx[group_cfg.id] = { "gek": gek, "roots": roots, "index": indexer.index, # Live reference, mutated in place by the indexer itself # (see IndexProgress in indexer.py) — read, never copied, # by the handshake ack and the periodic progress pusher. "progress": indexer.progress, # Bound method, called when a peer completes the # handshake — resets reconcile's backoff (indexer.py # _reconcile_loop) so the backstop is prompt again now # that someone is actually looking. "note_activity": indexer.note_activity, # Bound method, called when an upload finishes. The entry # it belongs to does not exist yet (see # webrtc/upload_handlers.py _register_uploader), so the indexer keeps # the record and stamps the entry when it creates it. "record_upload": indexer.record_upload, # Shown to the operator in Settings, and kept current in # place by set_scan_settings (ops.py) — same reasoning as # enabled_apps below. "reconcile_interval_secs": scan_settings["reconcile_interval_secs"], "debounce_secs": scan_settings["debounce_secs"], "visibility": group_cfg.visibility, # Admission policy comes from node.toml, never from the hub: # a hub that could declare a group open would be handed its key. "join_policy": group_cfg.join_policy, # Read once at load, kept current in place by the signed # operation that changes it — the upload handler is # synchronous and a database round trip per chunk would be # absurd. (Whether a member may upload is not here any # more: it is `writable` on the root being written to, # which the RootSet above already carries.) "enabled_apps": await self._roster.enabled_apps( group_cfg.id) if self._roster else list(Roster.DEFAULT_APPS), # How many transfers one member may run at once here. Empty # means the operator has not said, and the node's default # applies — never "unlimited" (transfers.member_cap). "transfer_limits": ( await self._roster.transfer_limits(group_cfg.id) if self._roster else {}), # Which folder(s) inside the shared roots each app works # over. One shape for every app (roster.py's # app_directories) — an empty list means nothing has been # chosen, which every app reads as "show nothing yet", # never "the whole group index". **(await self._app_directories_ctx(group_cfg.id)), # Whether the node unfurls links members post here. "chat_link_preview": await self._roster.chat_link_preview( group_cfg.id) if self._roster else True, # Whether members' cross-group Search lists this group. "search_listed": await self._roster.search_listed( group_cfg.id) if self._roster else True, # Which chat epoch key is current. Opened here if the # group has none, because chat is always encrypted (MNP # 2.0) and a group with no epoch is a group nobody can # speak in — the node cannot wait for an operator to notice. "chat_epoch": await self._ensure_chat_epoch(group_cfg.id), # Whether TMDB lookups run for this group at all — # per-group (2026-08-24, used to be node-wide), same # "read once, kept current in place by the signed op" # shape as the app directories above. "tmdb_enabled": await self._roster.tmdb_enabled( group_cfg.id) if self._roster else True, # Music app equivalent of tmdb_enabled — per-group from # the start (docs/MESHBAY_DESIGN.md §9.8). "musicbrainz_enabled": await self._roster.musicbrainz_enabled( group_cfg.id) if self._roster else True, } if not groups_ctx: log.warning("No groups configured yet — the control API and hub " "connection stay up; attach a group to go live") # 5. Chat stores (one SQLite DB per group) for gid in groups_ctx: chat_db = data_dir / gid[:16] / "chat.db" store = ChatStore(db_path=chat_db) await store.open() self._chat_stores[gid] = store groups_ctx[gid]["chat_store"] = store log.info("Chat stores opened: %d groups", len(self._chat_stores)) # 6. Audit store (legal compliance — IP + action logging) audit_db = data_dir / "audit.db" self._audit_store = AuditStore(db_path=audit_db) await self._audit_store.open() log.info("Audit store opened: %s", audit_db) self._tasks.append(asyncio.create_task(self._purge_audit_log())) # 6b. Media cache (Videos app — TMDB metadata + thumbnails). # Node-wide like audit.db, not per-group: a thumbnail is the same # bytes regardless of which group happens to share the file # (docs/MESHBAY_DESIGN.md §6.5, §9.7). media_cache_db = data_dir / "media_cache.db" self._media_cache = MediaCache(db_path=media_cache_db) await self._media_cache.open() self._enricher = Enricher(self._media_cache) self._tmdb_client = TmdbClient(roster=self._roster) # Token/language only — read once at load, kept current in place # by ops.set_tmdb_config (the signed op), exposed to every # group's handshake ack via `daemon_state` (already wired to # self._webrtc._ctx below) since these stay node-wide, one # shared credential/cache. Whether TMDB is used at all is now # per-group instead — see each group's own "tmdb_enabled" in # groups_ctx above. tmdb_token, tmdb_language = await self._roster.tmdb_config() self._state["tmdb_token_customized"] = bool(tmdb_token) self._state["tmdb_language"] = tmdb_language or "" # 6c. Music app (docs/MESHBAY_DESIGN.md §9.8) — same # media_cache.db, its own enricher (mutagen, not ffmpeg) and its # own MusicBrainz client. # The User-Agent contact is the owner's hub email, resolved at # login — no roster setting or env var needed any more. self._audio_enricher = AudioEnricher(self._media_cache) self._musicbrainz_client = MusicBrainzClient(owner_email=session.email) self._state["musicbrainz_contact_configured"] = bool(session.email) # 6d. Photos app (docs/MESHBAY_DESIGN.md §9.9) — same # media_cache.db, its own enricher (Pillow, not ffmpeg/mutagen). # No credential, no # third-party client to construct: EXIF is read locally. self._photo_enricher = PhotoEnricher(self._media_cache) self._state["media_cache"] = self._media_cache log.info("Media cache opened: %s", media_cache_db) # 5. Denylist denylist = self._denylist # 6. WebRTC transport (browser clients) # # The operator's answer about the GPU, before the first stream asks # the question. The probe itself is lazy — it costs a test encode, # and a node that never serves a video should never pay it. from meshbay_node import hwaccel hwaccel.set_enabled(self._config.node.hardware_video_encode) from meshbay_node.transport.ice_filter import install as install_ice_filter install_ice_filter( self._config.node.ice_interfaces or None, ) # aiortc keeps only the FIRST entry of RTCConfiguration.iceServers, so # the multi-STUN fallback only exists on the node if aioice itself # fans out — see transport/stun_multi. from meshbay_node.transport.stun_multi import install as install_stun_multi install_stun_multi(self._config.node.stun_servers or None) first = next(iter(groups_ctx.values()), None) if WEBRTC_AVAILABLE: self._webrtc = WebRTCTransport( sk_node=keys.sk_ed25519, hub_pk_pem=session.hub_pk_pem, gek=first["gek"] if first else None, roots=first["roots"] if first else None, index=first["index"] if first else None, groups=groups_ctx, denylist=denylist, max_concurrent_streams=self._config.node.max_concurrent_streams, max_concurrent_downloads=self._config.node.max_concurrent_downloads, max_concurrent_uploads=self._config.node.max_concurrent_uploads, max_upload_gb=self._config.node.max_upload_gb, transcode_incompatible_video=self._config.node.transcode_incompatible_video, stun_servers=self._config.node.stun_servers or None, ) # No global chat_store here: each group's store lives in # groups_ctx[gid]["chat_store"] and is resolved per session via # _group_ctx(). Assigning the first group's store transport-wide # sent every group's chat to one database and served it back to # members of every other group (finding H1). self._webrtc._ctx["groups"] = groups_ctx self._webrtc._ctx["hub_ws"] = _WsSender(hub) self._webrtc._ctx["node_user_id"] = session.user_id self._webrtc._ctx["audit_store"] = self._audit_store self._webrtc._ctx["bundle_store"] = self._bundle_store self._webrtc._ctx["media_cache"] = self._media_cache self._webrtc._ctx["tmdb_client"] = self._tmdb_client self._webrtc._ctx["musicbrainz_client"] = self._musicbrainz_client self._webrtc._ctx["sk_x25519_raw"] = sk_x_raw self._webrtc._ctx["pk_x25519_raw"] = pk_x_raw self._webrtc._ctx["pk_x25519_b64"] = keys.pk_x25519_b64 self._webrtc._ctx["roster"] = self._roster # The MNP adapter calls the same operations as the loopback API # (meshbay_node.ops), and those take the daemon's state. Handing # the transport a second set of lookups is how two paths to one # operation start disagreeing — the shape of C1 and C6. self._webrtc._ctx["daemon_state"] = self._state self._webrtc._ctx["invite_ttl"] = ( self._config.node.invite_ttl_hours * 3600) self._webrtc._ctx["device_request_ttl"] = ( self._config.node.device_request_ttl_minutes * 60) paired = await self._roster.has_operator() if self._roster else False self._webrtc._ctx["has_admin_authority"] = paired if paired: log.info("Node authority: paired operator") else: log.warning( "No operator paired — invites and file deletion are " "refused. Run: meshbay-node operator pair") log.info("WebRTC transport ready") else: log.warning("WebRTC not available (aiortc not installed)") # 7. QUIC chunk server (LAN / port-forwarded / hub-less direct access) # # Off unless `[node] quic_enabled = true`: no shipping client speaks # QUIC (browser and desktop use WebRTC; the `group://` sidecar is # unbuilt), so starting it by default only exposes a UDP port. if QUIC_AVAILABLE and self._config.node.quic_enabled: self._quic_server = QuicChunkServer( sk_node=keys.sk_ed25519, hub_pk_pem=session.hub_pk_pem, gek=first["gek"] if first else None, roots=first["roots"] if first else None, index=first["index"] if first else None, host="::", port=self._config.node.quic_port, groups=groups_ctx, denylist=denylist, ) self._quic_server._ctx["groups"] = groups_ctx await self._quic_server.start() log.info("QUIC server on port %d (%d groups)", self._config.node.quic_port, len(groups_ctx)) elif QUIC_AVAILABLE: log.info("QUIC server disabled ([node] quic_enabled = false)") # 8. Hub WebSocket (signaling + revocations + WebRTC offers) async def on_webrtc_offer(sdp, peer_id, ice_candidates): if not self._webrtc: return None try: answer_sdp, answer_ice = await self._webrtc.handle_offer( sdp, peer_id) log.info("WebRTC answer for peer=%s (%d peers)", peer_id, self._webrtc.active_peers) return (answer_sdp, answer_ice) except Exception as e: log.error("WebRTC offer failed: %s", e) return None async def on_incoming(peer_ip, peer_port): if self._quic_server: self._quic_server.punch_nat(peer_ip, peer_port) def on_revocation(token): if denylist and token: import jwt as _jwt try: payload = _jwt.decode( token, session.hub_pk_pem, algorithms=["EdDSA"], options={"verify_exp": False}) target = payload.get("target") tid = payload.get("target_id", "") if target == "user": denylist.deny_user(tid) elif target == "group": # H4: previously dropped on the floor, so "suspend a # group" was a hub-only gesture that no node enforced. denylist.deny_group(tid) self._drop_group_sessions(tid) elif target == "jti": denylist.deny_jti(tid) else: log.warning("Unknown revocation target: %r", target) except Exception as e: log.warning("Invalid revocation token: %s", e) ws_task = asyncio.create_task(hub.maintain_ws( on_incoming=on_incoming, on_revocation=on_revocation, on_webrtc_offer=on_webrtc_offer, group_ids=lambda: list((self._state.get("groups_ctx") or {}).keys()), )) self._tasks.append(ws_task) log.info("Hub WS task started") # 9. (removed in Phase 11.5) The per-group HTTP file API used to start here. # It served the Mesh Group Index and raw plaintext files on 0.0.0.0 with no # authentication, for private groups too — finding C1. Every client path now # goes through the MNP handshake (JWT + group claim + GEK proof). # 10. Update control API state (already running from step 2) self._state["groups_ctx"] = groups_ctx self._state["audit_store"] = self._audit_store self._state["bundle_store"] = self._bundle_store self._state["roster"] = self._roster self._state["node_user_id"] = session.user_id # The node's own Ed25519 key. Needed by `encrypt_chat_history`, # which seals migrated messages under a synthetic device of the # node's rather than pretending to hold a member's signing key. self._state["sk_node"] = keys.sk_ed25519 self._state["webrtc"] = self._webrtc self._state["quic_server"] = self._quic_server self._state["hub"] = hub self._state["reload_fn"] = self._reload_config # Keyed by app, so `ops.set_app_directories` finds the right # sweep without knowing which apps exist — an app with nothing to # enrich simply has no entry. self._state["enrich_app_dirs_fns"] = { "video": self._enrich_video_root_now, "music": self._enrich_audio_root_now, "photo": self._enrich_photo_roots_now, } # Rotating a key has to reach every transport holding a copy of it, # and clearing the denylist has to reach the one the handshake # consults — so both are published rather than reachable only # through the object that happens to own them. self._state["denylist"] = self._denylist self._state["pk_x25519_raw"] = pk_x_raw self._state["sk_x25519_raw"] = sk_x_raw self._state["sk_ed25519"] = keys.sk_ed25519 self._state["status"] = "running" log.info("Node ready — %d groups, WebRTC=%s, QUIC=%s", len(groups_ctx), "yes" if self._webrtc else "no", "yes" if self._quic_server else "no") # 11. Background initial scan — files appear progressively. async def _bg_scan(indexer, name, gctx): await indexer.initial_scan() log.info("Background scan complete for %s: %d files", name, indexer.index.count) # Swarm registration for public groups (after files are known). if gctx.get("visibility") == "public": endpoint = f"webrtc:{self._config.node.quic_port}" hashes = [e.id for e in gctx["index"].entries] if hashes: await self._register_swarm(hashes, endpoint) # initial_scan() itself never calls on_change (it predates # the concept — every existing caller only cared about the # scan finishing, not about notifying anyone) — but Videos # app enrichment (duration/thumb_hash/display_title/...) # hangs entirely off that callback (_broadcast_index_change). # Without this, every file already on disk at startup — the # common case, an existing library — would never get # enriched at all; only a file added later, while the node # is already running, would trigger it via the watchdog. await self._on_index_change(indexer) for idx, group_cfg in zip(self._indexers, self._config.groups): gctx = groups_ctx.get(group_cfg.id) if gctx: self._tasks.append(asyncio.create_task( _bg_scan(idx, group_cfg.name, gctx))) self._tasks.append(asyncio.create_task( self._progress_pusher(idx))) # 12. Wait for shutdown stop_event = asyncio.Event() loop = asyncio.get_event_loop() console_shutdown_done = None if sys.platform == "win32": # SIGBREAK: CTRL_BREAK_EVENT, how platform.autostart_end() asks # a per-user-mode daemon to stop gracefully instead of only # ever taskkill /F. for sig in (signal.SIGINT, signal.SIGTERM, signal.SIGBREAK): signal.signal(sig, lambda *_: stop_event.set()) # CTRL_CLOSE/LOGOFF/SHUTDOWN reach no Python signal at all -- # see platform.install_console_close_handler for why this is # a separate mechanism rather than another signal.signal() line. from meshbay_node.platform import install_console_close_handler console_shutdown_done = install_console_close_handler(loop, stop_event) else: for sig in (signal.SIGINT, signal.SIGTERM): loop.add_signal_handler(sig, stop_event.set) # Milestone 14.8: re-read node.toml without dropping connections. try: loop.add_signal_handler( signal.SIGHUP, lambda: spawn(self._reload_config())) except (NotImplementedError, AttributeError): pass # no SIGHUP on Windows; `reload` says so there await stop_event.wait() await self._shutdown() if console_shutdown_done is not None: # Releases the console-control handler's blocking wait (see # platform.install_console_close_handler) so it can return and # let Windows actually end the process for CTRL_CLOSE/LOGOFF/ # SHUTDOWN, now that cleanup is genuinely done rather than just # started. console_shutdown_done.set() async def _reload_config(self) -> None: """ Re-read node.toml and reconcile groups. Handles root changes on existing groups, hot-loads new groups, and tears down removed groups. Existing connections are untouched: a member watching a film keeps watching it. Serialised by _reload_lock: fire-and-forget reloads from config-mutating endpoints can overlap with the wizard's explicit /api/reload call, and two concurrent hot-loads of the same group corrupt the runtime state. The reload belongs to the node, never to whoever asked for it. A root added from a browser reaches here through the operator's WebRTC session, whose tasks are all cancelled when that session closes — and on 2026-09-14 one closed 47 s into the scan of a 900 GB root. The reload died without a line in the log, the new root was in node.toml and in the indexer but never in the group's context, and nothing ever tried again: the lock was free, and nobody was waiting on it. So the work runs in a task of its own, and a caller that goes away only stops waiting. """ await asyncio.shield(spawn(self._reload_config_locked(), what="config reload")) async def _reload_config_locked(self) -> None: try: async with self._reload_lock: await self._reload_config_inner() except asyncio.CancelledError: log.warning("Config reload cancelled before it finished — the node may " "be serving part of the previous configuration until the " "next reload") raise async def _reload_config_inner(self) -> None: log.info("Reloading config from %s", self._config_path) try: fresh = load_config(self._config_path) except Exception as e: log.error("Reload failed, keeping the running config: %s", e) return groups_ctx = self._state.get("groups_ctx", {}) hosted = set(groups_ctx) incoming = {g.id for g in fresh.groups if g.id} # ── Root changes on existing groups ────────────────────────────── changed = 0 for group_cfg in fresh.groups: ctx = groups_ctx.get(group_cfg.id) if not ctx: continue try: roots = await self._build_roots(group_cfg) except RootError as e: log.error("Group %r: %s — keeping the roots already loaded", group_cfg.name, e) continue # `writable` and `removable` are in the comparison because an # operator editing node.toml by hand and reloading is a supported # way to change them, and a set compared on name and path alone # reports "nothing changed" for exactly that edit. if _root_shape(ctx["roots"]) == _root_shape(roots): continue await off_disk(roots, roots.refresh_availability) indexer = next((i for i in self._indexers if i.group_id == group_cfg.id), None) if indexer is None: continue log.info("Group %r roots changed: %s", group_cfg.name, ", ".join(f"{r.name}={r.path}" for r in roots)) # The new set is what the node serves from this moment, and the # scan of an added root is not waited for. Awaiting it here held # `_reload_lock` and the old set for as long as the scan ran — # hours for a large drive — so every file request under the new # root found no root to resolve against, and any op answering with # the live table (a writable/removable toggle) showed the directory # gone from the operator's settings. ctx["roots"] = roots await indexer.retarget(roots, wait=False) changed += 1 # ── Hot-load new groups ────────────────────────────────────────── added_names = [] sk_ed = self._state.get("sk_ed25519") sk_x_raw = self._state.get("sk_x25519_raw") pk_x_raw = self._state.get("pk_x25519_raw") node_user_id = self._state.get("node_user_id") data_dir = fresh.data_dir for group_cfg in fresh.groups: if group_cfg.id in hosted: continue if not group_cfg.id or not group_cfg.roots: log.warning("New group %r has no id or roots — skipping", group_cfg.name) continue if not sk_ed: log.warning("Cannot hot-load %r — signing key not available", group_cfg.name) continue try: roots = await self._build_roots(group_cfg) except RootError as e: log.error("New group %r: %s — skipping", group_cfg.name, e) continue await off_disk(roots, roots.refresh_availability) gek = None if sk_x_raw and pk_x_raw: gek = await self._load_gek( group_cfg.id, node_user_id, sk_x_raw, pk_x_raw) if gek: log.info("GEK loaded for new group %s", group_cfg.id[:8]) scan_settings = ( await self._roster.scan_settings(group_cfg.id) if self._roster else { "reconcile_interval_secs": DirectoryIndexer.DEFAULT_RECONCILE_SECS, "debounce_secs": DirectoryIndexer.DEFAULT_DEBOUNCE_SECS, }) indexer = DirectoryIndexer( roots=roots, group_id=group_cfg.id, sk_node=sk_ed, gek=gek, on_change=self._on_index_change, on_root_ejected=self._eject_persister(group_cfg.id), cache=self._index_cache, reconcile_secs=scan_settings["reconcile_interval_secs"], debounce_secs=scan_settings["debounce_secs"], ) # Registered *before* start() runs its (blocking, possibly very # long — see the StarWars benchmark) initial scan, specifically # so /api/groups/{id}/index-status can see indexer.progress # while a brand-new group is still scanning — this is the one # group state that must stay visible during the very window the # group is not yet authorized for member connections (below). self._indexers.append(indexer) self._state["indexes"][group_cfg.id] = indexer.index self._state["indexers"][group_cfg.id] = indexer await indexer.start() data_dir.mkdir(parents=True, exist_ok=True) chat_db = data_dir / group_cfg.id[:16] / "chat.db" store = ChatStore(db_path=chat_db) await store.open() self._chat_stores[group_cfg.id] = store new_ctx = { "gek": gek, "roots": roots, "index": indexer.index, "progress": indexer.progress, "note_activity": indexer.note_activity, "record_upload": indexer.record_upload, "reconcile_interval_secs": scan_settings["reconcile_interval_secs"], "debounce_secs": scan_settings["debounce_secs"], "visibility": group_cfg.visibility, "join_policy": group_cfg.join_policy, "enabled_apps": ( await self._roster.enabled_apps(group_cfg.id) if self._roster else list(Roster.DEFAULT_APPS)), "transfer_limits": ( await self._roster.transfer_limits(group_cfg.id) if self._roster else {}), **(await self._app_directories_ctx(group_cfg.id)), "chat_link_preview": ( await self._roster.chat_link_preview(group_cfg.id) if self._roster else True), "search_listed": ( await self._roster.search_listed(group_cfg.id) if self._roster else True), "chat_epoch": await self._ensure_chat_epoch(group_cfg.id), "tmdb_enabled": ( await self._roster.tmdb_enabled(group_cfg.id) if self._roster else True), "musicbrainz_enabled": ( await self._roster.musicbrainz_enabled(group_cfg.id) if self._roster else True), "chat_store": store, } groups_ctx[group_cfg.id] = new_ctx if self._webrtc: self._webrtc._ctx["groups"][group_cfg.id] = new_ctx log.info("Hot-loaded group %s (%s, %d roots)", group_cfg.name, group_cfg.id[:8], len(roots)) added_names.append(group_cfg.name) # The initial scan above already ran to completion (indexer.start() # is not deferred here), so this only matters for whatever scans # this group as time goes on — a root added later, reconcile # picking one back up. self._tasks.append(asyncio.create_task(self._progress_pusher(indexer))) # ── Tear down removed groups ───────────────────────────────────── removed_names = [] for gid in hosted - incoming: indexer = next((i for i in self._indexers if i.group_id == gid), None) if indexer: try: await indexer.stop() except Exception: pass self._indexers.remove(indexer) store = self._chat_stores.pop(gid, None) if store: try: await store.close() except Exception: pass # No per-group index cache to close here (2026-08-25): the # (path, size, mtime) -> hash cache is now one shared instance, # open for the life of the daemon, since another group may still # reference the same physical folder — see indexer/cache.py. pending = self._pending_broadcasts.pop(gid, None) if pending: pending.cancel() self._last_broadcast_snapshot.pop(gid, None) self._state["indexes"].pop(gid, None) self._state["indexers"].pop(gid, None) old_name = gid[:8] for g_cfg in self._config.groups: if g_cfg.id == gid: old_name = g_cfg.name break groups_ctx.pop(gid, None) if self._webrtc and self._webrtc._ctx.get("groups") is not groups_ctx: self._webrtc._ctx["groups"].pop(gid, None) log.info("Unloaded group %s (%s)", old_name, gid[:8]) removed_names.append(old_name) self._config = fresh self._state["config"] = fresh self._state["groups"] = [g.name for g in fresh.groups] log.info("Reload complete — %d re-rooted, %d added, %d removed", changed, len(added_names), len(removed_names)) if (added_names or removed_names) and self._hub: gids = list((self._state.get("groups_ctx") or {}).keys()) await self._hub.update_ws_groups(gids) async def _login_with_retry(self, hub: HubClient): """Login to hub, retrying if the node key hasn't been linked yet.""" import httpx as _httpx while True: try: return await hub.startup(endpoint_hint=None) except _httpx.HTTPStatusError as e: body = e.response.text if hasattr(e.response, 'text') else '' # Any 401 here needs a human at a browser, and the operator needs # this daemon alive to read its public key (via `meshbay-node # status` or the desktop client, both of which query the control # API). Exiting would strand them — which is exactly what # happened when a node was started before its owner had # registered. if e.response.status_code == 401: if "No node key" in body: self._state["status"] = "waiting_for_node_key" log.warning( "Node key not linked. Get it from `meshbay-node " "status` and paste it in Settings > Link Node on %s " "(the desktop client links it automatically). " "Retrying in 5s...", self._config.hub.url, ) else: self._state["status"] = "waiting_for_account" log.warning( "Hub rejected the node credentials for user %r. " "Register that account on %s first, then link this " "node's key. Retrying in 5s...", self._config.hub.username, self._config.hub.url, ) await asyncio.sleep(5) else: raise except Exception as e: log.warning("Hub login failed: %s — retrying in 10s", e) await asyncio.sleep(10) async def _load_gek( self, group_id: str, node_user_id: str, sk_x_raw: bytes, pk_x_raw: bytes, ) -> bytes | None: """Load GEK from local bundle store (node-only, hub never touches crypto).""" from meshbay_common.crypto import unwrap_gek_aes if not self._bundle_store: return None # Try node-specific bundle first (stored by init_gek for daemon reload), # then fall back to operator's user bundle (legacy / pre-dual-key) for user_key in [f"_node_{node_user_id}", node_user_id]: bundle = await self._bundle_store.fetch(group_id, user_key) if not bundle: continue try: gek = unwrap_gek_aes(bundle, sk_x_raw, pk_x_raw) log.info("GEK loaded from local bundle store for group %s (key=%s)", group_id[:8], user_key[:16]) return gek except Exception as e: log.debug("Failed to unwrap GEK bundle (key=%s): %s", user_key[:16], e) log.warning("No unwrappable GEK bundle found for group %s", group_id[:8]) return None async def _ensure_chat_epoch(self, group_id: str) -> int: """ The group's current chat epoch, opening the first one if it has none. Chat is always encrypted, so a group with no epoch key is a group in which nobody can say anything. Attaching one is the node's job and happens here rather than on the first message: a failure at start-up is in the log the operator is already reading, and a failure on someone's first message is a chat that mysteriously refuses them. Never fatal. A group whose epoch cannot be opened keeps every other function — files, video, the index — and only its chat is unusable, which is strictly better than refusing to host the group at all. """ if not self._bundle_store: return 0 try: epoch = await self._bundle_store.latest_chat_epoch(group_id) if epoch: return epoch from meshbay_node import ops return (await ops.open_chat_epoch(self._state, group_id))["epoch"] except Exception as e: log.error("chat: no epoch key for group %s (%s) — chat is " "unusable in this group until this is fixed", group_id[:8], e) return 0 async def _reap_partial_uploads(self, interval: float = 3600.0, first_delay: float = 60.0) -> None: """ Delete `.part` files that no upload will ever finish. An upload interrupted for good leaves its partial file behind, and nothing else ever looks at it: `.part` is not an index entry, so it is invisible to every group member and to the operator's own file list. One abandoned film is a gigabyte of their disk, kept for ever. Two conditions, both required, and `uploads.orphaned_parts` is where they are stated and tested. What this adds is the walk and the deletion, and one rule of its own: it runs a minute after start rather than at once, so a client reconnecting to finish an upload that outlived a node restart is not raced by the janitor that would have deleted it — the age threshold makes that impossible in practice, and doing it anyway costs a minute. `interval` and `first_delay` are parameters so a test can drive this without waiting an hour. """ await asyncio.sleep(first_delay) while True: try: self._reap_once() except Exception as exc: # never let the janitor kill the node log.warning("Reaping partial uploads failed: %s", exc) await asyncio.sleep(interval) async def _purge_audit_log(self, interval: float = 86400.0) -> None: """ Delete audit entries past the retention period, at start then daily. `AuditStore.cleanup` leaves scheduling to its caller; until this loop nobody called it and a node kept every entry for ever. `interval` is a parameter so a test can drive this without waiting a day. """ while True: try: if self._audit_store: deleted = await self._audit_store.cleanup() if deleted: log.info("Purged %d audit entries older than %d days", deleted, AUDIT_RETENTION_DAYS) except Exception as exc: # never let the janitor kill the node log.warning("Purging the audit log failed: %s", exc) await asyncio.sleep(interval) def _reap_once(self, now: float | None = None) -> int: """One pass over every group. Returns how many files were deleted.""" groups = (self._webrtc._ctx.get("groups") or {}) if self._webrtc else {} when = time.time() if now is None else now deleted = 0 for gid, ctx in groups.items(): roots = ctx.get("roots") if roots is None: continue store = ctx.get("partial_uploads") live = store.live_paths() if store is not None else set() for path in uploads_mod.orphaned_parts( uploads_mod.find_parts(roots.roots), live, when): try: size = path.stat().st_size path.unlink() except OSError as exc: log.warning("Could not remove abandoned upload %s: %s", path.name, exc) continue deleted += 1 log.info("Removed abandoned upload %s (%d bytes, group %s)", path.name, size, gid[:8]) return deleted async def _progress_pusher(self, indexer: DirectoryIndexer, interval: float = 2.0) -> None: """ Watches indexer.progress and pushes a light INDEX_PROGRESS message to this group's connected peers — never the index itself, that stays _on_index_change's job. Runs for the node's whole lifetime: a scan can start from several places (initial scan, a root added later, reconcile picking a root back up), and this only needs to notice the flag, not why it changed. The final push at the False transition is what lets a presence dot reliably turn back off on an already-connected client — the handshake ack only covers the moment of connecting. `interval` is a parameter (not a bare constant) only so a test can drive this loop without waiting on the real 2s cadence. """ was_busy = False while True: await asyncio.sleep(interval) progress = indexer.progress # A root waiting for the scan lock is work the operator is owed a # sight of, even in the gap where nothing is walking yet. busy = progress.scanning or bool(progress.queued) if busy or was_busy: self._push_index_progress(indexer.group_id, progress) was_busy = busy def _push_index_progress(self, group_id: str, progress) -> None: if not self._webrtc: return # Deliberately NOT sealed, unlike index_sync/index_delta (decision D3). # Counters only — never a path, never a filename, see IndexProgress in # indexer.py — pushed every couple of seconds for the whole length of a # scan. Sealing it would buy an attacker's rough estimate of a library's # size and cost a key derivation and a decrypt per push. If a field that # names anything is ever added here, that trade is void and this message # joins the other two. `kind` is one of four fixed words; the root under # way is a position in the roots table the member already opened from # the sealed index, and the roots waiting are a count, never names. msg = { "type": MNP.INDEX_PROGRESS, "v": MNP_VERSION, "group_id": group_id, "scanning": progress.scanning, "scanned_bytes": progress.scanned_bytes, "total_bytes": progress.total_bytes, "files_done": progress.files_done, "files_total": progress.files_total, "kind": progress.kind, "root_pos": progress.root_pos, "queued": len(progress.queued), } pushed = 0 for session in list(self._webrtc._sessions.values()): if session._group_id == group_id: try: session._send(msg) pushed += 1 except Exception: pass if pushed: log.debug("Index progress pushed to %d peer(s) for group %s", pushed, group_id[:8]) async def _build_roots(self, group_cfg) -> RootSet: """ Build a group's RootSet from node.toml, with the ejected state restored. node.toml carries configuration (`writable`, `removable`); the roster carries the runtime answer to "is this drive ejected right now". They are merged here, in the one place every caller goes through, because a root that quietly comes back available across a restart is exactly the surprise unplug that eject exists to survive. """ specs = [asdict(r) for r in group_cfg.roots] if self._roster: ejected = await self._roster.ejected_roots(group_cfg.id) if ejected: for spec in specs: name = spec.get("name") or Path(spec.get("path", "")).name if fold(name) in ejected: spec["ejected"] = True return RootSet.build(specs) # Every application that keeps directories. This is the one list, and it # lives here because the daemon is what wires a group's context: `roster.py`, # `ops.py` and the rest must name no application at all — that is the # property the reference app exists to demonstrate # (`test_helloworld_proves_the_plugin_claim.py`). # # Not derived from `enabled_apps`: the context is read once at load, and an # application enabled later must not find its own setting missing. # # The handshake ack does **not** get a copy of this. It emits whatever # `_directories` the context holds, so the two cannot drift — a copy # lived in `webrtc_server.py` until 2026-09-10 and had already lost # `helloworld`, which made the app that proves a new one needs no # special-casing the single app whose directories never reached a client. APP_DIR_KEYS = ("video", "music", "photo", "chat", "helloworld") async def _app_directories_ctx(self, group_id: str) -> dict: """ Each app's configured directories, plus the second name an app is also published under where something reads one (`chat_directory`). Derived here rather than stored, so the two can never disagree. """ dirs = {} for app in self.APP_DIR_KEYS: dirs[f"{app}_directories"] = ( await self._roster.app_directories(group_id, app) if self._roster else []) aliases = {} for app in self.APP_DIR_KEYS: alias = Roster.ctx_alias(app, dirs[f"{app}_directories"]) if alias: aliases[alias[0]] = alias[1] return {**dirs, **aliases} def _eject_persister(self, group_id: str): """`on_root_ejected` bound to one group, for that group's indexer.""" async def persist(root_name: str, ejected: bool) -> None: if self._roster: await self._roster.set_root_ejected( group_id, root_name, ejected, set_by=self._state.get("node_user_id", "")) return persist async def _on_index_change(self, indexer: DirectoryIndexer) -> None: """ Called when a DirectoryIndexer detects file changes — once per debounced watchdog event, so dropping N files into a watched folder calls this N times in quick succession. Coalesces those into one broadcast (_broadcast_index_change) rather than one push per file: the timer is reset on every call and only fires once calls stop arriving for _broadcast_coalesce_secs. """ group_id = indexer.group_id loop = asyncio.get_event_loop() pending = self._pending_broadcasts.pop(group_id, None) if pending: pending.cancel() def fire() -> None: self._pending_broadcasts.pop(group_id, None) spawn(self._broadcast_index_change(indexer)) self._pending_broadcasts[group_id] = loop.call_later( self._broadcast_coalesce_secs, fire) async def _broadcast_index_change(self, indexer: DirectoryIndexer) -> None: """ The actual push, run once per coalesced burst. Sends a full INDEX_SYNC the first time a group is ever broadcast (no previous snapshot to diff against — the client's own first fetchIndex() call already covers that case) and an INDEX_DELTA every time after, computed against the last thing this method actually sent. """ group_id = indexer.group_id idx = indexer.index log.info("Index changed for group %s: %d files (v%d)", group_id[:8], idx.count, idx.version) prev = self._last_broadcast_snapshot.get(group_id) delta = None previous = None if prev is not None: prev_version, prev_entries = prev previous = GroupIndex._snapshot( idx.group_id, idx.sk_node, idx.gek, prev_version, prev_entries) delta = idx.diff(previous) self._last_broadcast_snapshot[group_id] = (idx.version, idx.entries_by_id()) # Videos app (docs/MESHBAY_DESIGN.md §6.5): schedule async technical # probe + title parse + thumbnail generation for every newly-seen # video entry under the group's configured video_root. Never blocks # this broadcast — enrichment fields arrive later as their own # INDEX_DELTA update (_on_enriched below). new_entries = delta.additions if delta is not None else list(idx.entries) # A root that was ejected and plugged back in, or that fell off and # re-mounted, has had its entries thrown away and rebuilt from disk # (`indexer._drop_root_entries`). The rebuilt entry has the same # content-hash id and none of the enrichment fields, so the diff above # reports neither an addition nor a deletion — and `_enriched_attempted` # still says "done" for a file whose album and cover no longer exist. # Found live: a Music library came back with its files and without its # albums, and stayed that way, because only a restart (which starts # with no snapshot, making every entry an addition) could clear either # gate. Treated here as what it is — those entries are new again. rebuilt_ids = indexer.drain_rescanned_ids() if rebuilt_ids: rebuilt = [e for e in idx.entries if e.id in rebuilt_ids] for entry in rebuilt: self._enriched_attempted.discard((group_id, entry.id)) seen = {e.id for e in new_entries} new_entries = new_entries + [e for e in rebuilt if e.id not in seen] spawn(self._enrich_new_video_entries(indexer, new_entries)) # Music app (docs/MESHBAY_DESIGN.md §9.8): same shape, gated on # audio_root exactly like video_root above (added later — the original # "no root, whole shared tree" call didn't hold up). spawn(self._enrich_new_audio_entries(indexer, new_entries)) # Photos app (docs/MESHBAY_DESIGN.md §9.9): same shape, gated on # photo_roots (a list, not a single string). spawn(self._enrich_new_photo_entries(indexer, new_entries)) # A rename/move changes the very filename (or season folder) that # docs/MESHBAY_DESIGN.md §9.7's title-parse read # display_title/season/episode from, but leaves the file's content — # and so its id and everything ffprobe/thumbnailing already found — # untouched. Only entries # whose name or path actually differ from the last broadcast get a # fresh pass; an update that is enrichment's own field-fill # (duration/thumb_hash/... landing via _on_enriched below) leaves # name/path alone and must not re-trigger itself forever. if delta is not None and delta.updates and previous is not None: spawn( self._reenrich_renamed_video_entries(indexer, delta.updates, previous)) spawn( self._reenrich_renamed_audio_entries(indexer, delta.updates, previous)) spawn( self._reenrich_renamed_photo_entries(indexer, delta.updates, previous)) # Videos/Music/Photos apps: a file that leaves the index also loses # its thumbnail/cover and file->tmdb/file->mbid mapping — the "real # deletion obligation" docs/MESHBAY_DESIGN.md §6.5 calls out # explicitly rather than leaving implicit (docs/MESHBAY_DESIGN.md §9.8 # follows the same rule). tmdb_meta/mbid_meta rows are left alone # (shared across files). # # Found live (docs/MESHBAY_DESIGN.md §9.9): a root removed and a new # one added for the identical content (an operator renaming/relocating a # shared folder) pruned the thumbnail here — correctly, the content # is gone from *this* root — but left the hash in # `_enriched_attempted`, which is never otherwise cleared. The same # bytes reappearing under the new root's path were then permanently # skipped: "already attempted" was true forever, for a thumbnail # that no longer existed. Discarding the attempt alongside the # cache entry is what makes pruning actually reversible — the next # sweep re-enriches it exactly as if it were new, which content # that is content-addressed and simply moved effectively is. if delta is not None and delta.deletions and self._media_cache: for file_id in delta.deletions: self._enriched_attempted.discard((indexer.group_id, file_id)) # media_cache.db is node-wide, keyed by content hash — a file # shared into two groups is one row there, same reasoning as # `_enriched_attempted`'s own docstring above. This group's # copy is genuinely gone (that's what a deletion delta is), # but another group may still hold the same content: only # prune once *no* group's index has this file_id any more, # or the surviving group pays for a redundant re-fetch/ # re-probe/re-thumbnail for content it never actually lost. still_referenced = any( i.index.get_entry(file_id) is not None for i in self._indexers) if not still_referenced: spawn(self._media_cache.prune_file(file_id)) # 11.5 — Push to connected WebRTC peers in this group if self._webrtc: peers = [s for s in list(self._webrtc._sessions.values()) if s._group_id == group_id] # Both messages are sealed under a GEK-derived subkey, so building one # needs a key. A group without one has no peers to push to either — the # node refuses every handshake while the GEK is None (NS8) — so this is # "nobody is listening", not a case to send in clear for. if peers and idx.gek: msg = (index_delta_message(idx, delta, indexer.roots) if delta is not None else index_sync_message(idx, indexer.roots)) pushed = 0 for session in peers: try: session._send(msg) pushed += 1 except Exception: pass if pushed: log.info("Index %s pushed to %d WebRTC peers", "delta" if delta is not None else "sync", pushed) # 11.9 — Register file hashes with hub swarm table (public groups only, H7) group_cfg = next( (g for g in self._config.groups if g.id == group_id), None) if (self._hub and self._state.get("endpoint_hint") and group_cfg and group_cfg.visibility == "public"): # Only the newly added hashes once there is a delta to know them # from — registering the whole library again on every change is # the same O(changes x library size) cost the delta above exists # to avoid. hashes = ([e.id for e in delta.additions] if delta is not None else [e.id for e in idx.entries]) if hashes: endpoint = f"webrtc:{self._config.node.quic_port}" spawn(self._register_swarm(hashes, endpoint)) async def _enrich_new_video_entries(self, indexer: DirectoryIndexer, entries: list) -> None: """ Videos app: fire (never await further) enrichment for unattempted video entries under the group's configured video_root. A group with no video_root set yet does not enrich anything — TMDB lookups and ffmpeg thumbnailing are real, ongoing per-file cost (docs/MESHBAY_DESIGN.md §6.5), and running them over an operator's whole shared index before they have chosen which folder is actually their media library would burn both TMDB's rate limit and the node's CPU on files that were never meant to be in the Videos app at all. Once a root is set, `_enrich_video_root_now` (called when it changes) separately sweeps whatever it already contains — this path alone only ever sees entries new since the last broadcast. """ if not self._enricher or not self._roster: return video_dirs = await self._roster.app_directories(indexer.group_id, "video") if not video_dirs: return for entry in entries: if entry.type != "video" or (indexer.group_id, entry.id) in self._enriched_attempted: continue if not _under_any_directory(entry.path, video_dirs): continue file_path = entry_abs_path(indexer.roots, entry) if not file_path or not file_path.exists(): continue self._enriched_attempted.add((indexer.group_id, entry.id)) async def on_done(file_id: str, fields: dict, _indexer=indexer) -> None: await self._on_enriched(_indexer, file_id, fields) self._enricher.spawn(entry, file_path, on_done) async def _enrich_video_root_now(self, group_id: str) -> None: """ Videos app: sweep a group's existing index for enrichment right after its video root is set or changed. The ordinary path above only ever looks at entries new since the last broadcast, so a folder that already had files sitting in it before it became the video_root would otherwise never get enriched at all — nothing else re-visits already-indexed entries once they have been broadcast once. """ indexer = self._state.get("indexers", {}).get(group_id) if not indexer: return await self._enrich_new_video_entries(indexer, list(indexer.index.entries)) async def _reenrich_renamed_video_entries( self, indexer: DirectoryIndexer, updates: list, previous: GroupIndex, ) -> None: """ Videos app: found live — a French-named episode file, renamed by the operator to match its English-named siblings, kept showing as its own separate poster-grid card (and its own row in Flat list) indefinitely, because `_enriched_attempted` — there specifically to stop enrichment's own field-fill from re-triggering itself forever (see the caller) — also silently blocked the *new* filename from ever being title-parsed at all. `entry.id in self._enriched_attempted` is the same content, so simply discarding it here and re-running the ordinary enrichment path is enough: a fresh ffprobe/thumbnail for an unchanged file is redundant work, not a correctness issue, and renames are rare enough that the redundancy is not worth a separate "title-parse only" code path. """ for entry in updates: if entry.type != "video": continue old = previous.get_entry(entry.id) if old is None or (old.name == entry.name and old.path == entry.path): continue self._enriched_attempted.discard((indexer.group_id, entry.id)) # The rename re-derives the title (the whole point of this # method), which can change the correct TMDB match — but # file_tmdb is keyed by content hash, unchanged by a rename, so # nothing else would ever dislodge the old name's match. A # manual "Fix match" correction is kept (clear_file_tmdb skips # anything in media_cache.tmdb_override). if self._media_cache is not None: await self._media_cache.clear_file_tmdb(entry.id) await self._enrich_new_video_entries(indexer, updates) async def _enrich_new_audio_entries(self, indexer: DirectoryIndexer, entries: list) -> None: """ Music app (docs/MESHBAY_DESIGN.md §9.8): fire (never await further) tag/cover enrichment for unattempted audio entries under the group's configured audio_root — same gate as `_enrich_new_video_entries` above (the original "no root, whole shared tree" call turned out wrong against a real messy library: everything under every shared folder got mixed together with no way to scope it down). `_enriched_attempted` is shared with the video path — content-addressed ids never collide across the two. """ if not self._audio_enricher or not self._roster: return audio_dirs = await self._roster.app_directories(indexer.group_id, "music") if not audio_dirs: return # Resolved once per directory, not once per file: a library is # thousands of entries and this is a filesystem call each time. boundaries = {d: indexer.roots.resolve(d, require_available=False) for d in audio_dirs} for entry in entries: if entry.type != "audio" or (indexer.group_id, entry.id) in self._enriched_attempted: continue owner = _owning_directory(entry.path, audio_dirs) if owner is None: continue file_path = entry_abs_path(indexer.roots, entry) if not file_path or not file_path.exists(): continue self._enriched_attempted.add((indexer.group_id, entry.id)) async def on_done(file_id: str, fields: dict, _indexer=indexer) -> None: await self._on_enriched(_indexer, file_id, fields) # The boundary is *the configured directory this file is under*, # not the shared root it lives in — so the ancestor walk # (enrich_audio._artist_album_from_ancestors) treats a flat # top-level folder right under the configured Music directory as # ambiguous (artist-or-release, docs/MESHBAY_DESIGN.md §9.8), # rather than one level too shallow when that directory is # itself a subfolder. With several configured, each file is # measured against its own: # a single shared boundary would be wrong for all but one of them. self._audio_enricher.spawn(entry, file_path, on_done, boundaries.get(owner)) async def _enrich_audio_root_now(self, group_id: str) -> None: """ Music app: sweep a group's existing index right after its audio root is set or changed. Mirrors `_enrich_video_root_now` exactly — the ordinary path above only ever looks at entries new since the last broadcast, so a folder that already had files in it before it became the audio_root would otherwise never get enriched at all. """ indexer = self._state.get("indexers", {}).get(group_id) if not indexer: return await self._enrich_new_audio_entries(indexer, list(indexer.index.entries)) async def _reenrich_renamed_audio_entries( self, indexer: DirectoryIndexer, updates: list, previous: GroupIndex, ) -> None: """ Music app equivalent of `_reenrich_renamed_video_entries` — a rename can change the filename-parse fallback (title/track_no) even though embedded tags, when present, are unaffected. Re-running the whole pass on a rename is redundant work for a tagged file and a real fix for an untagged one, and renames are rare enough not to need a cheaper, tags-only special case. """ for entry in updates: if entry.type != "audio": continue old = previous.get_entry(entry.id) if old is None or (old.name == entry.name and old.path == entry.path): continue self._enriched_attempted.discard((indexer.group_id, entry.id)) await self._enrich_new_audio_entries(indexer, updates) async def _enrich_new_photo_entries(self, indexer: DirectoryIndexer, entries: list) -> None: """ Photos app (docs/MESHBAY_DESIGN.md §9.9): fire (never await further) thumbnail/EXIF enrichment for unattempted image entries under any of the group's configured photo_roots. Same gate shape as `_enrich_new_video_entries`/`_enrich_new_audio_entries` — no root configured yet means no work, since thumbnailing every image in a whole shared tree before the operator has chosen which folders are actually photo albums would burn CPU on files never meant to be in the Photos app at all. `_enriched_attempted` is shared with the video/audio paths — content-addressed ids never collide across them. """ if not self._photo_enricher or not self._roster: return photo_dirs = await self._roster.app_directories(indexer.group_id, "photo") if not photo_dirs: return for entry in entries: if entry.type != "image" or (indexer.group_id, entry.id) in self._enriched_attempted: continue if not _under_any_directory(entry.path, photo_dirs): continue file_path = entry_abs_path(indexer.roots, entry) if not file_path or not file_path.exists(): continue self._enriched_attempted.add((indexer.group_id, entry.id)) async def on_done(file_id: str, fields: dict, _indexer=indexer) -> None: await self._on_enriched(_indexer, file_id, fields) self._photo_enricher.spawn(entry, file_path, on_done) async def _enrich_photo_roots_now(self, group_id: str) -> None: """ Photos app: sweep a group's existing index right after its photo roots change. Mirrors `_enrich_video_root_now`/`_enrich_audio_root_now` — the ordinary path above only ever looks at entries new since the last broadcast, so a folder that already had photos in it before it was added to photo_roots would otherwise never get enriched at all. Also covers a root being *removed*: nothing un-enriches on removal (the cache entry is harmless, just unused — the media cache is disposable and tied to the index, docs/MESHBAY_DESIGN.md §6.5), so re-sweeping the new set is enough. """ indexer = self._state.get("indexers", {}).get(group_id) if not indexer: return await self._enrich_new_photo_entries(indexer, list(indexer.index.entries)) async def _reenrich_renamed_photo_entries( self, indexer: DirectoryIndexer, updates: list, previous: GroupIndex, ) -> None: """ Photos app equivalent of `_reenrich_renamed_video_entries` — a rename changes nothing about the image's own bytes (thumbnail, EXIF fields are content-derived, not name-derived), so this exists only for consistency/symmetry with Videos/Music and to catch the case of a file moving *into* a newly-covered photo_roots subtree via a rename rather than a fresh add. Re-running enrichment on an unchanged file is redundant work, not a correctness issue. """ for entry in updates: if entry.type != "image": continue old = previous.get_entry(entry.id) if old is None or (old.name == entry.name and old.path == entry.path): continue self._enriched_attempted.discard((indexer.group_id, entry.id)) await self._enrich_new_photo_entries(indexer, updates) async def _on_enriched(self, indexer: DirectoryIndexer, file_id: str, fields: dict) -> None: """ Merge enrichment fields into the live index and re-trigger a broadcast so they reach connected clients as an INDEX_DELTA update (GroupIndex.diff's `updates`, not `additions` — same id, new fields). Builds a *new* IndexEntry via dataclasses.replace rather than mutating the existing one in place: the diff mechanism compares against a shallow snapshot of entry *references*, so an in-place mutation would silently also change what "previous" looks like, and the change would never show up as a diff (see group_index.py's diff() docstring). """ idx = indexer.index entry = idx.get_entry(file_id) if entry is None: return # removed from the index while enrichment was in flight idx.add_entry(replace(entry, **fields)) await self._on_index_change(indexer) def _drop_group_sessions(self, group_id: str) -> None: """Close live sessions for a revoked group (H4).""" if not self._webrtc or not group_id: return for session in list(self._webrtc._sessions.values()): if session._group_id == group_id: spawn(session.close()) log.info("Dropped session for revoked group %s", group_id[:8]) async def _register_swarm(self, hashes: list[str], endpoint: str) -> None: try: n = await self._hub.register_swarm(hashes, endpoint) log.info("Swarm: registered %d/%d hashes", n, len(hashes)) except Exception as e: log.warning("Swarm registration failed: %s", e) async def _shutdown(self) -> None: log.info("Shutting down...") self._state["status"] = "stopping" for handle in self._pending_broadcasts.values(): handle.cancel() self._pending_broadcasts.clear() for task in self._tasks: task.cancel() for task in self._tasks: try: await task except (asyncio.CancelledError, Exception): pass if self._webrtc: await self._webrtc.close_all() if self._audit_store: await self._audit_store.close() if self._bundle_store: await self._bundle_store.close() if self._tmdb_client: await self._tmdb_client.close() if self._musicbrainz_client: await self._musicbrainz_client.close() if self._media_cache: await self._media_cache.close() if self._roster: await self._roster.close() for store in self._chat_stores.values(): await store.close() if self._index_cache: await self._index_cache.close() for indexer in self._indexers: await indexer.stop() # The thread each group's roots are read from (roots.py `io_executor`). # Created on the first read, so a group nobody downloaded from never # started one and this is a no-op for it. groups = self._webrtc._ctx.get("groups") if self._webrtc else None for group in (groups or {}).values(): roots = group.get("roots") if roots is not None: roots.close_io() if self._quic_server: await self._quic_server.stop() token_file = getattr(self, "_ui_token_file", None) if token_file is not None: token_file.unlink(missing_ok=True) log.info("Node stopped") # ── Entry point ─────────────────────────────────────────────────────────────── def main() -> None: args = start() if run(args): return cfg = load_config(args.config or DEFAULT_CONFIG_PATH) if not cfg.hub.username: print("Error: hub.username not set in config. Run: meshbay-node init") sys.exit(1) from meshbay_node.platform import check_media_tools try: check_media_tools(cfg.node.ffmpeg_path, cfg.node.ffprobe_path) except RuntimeError as e: print(f"Error: {e}") sys.exit(1) daemon = NodeDaemon(cfg, Path(args.config or DEFAULT_CONFIG_PATH)) asyncio.run(daemon.run()) if __name__ == "__main__": main()