1 files changed, 133 insertions, 0 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 1d10aa8..ab95550 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -46,6 +46,58 @@ CHUNK_SIZE = 1024 * 1024
 MAX_MSG = 64 * 1024 * 1024
 
 
+STREAM_SEGMENT_SIZE = 256 * 1024
+
+_H264_PROFILES = {"Baseline": "42", "Main": "4d", "High": "64", "High 10": "6e"}
+
+
+async def _probe_video(path: str) -> tuple[str | None, float]:
+    """Probe video file with ffprobe, return (MSE codec string, duration)."""
+    import json as _json
+    proc = await asyncio.create_subprocess_exec(
+        "ffprobe", "-v", "error",
+        "-show_entries", "stream=codec_name,profile,level,codec_type",
+        "-show_entries", "format=duration",
+        "-of", "json", path,
+        stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
+    )
+    stdout, _ = await proc.communicate()
+    info = _json.loads(stdout)
+    duration = float(info.get("format", {}).get("duration", 0))
+
+    v_codec = a_codec = ""
+    for s in info.get("streams", []):
+        if s.get("codec_type") == "video" and not v_codec:
+            cn = s.get("codec_name", "")
+            if cn == "h264":
+                p = _H264_PROFILES.get(s.get("profile", "High"), "64")
+                lvl = int(s.get("level", 40))
+                v_codec = f"avc1.{p}00{lvl:02x}"
+            elif cn == "hevc":
+                v_codec = "hev1.1.6.L93.B0"
+            elif cn == "vp9":
+                v_codec = "vp09.00.10.08"
+            elif cn == "av1":
+                v_codec = "av01.0.01M.08"
+        elif s.get("codec_type") == "audio" and not a_codec:
+            cn = s.get("codec_name", "")
+            if cn == "aac":
+                a_codec = "mp4a.40.2"
+            elif cn in ("mp3", "mp2"):
+                a_codec = "mp4a.6b"
+            elif cn == "opus":
+                a_codec = "opus"
+            elif cn == "ac3":
+                a_codec = "ac-3"
+            elif cn == "flac":
+                a_codec = "flac"
+
+    if not v_codec:
+        return None, duration
+    codec = f"{v_codec},{a_codec}" if a_codec else v_codec
+    return codec, duration
+
+
 def _pack(obj: dict) -> bytes:
     data = msgpack.packb(obj, use_bin_type=True)
     return struct.pack(">I", len(data)) + data
@@ -118,6 +170,8 @@ class WebRTCPeerSession:
                 self._do_file_upload(msg)
             elif mtype == MNP.FILE_DELETE:
                 self._do_file_delete(msg)
+            elif mtype == MNP.STREAM_REQUEST:
+                asyncio.ensure_future(self._stream_video(msg))
             else:
                 log.warning("Unknown MNP message type on DataChannel: %s", mtype)
         except Exception as e:
@@ -417,6 +471,85 @@ class WebRTCPeerSession:
             "file_id": file_id,
         })
 
+    async def _stream_video(self, msg: dict) -> None:
+        """Stream a video file as fMP4 segments via MSE-compatible output."""
+        ctx = self._group_ctx()
+        file_id = msg.get("file_id", "")
+        entry = ctx["index"].get_entry(file_id)
+        if not entry:
+            self._send({"type": "error", "detail": "File not found"})
+            return
+
+        file_path = ctx["shared_root"] / entry.path / entry.name
+        if not file_path.exists():
+            self._send({"type": "error", "detail": "File not on disk"})
+            return
+
+        gek = ctx.get("gek")
+        file_hash = bytes.fromhex(entry.id)
+
+        try:
+            codec_str, duration = await _probe_video(str(file_path))
+        except Exception as e:
+            self._send({"type": "error", "detail": f"Probe failed: {e}"})
+            return
+
+        if not codec_str:
+            self._send({"type": "error", "detail": "Unsupported video codec"})
+            return
+
+        proc = await asyncio.create_subprocess_exec(
+            "ffmpeg", "-hide_banner", "-loglevel", "error",
+            "-i", str(file_path),
+            "-c", "copy",
+            "-movflags", "frag_keyframe+empty_moov+default_base_moof",
+            "-f", "mp4", "pipe:1",
+            stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
+        )
+
+        self._send({
+            "type": MNP.STREAM_INIT,
+            "v": MNP_VERSION,
+            "file_id": file_id,
+            "codec": codec_str,
+            "duration": duration,
+        })
+
+        index = 0
+        try:
+            while True:
+                data = await proc.stdout.read(STREAM_SEGMENT_SIZE)
+                if not data:
+                    break
+                ckey = chunk_key_aes(gek, file_hash, index)
+                nonce, ct = encrypt_chunk_aes(ckey, data)
+                self._send({
+                    "type": MNP.STREAM_DATA,
+                    "v": MNP_VERSION,
+                    "file_id": file_id,
+                    "segment_index": index,
+                    "nonce": nonce,
+                    "ct": ct,
+                    "plaintext_size": len(data),
+                })
+                index += 1
+                await asyncio.sleep(0)
+        except Exception as e:
+            log.error("Stream error: %s", e)
+        finally:
+            try:
+                proc.kill()
+            except ProcessLookupError:
+                pass
+            await proc.wait()
+
+        self._send({
+            "type": MNP.STREAM_END,
+            "v": MNP_VERSION,
+            "file_id": file_id,
+        })
+        log.info("Streamed %s: %d segments", entry.name, index)
+
     def _send(self, obj: dict) -> None:
         if self._channel and self._channel.readyState == "open":
             self._channel.send(_pack(obj))