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, 2 insertions, 243 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
index 92e1cd4..20636fa 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -31,9 +31,6 @@ from aiortc import RTCDataChannel, RTCPeerConnection, RTCSessionDescription
 from cryptography.hazmat.primitives.asymmetric.ed25519 import (
     Ed25519PrivateKey,
 )
-from meshbay_common.protocol import (
-    MNP,
-)
 
 from meshbay_node import transfers as transfers_mod
 from meshbay_node.indexer import GroupIndex
@@ -54,6 +51,7 @@ from meshbay_node.transport.webrtc.apps.video_meta import VideoMetaMixin
 from meshbay_node.transport.webrtc.blobs import BlobsMixin
 from meshbay_node.transport.webrtc.chat import ChatMixin
 from meshbay_node.transport.webrtc.core import _WEBRTC_TRACE, SessionCore
+from meshbay_node.transport.webrtc.dispatch import DispatchMixin
 from meshbay_node.transport.webrtc.files import FilesMixin
 from meshbay_node.transport.webrtc.group_ops import GroupOpsMixin
 from meshbay_node.transport.webrtc.handshake import HandshakeMixin
@@ -75,254 +73,15 @@ log = logging.getLogger(__name__)
 MAX_PEER_SESSIONS = 64
 UNAUTHENTICATED_SESSION_TIMEOUT = 60   # seconds
 
-# 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 WebRTCPeerSession(
-    AdminMixin, AdmissionMixin, BlobsMixin, ChatMixin, FilesMixin, GroupOpsMixin,
+    DispatchMixin, AdminMixin, AdmissionMixin, BlobsMixin, ChatMixin, FilesMixin, GroupOpsMixin,
     HandshakeMixin, NodeOpsMixin, TransferMixin, UploadMixin,
     StreamingMixin, VideoMetaMixin, MusicMixin, SubtitlesMixin,
     SessionCore,
 ):
     """One WebRTC peer connection, handling MNP over a DataChannel."""
 
-    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"})
-
 
 class WebRTCTransport:
     """