feat(dgw): stream reconnecting WebM sessions

crates/video-streamer/README.md

@@ -1,8 +1,37 @@
 # video-streamer
 
-This crate takes an unseekable WebM recording (typically from Chrome CaptureStream) and rewrites it into a “fresh” WebM stream that can start playing immediately.
-It does this by parsing the incoming WebM, finding the correct cut point, and re-encoding frames.
-The output stream begins with a keyframe and valid headers.
+This crate rewrites an unseekable WebM recording into a stream that can start playing immediately.
+
+`webm_stream` still serves one growing file over the original Start/Pull protocol.
+`stream_session` accepts a multi-clip recording event stream, reconnects across clips, and emits independent VP8 WebM segments over the same Start/Pull codes.
+
+The input event grammar is:
+
+```text
+(ClipStarted Bytes* CaughtUp Bytes* ClipEnded)* SessionEnded
+```
+
+`ClipEnded` closes one input clip but does not end the recording session.
+After `ClipEnded`, an existing viewer waits for a reconnecting `ClipStarted` until `SessionEnded` confirms the final end.
+
+## Session protocol
+
+The client sends `Start` (`00`) once.
+After fully handling `Segment started` or `Chunk`, the client sends one `Pull` (`01`).
+The client does not send `Pull` after `Error` or `Stream ended`.
+The server sends exactly one response for each accepted request and buffers at most one early `Pull` while a response is pending.
+An accepted queued `Pull` receives its own `Stream ended` response if the session ends before more segment data arrives.
+If another overlapping request exceeds that limit, the current and queued requests receive `Error`, the excess request is rejected, and the stream fails.
+
+`Segment started` (`01` + JSON) carries `{codec,sequence,width,height}` and begins an independent WebM segment.
+The output `sequence` starts at zero, is independent of the input `ClipStarted.sequence`, and increments for each output segment.
+A reconnecting clip or resolution change starts the next output segment.
+Another `Segment started` message implicitly closes the previous segment.
+Legacy `{codec}` metadata remains valid for one segment with sequence zero.
+
+`Chunk` (`00` + bytes) belongs to the current segment.
+`Stream ended` (`03`) cleanly closes the final segment and confirms that the recording session ended.
+`Error` (`02` + JSON), an abrupt transport close, or a transport error does not confirm a clean session end.
 
 ## Prerequisites
 

crates/video-streamer/src/decoder.rs

@@ -0,0 +1,55 @@
+use anyhow::Context as _;
+use cadeau::xmf::vpx::{VpxCodec, VpxDecoder, VpxImage};
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub(crate) struct Dimensions {
+    pub width: u32,
+    pub height: u32,
+}
+
+pub(crate) struct DecodedFrame<'decoder> {
+    pub image: VpxImage<'decoder>,
+    pub dimensions: Dimensions,
+}
+
+pub(crate) struct InputDecoder {
+    codec: VpxCodec,
+    threads: u32,
+    decoder: Option<VpxDecoder>,
+}
+
+impl InputDecoder {
+    pub(crate) fn new(codec: VpxCodec, threads: u32) -> Self {
+        Self {
+            codec,
+            threads,
+            decoder: None,
+        }
+    }
+
+    pub(crate) fn decode<'decoder>(&'decoder mut self, data: &[u8]) -> anyhow::Result<DecodedFrame<'decoder>> {
+        if self.decoder.is_none() {
+            self.decoder = Some(
+                VpxDecoder::builder()
+                    .threads(self.threads)
+                    .width(0)
+                    .height(0)
+                    .codec(self.codec)
+                    .build()?,
+            );
+        }
+
+        let decoder = self.decoder.as_mut().context("input decoder is missing")?;
+        decoder.decode(data)?;
+        let image = decoder.next_frame()?;
+        let dimensions = Dimensions {
+            width: image.width(),
+            height: image.height(),
+        };
+        anyhow::ensure!(
+            dimensions.width > 0 && dimensions.height > 0,
+            "decoder returned invalid frame dimensions"
+        );
+        Ok(DecodedFrame { image, dimensions })
+    }
+}

crates/video-streamer/src/lib.rs

@@ -25,7 +25,11 @@ macro_rules! perf_debug {
 
 pub mod config;
 pub mod debug;
+mod decoder;
+mod normalizer;
+mod protocol;
 pub mod reopenable;
+mod session;
 pub(crate) mod streamer;
 
 #[macro_use]
@@ -39,6 +43,8 @@ pub use streamer::reopenable_file::ReOpenableFile;
 pub use streamer::signal_writer::SignalWriter;
 #[rustfmt::skip]
 pub use streamer::webm_stream;
+#[rustfmt::skip]
+pub use session::{RecordingEvent, SessionConfig, StartAt, stream_session};
 
 #[cfg(feature = "bench")]
 pub mod bench_support;