feat(proxy-engine,webrtc): add B2BUA SIP leg handling and WebRTC call bridging for outbound calls

rust/crates/proxy-engine/src/webrtc_engine.rs

@@ -29,8 +29,10 @@ pub struct SipBridgeInfo {
     pub provider_media: SocketAddr,
     /// Provider's codec payload type (e.g. 9 for G.722).
     pub sip_pt: u8,
-    /// The SIP UDP socket for sending RTP to the provider.
-    pub sip_socket: Arc<UdpSocket>,
+    /// The allocated RTP socket for bidirectional audio with the provider.
+    /// This is the socket whose port was advertised in SDP, so the provider
+    /// sends RTP here and expects RTP from this port.
+    pub rtp_socket: Arc<UdpSocket>,
 }
 
 /// A managed WebRTC session.
@@ -206,7 +208,10 @@ impl WebRtcEngine {
         Ok(answer_sdp)
     }
 
-    /// Link a WebRTC session to a SIP call — sets up the audio bridge.
+    /// Link a WebRTC session to a SIP call — sets up bidirectional audio bridge.
+    /// - Browser→SIP: already running via on_track handler, will start forwarding
+    ///   once bridge info is set.
+    /// - SIP→Browser: spawned here, reads from the RTP socket and sends to browser.
     pub async fn link_to_sip(
         &mut self,
         session_id: &str,
@@ -215,6 +220,18 @@ impl WebRtcEngine {
     ) -> bool {
         if let Some(session) = self.sessions.get_mut(session_id) {
             session.call_id = Some(call_id.to_string());
+
+            // Spawn SIP → browser audio loop (provider RTP → transcode → Opus → WebRTC track).
+            let local_track = session.local_track.clone();
+            let rtp_socket = bridge_info.rtp_socket.clone();
+            let sip_pt = bridge_info.sip_pt;
+            let out_tx = self.out_tx.clone();
+            let sid = session_id.to_string();
+            tokio::spawn(sip_to_browser_loop(
+                rtp_socket, local_track, sip_pt, out_tx, sid,
+            ));
+
+            // Set bridge info — this unblocks the browser→SIP loop (already running).
             let mut bridge = session.sip_bridge.lock().await;
             *bridge = Some(bridge_info);
             true
@@ -223,45 +240,6 @@ impl WebRtcEngine {
         }
     }
 
-    /// Send transcoded audio from the SIP side to the browser.
-    /// Called by the RTP relay when it receives a packet from the provider.
-    pub async fn forward_sip_to_browser(
-        &self,
-        session_id: &str,
-        sip_rtp_payload: &[u8],
-        sip_pt: u8,
-    ) -> Result<(), String> {
-        let session = self
-            .sessions
-            .get(session_id)
-            .ok_or_else(|| format!("session {session_id} not found"))?;
-
-        // Transcode SIP codec → Opus.
-        // We create a temporary TranscodeState per packet for simplicity.
-        // TODO: Use a per-session persistent state for proper codec continuity.
-        let mut transcoder = TranscodeState::new().map_err(|e| format!("codec: {e}"))?;
-        let opus_payload = transcoder
-            .transcode(sip_rtp_payload, sip_pt, PT_OPUS, Some("to_browser"))
-            .map_err(|e| format!("transcode: {e}"))?;
-
-        if opus_payload.is_empty() {
-            return Ok(());
-        }
-
-        // Build RTP header for Opus.
-        // TODO: Track seq/ts/ssrc per session for proper continuity.
-        let header = build_rtp_header(PT_OPUS, 0, 0, 0);
-        let mut packet = header.to_vec();
-        packet.extend_from_slice(&opus_payload);
-
-        session
-            .local_track
-            .write(&packet)
-            .await
-            .map(|_| ())
-            .map_err(|e| format!("write: {e}"))
-    }
-
     pub async fn add_ice_candidate(
         &self,
         session_id: &str,
@@ -365,9 +343,9 @@ async fn browser_to_sip_loop(
                 to_sip_seq = to_sip_seq.wrapping_add(1);
                 to_sip_ts = to_sip_ts.wrapping_add(rtp_clock_increment(bridge_info.sip_pt));
 
-                // Send to provider.
+                // Send to provider via the RTP socket (correct source port matching our SDP).
                 let _ = bridge_info
-                    .sip_socket
+                    .rtp_socket
                     .send_to(&sip_rtp, bridge_info.provider_media)
                     .await;
 
@@ -387,3 +365,86 @@ async fn browser_to_sip_loop(
         }
     }
 }
+
+/// SIP → Browser audio forwarding loop.
+/// Reads RTP from the provider (via the allocated RTP socket), transcodes to Opus,
+/// and writes to the WebRTC local track for delivery to the browser.
+async fn sip_to_browser_loop(
+    rtp_socket: Arc<UdpSocket>,
+    local_track: Arc<TrackLocalStaticRTP>,
+    sip_pt: u8,
+    out_tx: OutTx,
+    session_id: String,
+) {
+    let mut transcoder = match TranscodeState::new() {
+        Ok(t) => t,
+        Err(e) => {
+            emit_event(
+                &out_tx,
+                "webrtc_error",
+                serde_json::json!({
+                    "session_id": session_id,
+                    "error": format!("sip_to_browser codec init: {e}"),
+                }),
+            );
+            return;
+        }
+    };
+
+    let mut buf = vec![0u8; 1500];
+    let mut count = 0u64;
+    let mut seq: u16 = 0;
+    let mut ts: u32 = 0;
+    let ssrc: u32 = rand::random();
+
+    loop {
+        match rtp_socket.recv_from(&mut buf).await {
+            Ok((n, _from)) => {
+                if n < 12 {
+                    continue; // Too small for RTP header.
+                }
+                count += 1;
+
+                // Extract payload (skip 12-byte RTP header).
+                let payload = &buf[12..n];
+                if payload.is_empty() {
+                    continue;
+                }
+
+                // Transcode SIP codec → Opus.
+                let opus_payload = match transcoder.transcode(
+                    payload,
+                    sip_pt,
+                    PT_OPUS,
+                    Some("sip_to_browser"),
+                ) {
+                    Ok(p) if !p.is_empty() => p,
+                    _ => continue,
+                };
+
+                // Build Opus RTP packet.
+                let header = build_rtp_header(PT_OPUS, seq, ts, ssrc);
+                let mut packet = header.to_vec();
+                packet.extend_from_slice(&opus_payload);
+
+                seq = seq.wrapping_add(1);
+                ts = ts.wrapping_add(960); // Opus: 48000 Hz × 20ms = 960 samples
+
+                let _ = local_track.write(&packet).await;
+
+                if count == 1 || count == 50 || count % 500 == 0 {
+                    emit_event(
+                        &out_tx,
+                        "webrtc_audio_rx",
+                        serde_json::json!({
+                            "session_id": session_id,
+                            "direction": "sip_to_browser",
+                            "packet_count": count,
+                        }),
+                    );
+                }
+            }
+            Err(_) => break, // Socket closed.
+        }
+    }
+}