meerkat_live/
wire_input.rs

1//! Shared live-input wire-contract decode owner (D243).
2//!
3//! Every Meerkat live transport — the RPC `live/send_input` JSON-RPC method,
4//! the direct `/live/ws` WebSocket text path, and the WebRTC data channel —
5//! receives client input as the wire-format [`LiveInputChunkWire`]
6//! (`meerkat-contracts`) and must convert it into the core
7//! [`LiveInputChunk`] (`meerkat-core::live_adapter`) through a **single**
8//! conversion + validation step before handing it to the host.
9//!
10//! Previously each direct transport deserialized the core `LiveInputChunk`
11//! shape directly (`serde_json::from_slice::<LiveInputChunk>`), bypassing the
12//! generated `LiveInputChunkWire` boundary that the RPC path runs. That split
13//! meant a payload rejected by `live/send_input` wire validation could still be
14//! accepted on the WebRTC / WS direct paths. This module is the one
15//! conversion owner so all transports validate identically; the base64 decode
16//! is the only step that can fail and it fails with a typed
17//! [`LiveInputChunkDecodeError`] rather than a free-form provider string.
18
19use base64::Engine;
20use meerkat_contracts::LiveInputChunkWire;
21use meerkat_core::live_adapter::LiveInputChunk;
22
23/// Typed failure decoding a wire-format [`LiveInputChunkWire`] into the core
24/// [`LiveInputChunk`]. Each base64 field that can be malformed lands in a
25/// distinct variant so callers route on the typed class instead of reparsing a
26/// prose string.
27#[derive(Debug, thiserror::Error)]
28#[non_exhaustive]
29pub enum LiveInputChunkDecodeError {
30    /// The audio chunk's base64 `data` field was not valid base64.
31    #[error("invalid base64 in live audio input chunk: {0}")]
32    InvalidAudioBase64(#[source] base64::DecodeError),
33    /// The image chunk's base64 `data` field was not valid base64.
34    #[error("invalid base64 in live image input chunk: {0}")]
35    InvalidImageBase64(#[source] base64::DecodeError),
36    /// The video-frame chunk's base64 `data` field was not valid base64.
37    #[error("invalid base64 in live video-frame input chunk: {0}")]
38    InvalidVideoFrameBase64(#[source] base64::DecodeError),
39}
40
41/// Decode a wire-format input chunk into the core [`LiveInputChunk`] shape.
42///
43/// This is the canonical conversion owner shared by every live transport so
44/// the wire contract validation (base64 decoding of `data`) is identical
45/// everywhere. The RPC `live/send_input` handler and the direct WS / WebRTC
46/// transports route through this function.
47pub fn live_input_chunk_from_wire(
48    wire: LiveInputChunkWire,
49) -> Result<LiveInputChunk, LiveInputChunkDecodeError> {
50    match wire {
51        LiveInputChunkWire::Audio {
52            data,
53            sample_rate_hz,
54            channels,
55        } => {
56            let decoded = base64::engine::general_purpose::STANDARD
57                .decode(&data)
58                .map_err(LiveInputChunkDecodeError::InvalidAudioBase64)?;
59            Ok(LiveInputChunk::Audio {
60                data: decoded,
61                sample_rate_hz,
62                channels,
63            })
64        }
65        LiveInputChunkWire::Text { text } => Ok(LiveInputChunk::Text { text }),
66        LiveInputChunkWire::Image { mime, data } => {
67            let decoded = base64::engine::general_purpose::STANDARD
68                .decode(&data)
69                .map_err(LiveInputChunkDecodeError::InvalidImageBase64)?;
70            Ok(LiveInputChunk::Image {
71                mime,
72                data: decoded,
73            })
74        }
75        LiveInputChunkWire::VideoFrame {
76            codec,
77            data,
78            timestamp_ms,
79        } => {
80            let decoded = base64::engine::general_purpose::STANDARD
81                .decode(&data)
82                .map_err(LiveInputChunkDecodeError::InvalidVideoFrameBase64)?;
83            Ok(LiveInputChunk::VideoFrame {
84                codec,
85                data: decoded,
86                timestamp_ms,
87            })
88        }
89    }
90}
91
92#[cfg(test)]
93#[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
94mod tests {
95    use super::*;
96
97    // D243: a chunk rejected by the wire decode owner (malformed base64) is
98    // rejected by the single shared helper that every transport runs — the WS
99    // text path, the WebRTC data channel, and RPC `live/send_input` all route
100    // through `live_input_chunk_from_wire`, so a payload that fails here fails
101    // on every transport.
102    #[test]
103    fn malformed_audio_base64_is_rejected_with_typed_variant() {
104        let wire = LiveInputChunkWire::Audio {
105            data: "not valid base64!!!".to_string(),
106            sample_rate_hz: 24_000,
107            channels: 1,
108        };
109        match live_input_chunk_from_wire(wire) {
110            Err(LiveInputChunkDecodeError::InvalidAudioBase64(_)) => {}
111            other => panic!("expected InvalidAudioBase64, got {other:?}"),
112        }
113    }
114
115    #[test]
116    fn malformed_image_base64_is_rejected_with_typed_variant() {
117        let wire = LiveInputChunkWire::Image {
118            mime: "image/png".to_string(),
119            data: "@@@not-base64@@@".to_string(),
120        };
121        match live_input_chunk_from_wire(wire) {
122            Err(LiveInputChunkDecodeError::InvalidImageBase64(_)) => {}
123            other => panic!("expected InvalidImageBase64, got {other:?}"),
124        }
125    }
126
127    #[test]
128    fn malformed_video_frame_base64_is_rejected_with_typed_variant() {
129        let wire = LiveInputChunkWire::VideoFrame {
130            codec: "vp8".to_string(),
131            data: "%%%".to_string(),
132            timestamp_ms: 42,
133        };
134        match live_input_chunk_from_wire(wire) {
135            Err(LiveInputChunkDecodeError::InvalidVideoFrameBase64(_)) => {}
136            other => panic!("expected InvalidVideoFrameBase64, got {other:?}"),
137        }
138    }
139
140    #[test]
141    fn well_formed_wire_chunks_decode_to_core_chunks() {
142        let encoded = base64::engine::general_purpose::STANDARD.encode([1u8, 2, 3, 4]);
143
144        let audio = live_input_chunk_from_wire(LiveInputChunkWire::Audio {
145            data: encoded.clone(),
146            sample_rate_hz: 24_000,
147            channels: 1,
148        })
149        .expect("valid audio wire chunk decodes");
150        assert!(matches!(
151            audio,
152            LiveInputChunk::Audio { ref data, .. } if data == &[1u8, 2, 3, 4]
153        ));
154
155        let text = live_input_chunk_from_wire(LiveInputChunkWire::Text {
156            text: "hello".to_string(),
157        })
158        .expect("text wire chunk decodes");
159        assert!(matches!(text, LiveInputChunk::Text { text } if text == "hello"));
160
161        let image = live_input_chunk_from_wire(LiveInputChunkWire::Image {
162            mime: "image/png".to_string(),
163            data: encoded.clone(),
164        })
165        .expect("valid image wire chunk decodes");
166        assert!(matches!(
167            image,
168            LiveInputChunk::Image { ref mime, ref data } if mime == "image/png" && data == &[1u8, 2, 3, 4]
169        ));
170
171        let frame = live_input_chunk_from_wire(LiveInputChunkWire::VideoFrame {
172            codec: "vp8".to_string(),
173            data: encoded,
174            timestamp_ms: 7,
175        })
176        .expect("valid video-frame wire chunk decodes");
177        assert!(matches!(
178            frame,
179            LiveInputChunk::VideoFrame { ref codec, ref data, timestamp_ms }
180                if codec == "vp8" && data == &[1u8, 2, 3, 4] && timestamp_ms == 7
181        ));
182    }
183}
meerkat_live/wire_input.rs

meerkat_live/
wire_input.rs
