refactor(node): move framing, shared limits and ffmpeg jobs out of webrtc_server

transport/webrtc/channel.py, limits.py and media_tools.py, cut from
webrtc_server.py as text; the facade imports them back.

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

1 files changed, 96 insertions, 0 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/channel.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/channel.py
new file mode 100644
index 0000000..107a43e
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/channel.py
@@ -0,0 +1,96 @@
+"""The data channel itself: framing, the reply correlation id, the peer's
+address and DTLS fingerprint."""
+
+import contextvars
+import struct
+
+import msgpack
+from aiortc import RTCPeerConnection
+
+from meshbay_node.transport.webrtc.limits import MAX_MSG
+
+
+def _extract_dtls_fingerprint(sdp: str) -> bytes:
+    """Extract the DTLS SHA-256 fingerprint from SDP as raw 32 bytes."""
+    for line in sdp.splitlines():
+        if line.startswith("a=fingerprint:sha-256 "):
+            hex_str = line.split(" ", 1)[1].replace(":", "")
+            return bytes.fromhex(hex_str)
+    return b""
+
+
+def _pack(obj: dict) -> bytes:
+    data = msgpack.packb(obj, use_bin_type=True)
+    return struct.pack(">I", len(data)) + data
+
+
+# The request this session is currently answering, as (session, req_id).
+#
+# MNP has never carried a correlation id: a reply named its own type and
+# nothing else, so a client with more than one request outstanding had to guess
+# which one a message answered — by arrival order, for every reply the client
+# could not key off a field of its own. The guess is wrong whenever two replies
+# reorder, and catastrophically wrong for the replies that name *nothing*: this
+# module sends `{"type": "error"}` from 240 places and two of them name what
+# they are about. A refusal therefore reached no caller at all, and the request
+# it belonged to waited out the client's 30s timeout while some unrelated
+# request was resolved with the refusal instead. Live symptom, found 2026-09-06:
+# the Chat composer is disabled while a send is in flight, so a chat message
+# whose reply went astray froze the tab for 30 seconds.
+#
+# `req_id` closes it: whatever the caller put on the request is stamped on the
+# reply. A ContextVar rather than a parameter because the alternative is
+# threading an argument through all 240 send sites — and asyncio copies the
+# current context into a task, so a handler that `_spawn`s its real work still
+# answers under the id of the request that started it.
+#
+# The session is held alongside the id because a handler may send to *other*
+# sessions as well as its own (a chat broadcast, an index push): those are not
+# replies to anything and must not be stamped. _send checks the owner.
+_REPLY_TO: contextvars.ContextVar[tuple] = contextvars.ContextVar(
+    "meshbay_reply_to", default=(None, None))
+
+
+class _DataChannelBuffer:
+    """
+    Accumulate DataChannel messages and extract length-prefixed msgpack.
+
+    Finding H6: the limit was a flat 64 MB applied even before the handshake, so an
+    unauthenticated peer could announce a 64 MB frame and dribble bytes into it,
+    holding that much memory per connection. Until a peer has proved GEK
+    possession it gets a small budget; the large one is for file uploads.
+    """
+
+    def __init__(self, max_message: int = MAX_MSG):
+        self._buf = bytearray()
+        self.max_message = max_message
+
+    def feed(self, data: bytes):
+        self._buf.extend(data)
+
+    def messages(self):
+        while len(self._buf) >= 4:
+            length = struct.unpack(">I", self._buf[:4])[0]
+            if length > self.max_message:
+                raise ValueError(f"Message too large: {length}")
+            if len(self._buf) < 4 + length:
+                break
+            msg_bytes = bytes(self._buf[4:4 + length])
+            del self._buf[:4 + length]
+            yield msgpack.unpackb(msg_bytes, raw=False)
+
+
+def _get_remote_ip(pc: RTCPeerConnection) -> str:
+    """Best-effort extraction of the remote peer IP from the ICE transport."""
+    try:
+        dtls = pc.sctp and pc.sctp.transport
+        ice = dtls and dtls.transport
+        conn = ice and ice._connection
+        if conn and hasattr(conn, '_nominated') and conn._nominated:
+            for pair in conn._nominated.values():
+                return pair.remote_candidate.host
+        if conn and conn.remote_candidates:
+            return conn.remote_candidates[0].host
+    except Exception:
+        pass
+    return ""