refactor(node): move _dispatch_message out of webrtc_server, unchanged

DispatchMixin in transport/webrtc/dispatch.py, with the pre-proof fetch
bound it enforces. The elif chain moves as it is; turning it into a table
is the next commit, on its own.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

1 files changed, 251 insertions, 0 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/dispatch.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/dispatch.py
new file mode 100644
index 0000000..cb2db8f
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/dispatch.py
@@ -0,0 +1,251 @@
+"""Which handler answers which MNP message, and the guards that come before any
+of them."""
+
+import logging
+import time
+
+from meshbay_common.protocol import MNP
+
+log = logging.getLogger("meshbay_node.transport.webrtc_server")
+
+
+# Bundle fetches are served in the pre-proof window (C4). Bounded and audited
+# until the native client removes remote keypair bundles entirely.
+MAX_PRE_PROOF_FETCHES = 4
+
+
+class DispatchMixin:
+    def _dispatch_message(self, msg: dict) -> None:
+        mtype = msg.get("type")
+        log.debug("WebRTC recv: %s", mtype)
+        try:
+            if mtype == MNP.HANDSHAKE:
+                self._do_handshake(msg)
+            elif mtype == MNP.HANDSHAKE_RESPONSE:
+                self._do_handshake_response(msg)
+            elif mtype in (MNP.GEK_BUNDLE_FETCH, MNP.KEYPAIR_BUNDLE_FETCH) \
+                    and self._gek_challenge is not None:
+                # Served before the GEK proof by necessity: the client needs its
+                # wrapped bundle in order to compute the proof. That window is a
+                # disclosure surface (C4) — a hub that forges a JWT reaches it — so
+                # it is bounded and audited here, and closed properly when clients
+                # stop storing keypair bundles on other people's nodes.
+                self._pre_proof_fetches += 1
+                if self._pre_proof_fetches > MAX_PRE_PROOF_FETCHES:
+                    self._audit_auth_failed(
+                        getattr(self, "_pending_group", ""), "pre-proof fetch flood")
+                    self._send({"type": "error", "detail": "Too many requests"})
+                    return
+                self._audit_pre_proof_fetch(mtype)
+                if mtype == MNP.GEK_BUNDLE_FETCH:
+                    self._spawn(self._do_gek_bundle_fetch())
+                else:
+                    self._spawn(self._do_keypair_bundle_fetch())
+            elif mtype == MNP.JOIN_REQUEST and self._nonce_node:
+                # Valid both before the GEK proof (a new member has no GEK to prove
+                # with) and after it (an operator pairing a browser is already
+                # connected). Authority comes from the pairing code and the
+                # signature, never from the session state.
+                self._spawn(self._do_join_request(msg))
+            elif self._user_id is None:
+                self._send({"type": "error", "detail": "Handshake required"})
+            elif mtype == MNP.INDEX_SYNC:
+                self._do_index_sync()
+            elif mtype == MNP.FILE_REQUEST:
+                # Spawned rather than answered inline: the reply waits for room
+                # on the channel, and blocking the message loop for that would
+                # stop everything else this peer is doing — including the
+                # uploads whose acks free the very buffer we are waiting on.
+                # Chunks are matched by file and index on the client, so
+                # answering out of order is safe.
+                self._spawn(self._do_file_request(msg))
+            elif mtype == MNP.CHAT_MESSAGE:
+                self._do_chat_message(msg)
+            elif mtype == MNP.CHAT_HISTORY:
+                self._do_chat_history(msg)
+            elif mtype == MNP.LINK_PREVIEW_REQ:
+                self._spawn(self._do_link_preview_request(msg))
+            elif mtype == MNP.PING:
+                self._do_ping(msg)
+            elif mtype == MNP.TRANSFER_OPEN:
+                self._do_transfer_open(msg)
+            elif mtype == MNP.TRANSFER_CLOSE:
+                self._do_transfer_close(msg)
+            elif mtype == MNP.FILE_UPLOAD:
+                self._spawn(self._do_file_upload(msg))
+            elif mtype == MNP.DIR_CREATE:
+                self._spawn(self._do_dir_create(msg))
+            elif mtype == MNP.DIR_DELETE:
+                self._spawn(self._do_dir_delete(msg))
+            elif mtype == MNP.FILE_DELETE:
+                self._do_file_delete(msg)
+            elif mtype == MNP.ADMIN_RESPONSE:
+                self._do_admin_response(msg)
+            elif mtype == MNP.INVITE_CREATE:
+                self._do_invite_create(msg)
+            elif mtype == MNP.INVITE_LINK_CREATE:
+                self._do_invite_link_create(msg)
+            elif mtype == MNP.INVITE_CANCEL:
+                self._do_invite_cancel(msg)
+            elif mtype == MNP.MEMBER_REVOKE:
+                self._do_member_revoke(msg)
+            elif mtype == MNP.DEVICE_REQUEST:
+                self._spawn(self._do_device_request(msg))
+            elif mtype == MNP.DEVICE_LOOKUP:
+                self._spawn(self._do_device_lookup(msg))
+            elif mtype == MNP.DEVICE_ADD:
+                self._spawn(self._do_device_add(msg))
+            elif mtype == MNP.DEVICE_LIST:
+                self._spawn(self._do_device_list(msg))
+            elif mtype == MNP.DEVICE_REVOKE:
+                self._spawn(self._do_device_revoke(msg))
+            elif mtype == MNP.DEVICE_HELLO:
+                self._spawn(self._do_device_hello(msg))
+            elif mtype == MNP.APPS_ENABLED:
+                self._do_apps_enabled(msg)
+            elif mtype == MNP.TRANSFER_LIMITS:
+                self._do_transfer_limits(msg)
+            elif mtype == MNP.SET_SCAN_SETTINGS:
+                self._do_set_scan_settings(msg)
+            elif mtype == MNP.TMDB_CONFIG:
+                self._do_tmdb_config(msg)
+            elif mtype == MNP.TMDB_ENABLED:
+                self._do_tmdb_enabled(msg)
+            elif mtype == MNP.APP_DIRECTORIES:
+                self._do_app_directories(msg)
+            elif mtype == MNP.CHAT_DIRECTORY:
+                self._do_chat_directory(msg)
+            elif mtype == MNP.CHAT_LINK_PREVIEW:
+                self._do_chat_link_preview(msg)
+            elif mtype == MNP.SEARCH_LISTED:
+                self._do_search_listed(msg)
+            elif mtype == MNP.CHAT_EPOCH:
+                self._do_chat_epoch(msg)
+            elif mtype == MNP.CHAT_KEYS_REQ:
+                self._spawn(self._do_chat_keys_req(msg))
+            elif mtype == MNP.GROUP_ROSTER_REQ:
+                self._spawn(self._do_group_roster_req(msg))
+            elif mtype == MNP.MEDIA_META_REQ:
+                self._spawn(self._do_media_meta_request(msg))
+            elif mtype == MNP.SEASON_META_REQ:
+                self._spawn(self._do_season_meta_request(msg))
+            elif mtype == MNP.TMDB_SEARCH_REQ:
+                self._spawn(self._do_tmdb_search_request(msg))
+            elif mtype == MNP.TMDB_OVERRIDE:
+                self._do_tmdb_override(msg)
+            elif mtype == MNP.TMDB_REMATCH:
+                self._do_tmdb_rematch(msg)
+            elif mtype == MNP.MUSICBRAINZ_ENABLED:
+                self._do_musicbrainz_enabled(msg)
+            elif mtype == MNP.MUSIC_META_REQ:
+                self._spawn(self._do_music_meta_request(msg))
+            elif mtype == MNP.AUDIO_TRANSCODE_REQ:
+                self._spawn(self._do_audio_transcode_request(msg))
+            elif mtype == MNP.SUBTITLE_REQ:
+                self._spawn(self._do_subtitle_request(msg))
+            elif mtype == MNP.MEMBER_UNPIN:
+                self._do_member_unpin(msg)
+            elif mtype == MNP.GEK_ROTATE:
+                self._do_gek_rotate(msg)
+            elif mtype == MNP.NODE_STATUS:
+                self._spawn(self._do_node_status(msg))
+            elif mtype == MNP.ROOT_ADD:
+                self._do_root_add(msg)
+            elif mtype == MNP.ROOT_REMOVE:
+                self._do_root_remove(msg)
+            elif mtype == MNP.ROOT_UPDATE:
+                self._do_root_update(msg)
+            elif mtype == MNP.ROOT_EJECT:
+                self._do_root_eject(msg)
+            elif mtype == MNP.ROOT_PLUG:
+                self._do_root_plug(msg)
+            elif mtype == MNP.ROSTER_READ:
+                self._spawn(self._do_roster_read(msg))
+            elif mtype == MNP.DENYLIST_READ:
+                self._spawn(self._do_denylist_read(msg))
+            elif mtype == MNP.DENYLIST_CLEAR:
+                self._spawn(self._do_denylist_clear(msg))
+            elif mtype == MNP.GROUP_ATTACH:
+                self._do_group_attach(msg)
+            elif mtype == MNP.GROUP_DETACH:
+                self._do_group_detach(msg)
+            elif mtype == MNP.NODE_SETTINGS_SET:
+                self._spawn(self._do_node_settings_set(msg))
+            elif mtype == MNP.NODE_RELOAD:
+                self._spawn(self._do_node_reload(msg))
+            elif mtype == MNP.KEYPAIR_BUNDLE_STORE:
+                self._spawn(self._do_keypair_bundle_store(msg))
+            elif mtype == MNP.KEYPAIR_BUNDLE_DELETE:
+                self._spawn(self._do_keypair_bundle_delete())
+            elif mtype == MNP.USER_BLOB_STORE:
+                self._spawn(self._do_user_blob_store(msg))
+            elif mtype == MNP.USER_BLOB_FETCH:
+                self._spawn(self._do_user_blob_fetch(msg))
+            elif mtype == MNP.USER_BLOB_LIST:
+                self._spawn(self._do_user_blob_list())
+            elif mtype == MNP.USER_BLOB_DELETE:
+                self._spawn(self._do_user_blob_delete(msg))
+            elif mtype == MNP.STREAM_REQUEST:
+                sem = self._ctx.get("_transcode_sem")
+                log.info("stream: req file=%s credits=%s slots_free=%s prev=%s",
+                         str(msg.get("file_id"))[:12], msg.get("credits"),
+                         getattr(sem, "_value", "?"),
+                         "alive" if (self._stream_task and
+                                     not self._stream_task.done()) else "none")
+                self._spawn(self._replace_stream(msg))
+            elif mtype == MNP.STREAM_MORE:
+                self._grant_stream_credit(msg)
+            elif mtype == "client_diag":
+                # Diagnostics only. The node acts on none of it — it writes it
+                # next to its own view of the same stream, which is the only
+                # place the two halves can be compared when the client is a
+                # phone with no console.
+                # Every field is peer-controlled, so each is stringified and
+                # cut short: this is a log line, not a channel for writing
+                # whatever one likes into the operator's file.
+                def _f(key: str, n: int = 24) -> str:
+                    return str(msg.get(key))[:n].replace("\n", " ")
+                if msg.get("event"):
+                    # Once per stream or per seek, not once per five seconds —
+                    # and a seek nobody asked for looks exactly like a viewer
+                    # dragging the scrubber from this side, so it has to be
+                    # visible without turning DEBUG on.
+                    log.info(
+                        "stream: client %s target=%s t=%ss offset=%s ready=%s "
+                        "duration=%s ranges=[%s]",
+                        _f("event", 16), _f("target"), _f("t"), _f("offset"),
+                        _f("ready"), _f("duration"), _f("ranges", 120))
+                # Debug: one line every five seconds per viewer. Run the daemon
+                # with --log-level debug to see inside a player that is
+                # misbehaving — it is the only view of the browser there is
+                # when the browser is a phone.
+                else:
+                    # `ahead` on its own cannot say whether a short buffer is
+                    # the player's own gate holding or the network failing to
+                    # keep up, and those two want opposite answers. `limit` is
+                    # what the gate is set to for this film and `budget` the
+                    # byte budget it was derived from, so the three read as one
+                    # sentence.
+                    log.debug(
+                        "stream: client t=%ss ahead=%ss/%ss budget=%sMB "
+                        "ready=%s paused=%s "
+                        "stalled=%s q=%s inflight=%s appending=%s updating=%s "
+                        "quota=%s ms=%s err=%s ranges=[%s] (sent=%d)",
+                        _f("t"), _f("ahead"), _f("limit"), _f("budgetMB"),
+                        _f("ready"), _f("paused"),
+                        _f("stalled"), _f("q"), _f("inflight"), _f("appending"),
+                        _f("updating"), _f("quota"), _f("ms"), _f("err", 80),
+                        _f("ranges", 120), self._stream_segments)
+            elif mtype == MNP.STREAM_STOP:
+                age = (time.monotonic() - self._stream_started_at
+                       if self._stream_started_at else -1)
+                log.info("stream: stop received %.1fs after start, %d segments sent",
+                         age, self._stream_segments)
+                self._stop_stream()
+            else:
+                log.warning("Unknown MNP message type on DataChannel: %s", mtype)
+        except Exception as e:
+            # Log the detail locally; send the peer a generic message. Exception
+            # text here carries filesystem paths and internal state (finding L3).
+            log.error("Error handling %s on DataChannel: %s", mtype, e, exc_info=True)
+            self._send({"type": "error", "detail": "Request failed"})