diff --git a/crates/rutster-trunk/src/twilio_media_streams.rs b/crates/rutster-trunk/src/twilio_media_streams.rs index 7cf910b..93a6053 100644 --- a/crates/rutster-trunk/src/twilio_media_streams.rs +++ b/crates/rutster-trunk/src/twilio_media_streams.rs @@ -1,4 +1,355 @@ -//! TwilioMediaStreamsServer -- axum WSS handler stub for atomic-commit-trial. -// Pivoting to commit-first approach to avoid external process interference; T3 quality restored in next edit cycle. +//! # TwilioMediaStreamsServer -- the inbound WSS pump task. -// Stub -- intentionally empty; full impl restarts under atomic workflow after commit. +use axum::{ + extract::{ + ws::{Message, WebSocket, WebSocketUpgrade}, + State, + }, + response::IntoResponse, + routing::get, + Router, +}; +use base64::Engine as _; +use rutster_call_model::ChannelId; +use rutster_media::PcmFrame; +use serde::{Deserialize, Serialize}; +use tokio::sync::{mpsc, oneshot}; +use tracing::{debug, info, warn}; + +use crate::g711::G711Codec; + +const MEDIA_CHANNEL_CAPACITY: usize = 16; + +pub struct RegisterTrunkInboundChannel { + pub call_sid: String, + pub inbound_from_twilio_rx: mpsc::Receiver, + pub outbound_to_twilio_tx: mpsc::Sender, + pub tap_url: url::Url, + pub reply: oneshot::Sender, +} + +pub struct TwilioMediaStreamsServer; + +impl TwilioMediaStreamsServer { + pub fn router(register_tx: mpsc::Sender) -> Router { + Router::new() + .route("/twilio/media-stream", get(handle_media_stream)) + .with_state(register_tx) + } +} + +#[derive(Debug, Deserialize)] +#[serde(tag = "event", rename_all = "lowercase")] +pub enum TwilioMediaEvent { + Connected, + Start { start: StartPayload }, + Media { media: MediaPayload }, + Stop, +} + +#[derive(Debug, Deserialize, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct StartPayload { + pub stream_sid: String, + pub call_sid: String, +} + +#[derive(Debug, Deserialize)] +pub struct MediaPayload { + pub payload: String, +} + +#[derive(Debug, Serialize)] +struct OutboundMediaFrame<'a> { + event: &'a str, + #[serde(rename = "streamSid")] + stream_sid: &'a str, + media: OutboundMediaPayload, +} + +#[derive(Debug, Serialize)] +struct OutboundMediaPayload { + payload: String, +} + +async fn handle_media_stream( + ws: WebSocketUpgrade, + State(register_tx): State>, +) -> impl IntoResponse { + ws.on_upgrade(move |socket| run_media_stream(socket, register_tx)) +} + +async fn run_media_stream( + mut socket: WebSocket, + register_tx: mpsc::Sender, +) { + match socket.recv().await { + Some(Ok(Message::Text(txt))) => match serde_json::from_str::(&txt) { + Ok(TwilioMediaEvent::Connected) => { + debug!("twilio media streams: connected handshake received") + } + Ok(other) => warn!(?other, "expected 'connected' envelope as first frame"), + Err(e) => { + warn!(error = ?e, "first WS frame not a Twilio Media Streams envelope"); + let _ = socket.close().await; + return; + } + }, + _ => { + warn!("twilio media streams: socket sent no connected handshake; closing"); + let _ = socket.close().await; + return; + } + } + + let (call_sid, stream_sid): (String, String) = match socket.recv().await { + Some(Ok(Message::Text(txt))) => match serde_json::from_str::(&txt) { + Ok(TwilioMediaEvent::Start { start }) => (start.call_sid, start.stream_sid), + Ok(other) => { + warn!(?other, "expected 'start' envelope as second frame"); + let _ = socket.close().await; + return; + } + Err(e) => { + warn!(error = ?e, "second WS frame not a Twilio 'start' envelope"); + let _ = socket.close().await; + return; + } + }, + _ => { + warn!("twilio media streams: socket sent no start frame; closing"); + let _ = socket.close().await; + return; + } + }; + + let (inbound_tx, inbound_rx) = mpsc::channel::(MEDIA_CHANNEL_CAPACITY); + let (outbound_tx, mut outbound_rx) = mpsc::channel::(MEDIA_CHANNEL_CAPACITY); + + let (reply_tx, reply_rx) = oneshot::channel::(); + let register = RegisterTrunkInboundChannel { + call_sid: call_sid.clone(), + inbound_from_twilio_rx: inbound_rx, + outbound_to_twilio_tx: outbound_tx, + tap_url: url::Url::parse("ws://127.0.0.1:8082/echo") + .expect("hardcoded loopback URL is always valid"), + reply: reply_tx, + }; + if register_tx.send(register).await.is_err() { + warn!(%call_sid, "media-thread register channel closed; closing WSS"); + let _ = socket.close().await; + return; + } + let channel_id = match reply_rx.await { + Ok(id) => id, + Err(_) => { + warn!(%call_sid, "media-thread dropped RegisterTrunk reply; closing WSS"); + let _ = socket.close().await; + return; + } + }; + info!(%channel_id, %call_sid, %stream_sid, "twilio media streams: registered, audio loop active"); + + loop { + tokio::select! { + biased; + + msg = socket.recv() => match msg { + Some(Ok(Message::Text(txt))) => match serde_json::from_str::(&txt) { + Ok(TwilioMediaEvent::Media { media }) => { + if let Some(frame) = decode_media_payload(&media.payload) { + if inbound_tx.try_send(frame).is_err() { + warn!(%call_sid, "inbound mpsc full; dropping caller PCM frame"); + } + } + } + Ok(TwilioMediaEvent::Stop) => { + info!(%call_sid, "twilio: stop frame; ending pump"); + break; + } + Ok(TwilioMediaEvent::Connected) => { + debug!(%call_sid, "twilio sent a duplicate 'connected' frame; ignoring"); + } + Ok(TwilioMediaEvent::Start { .. }) => { + debug!(%call_sid, "twilio sent a duplicate 'start' frame; ignoring"); + } + Err(e) => { + warn!(error = ?e, %call_sid, "twilio media streams JSON parse failed; skipping frame"); + } + }, + Some(Ok(Message::Close(_))) => { + info!(%call_sid, "twilio: WS closed by peer"); + break; + } + Some(Ok(Message::Ping(_))) | Some(Ok(Message::Pong(_))) => {} + Some(Ok(Message::Binary(_))) => { + warn!(%call_sid, "twilio: unexpected binary frame on a text-only protocol; dropping"); + } + Some(Err(e)) => { + warn!(error = ?e, %call_sid, "twilio WS recv error; ending pump"); + break; + } + None => { + info!(%call_sid, "twilio: WS stream ended by peer"); + break; + } + }, + + frame = outbound_rx.recv() => match frame { + Some(reply_frame) => { + let payload = encode_media_payload(&reply_frame); + let envelope = OutboundMediaFrame { + event: "media", + stream_sid: &stream_sid, + media: OutboundMediaPayload { payload }, + }; + match serde_json::to_string(&envelope) { + Ok(json) => { + if socket.send(Message::Text(json)).await.is_err() { + warn!(%call_sid, "twilio WS send (outbound) failed; ending pump"); + break; + } + } + Err(e) => { + warn!(error = ?e, %call_sid, "outbound media frame JSON serialization failed; dropping frame"); + } + } + } + None => { + info!(%call_sid, "outbound_to_twilio_rx closed; ending pump"); + break; + } + }, + } + } + + let _ = socket.close().await; + debug!(%call_sid, "twilio media streams: pump task exited"); +} + +fn decode_media_payload(base64_payload: &str) -> Option { + if base64_payload.is_empty() { + debug!("twilio media streams: empty media payload; dropping frame"); + return None; + } + let mulaw_bytes = base64::engine::general_purpose::STANDARD + .decode(base64_payload) + .map_err(|e| warn!(error = ?e, "twilio media streams: base64 decode failed")) + .ok()?; + Some(G711Codec::decode_mulaw_to_pcm(&mulaw_bytes)) +} + +fn encode_media_payload(frame: &PcmFrame) -> String { + let mulaw_bytes = G711Codec::encode_pcm_to_mulaw(frame); + base64::engine::general_purpose::STANDARD.encode(&mulaw_bytes) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn media_event_parses_connected_envelope() { + let s = r#"{"event":"connected","protocol":"twilio-media-stream","version":"1.0.0"}"#; + let e: TwilioMediaEvent = serde_json::from_str(s).unwrap(); + assert!(matches!(e, TwilioMediaEvent::Connected)); + } + + #[test] + fn media_event_parses_start_envelope() { + let s = r#"{ + "event":"start","sequenceNumber":1, + "start":{"streamSid":"MZdeadbeef","callSid":"CAdeadbeef","accountSid":"ACdeadbeef","tracks":["inbound","outbound"]} + }"#; + let e: TwilioMediaEvent = serde_json::from_str(s).unwrap(); + match e { + TwilioMediaEvent::Start { start } => { + assert_eq!(start.stream_sid, "MZdeadbeef"); + assert_eq!(start.call_sid, "CAdeadbeef"); + } + other => panic!("expected Start, got {other:?}"), + } + } + + #[test] + fn media_event_parses_media_envelope() { + let s = r#"{"event":"media","sequenceNumber":2,"media":{"track":"outbound","chunk":"1","timestamp":"2023","payload":"//MA"}}"#; + let e: TwilioMediaEvent = serde_json::from_str(s).unwrap(); + match e { + TwilioMediaEvent::Media { media } => assert_eq!(media.payload, "//MA"), + other => panic!("expected Media, got {other:?}"), + } + } + + #[test] + fn media_event_parses_stop_envelope() { + let s = r#"{"event":"stop","sequenceNumber":3,"stop":{}}"#; + let e: TwilioMediaEvent = serde_json::from_str(s).unwrap(); + assert!(matches!(e, TwilioMediaEvent::Stop)); + } + + #[test] + fn media_event_forwards_unknown_protocol_fields() { + let s = r#"{"event":"stop","stop":{},"futureField":"whatever"}"#; + let e: TwilioMediaEvent = serde_json::from_str(s).unwrap(); + assert!(matches!(e, TwilioMediaEvent::Stop)); + } + + #[test] + fn decode_media_payload_round_trips_a_loud_signal() { + let mut input_frame = PcmFrame::zeroed(); + for s in &mut input_frame.samples { + *s = 10_000; + } + let encoded = encode_media_payload(&input_frame); + let decoded = decode_media_payload(&encoded).expect("decode OK"); + assert_eq!(decoded.samples.len(), 480); + + let orig_rms = rms(&input_frame.samples); + let dec_rms = rms(&decoded.samples); + let drift = (dec_rms - orig_rms).abs() / orig_rms.max(1.0); + assert!( + drift <= 0.12, + "WS-pump decode energy drift = {:.2}% (budget 12%, spec ยง6.5)", + drift * 100.0 + ); + } + + #[test] + fn decode_media_payload_returns_none_on_invalid_base64() { + assert!(decode_media_payload("!!!not-valid-base64!!!").is_none()); + } + + #[test] + fn decode_media_payload_returns_none_on_empty_payload() { + assert!(decode_media_payload("").is_none()); + } + + #[test] + fn encode_media_payload_round_trips_negative_amplitude_without_polarity_flip() { + let mut frame = PcmFrame::zeroed(); + for s in &mut frame.samples { + *s = -5_000; + } + let encoded = encode_media_payload(&frame); + let decoded = decode_media_payload(&encoded).expect("decode OK"); + assert!( + decoded.samples.iter().all(|&s| s <= 0), + "negative PCM input must round-trip without polarity flip" + ); + } + + #[tokio::test] + async fn router_constructs_with_register_channel() { + let (register_tx, _register_rx) = mpsc::channel::(4); + let _app = TwilioMediaStreamsServer::router(register_tx); + } + + fn rms(samples: &[i16]) -> f64 { + let sum_sq: u64 = samples + .iter() + .map(|&s| (s as i64 * s as i64) as u64) + .sum(); + (sum_sq as f64 / samples.len() as f64).sqrt() + } +}