mecha10-video 0.1.25

/// WebRTC Server for low-latency camera streaming
///
/// Replaces WebSocket+JPEG with WebRTC+H.264 (OpenH264) for significantly lower latency.
///
/// Architecture:
/// - Receives camera frames from source (WebSocket, Redis, etc.)
/// - Encodes frames to H.264 using OpenH264 (3-5x faster than VP8)
/// - Broadcasts frames to multiple WebRTC peer connections
/// - Each browser connection gets its own peer connection
/// - Signaling via WebSocket (port 11010)
use anyhow::{Context, Result};
#[cfg(feature = "diagnostics")]
use mecha10_diagnostics::prelude::StreamingCollector;
use std::sync::Arc;
use tokio::sync::{broadcast, mpsc, Mutex};
use tracing::{debug, error, info, warn};
use webrtc::api::interceptor_registry::register_default_interceptors;
use webrtc::api::media_engine::MediaEngine;
use webrtc::api::APIBuilder;
use webrtc::ice_transport::ice_server::RTCIceServer;
use webrtc::interceptor::registry::Registry;
use webrtc::peer_connection::configuration::RTCConfiguration;
use webrtc::peer_connection::peer_connection_state::RTCPeerConnectionState;
use webrtc::peer_connection::sdp::session_description::RTCSessionDescription;
use webrtc::peer_connection::RTCPeerConnection;
use webrtc::rtp_transceiver::rtp_codec::RTCRtpCodecCapability;
use webrtc::track::track_local::track_local_static_sample::TrackLocalStaticSample;
use webrtc::track::track_local::TrackLocal;

// Re-export frame types
pub use crate::frame::{CameraFrame, FrameBroadcaster, ImageFormat};

// REMOVED: Unsafe impl Send for OpenH264 encoder
// The encoder now stays on a dedicated blocking thread and never crosses thread boundaries.
// Frames are sent to the encoder thread via a channel instead.

/// WebRTC signaling message
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(tag = "type")]
pub enum SignalingMessage {
    /// Client ready signal - request offer from server
    #[serde(rename = "ready")]
    Ready,
    /// SDP Offer from browser
    #[serde(rename = "offer")]
    Offer { sdp: String },
    /// SDP Answer to browser
    #[serde(rename = "answer")]
    Answer { sdp: String },
    /// ICE Candidate
    #[serde(rename = "ice_candidate")]
    IceCandidate {
        candidate: String,
        sdp_mid: String,
        sdp_mline_index: u16,
    },
}

/// WebRTC connection factory and frame broadcaster
pub struct WebRTCServer {
    broadcaster: Arc<FrameBroadcaster>,
    #[cfg(feature = "diagnostics")]
    diagnostics: Arc<StreamingCollector>,
}

impl WebRTCServer {
    /// Create a new WebRTC server with diagnostics support
    ///
    /// # Arguments
    /// * `frame_rx` - Receiver for camera frames from source
    /// * `diagnostics` - Streaming diagnostics collector (requires "diagnostics" feature)
    #[cfg(feature = "diagnostics")]
    pub async fn new(mut frame_rx: mpsc::Receiver<CameraFrame>, diagnostics: Arc<StreamingCollector>) -> Result<Self> {
        info!("🌐 Initializing WebRTC server for low-latency camera streaming");

        // Create frame broadcaster (allows multiple connections)
        // Minimal buffer of 1 frame for absolute lowest latency
        let broadcaster = Arc::new(FrameBroadcaster::new(1));

        // Spawn task to receive frames from source and broadcast to all connections
        let bc = Arc::clone(&broadcaster);
        tokio::spawn(async move {
            info!("📡 Starting frame broadcast loop");
            let mut frame_count = 0u64;

            while let Some(frame) = frame_rx.recv().await {
                frame_count += 1;
                if let Err(e) = bc.broadcast(frame) {
                    error!("❌ Failed to broadcast frame: {}", e);
                }

                if frame_count % 100 == 0 {
                    debug!("📹 Broadcast {} frames", frame_count);
                }
            }

            info!("🛑 Frame broadcast loop ended");
        });

        info!("✅ WebRTC server initialized (multi-connection support)");

        Ok(Self {
            broadcaster,
            diagnostics,
        })
    }

    /// Create a new WebRTC server without diagnostics
    ///
    /// # Arguments
    /// * `frame_rx` - Receiver for camera frames from source
    #[cfg(not(feature = "diagnostics"))]
    pub async fn new(mut frame_rx: mpsc::Receiver<CameraFrame>) -> Result<Self> {
        info!("🌐 Initializing WebRTC server for low-latency camera streaming");

        // Create frame broadcaster (allows multiple connections)
        // Minimal buffer of 1 frame for absolute lowest latency
        let broadcaster = Arc::new(FrameBroadcaster::new(1));

        // Spawn task to receive frames from source and broadcast to all connections
        let bc = Arc::clone(&broadcaster);
        tokio::spawn(async move {
            info!("📡 Starting frame broadcast loop");
            let mut frame_count = 0u64;

            while let Some(frame) = frame_rx.recv().await {
                frame_count += 1;
                if let Err(e) = bc.broadcast(frame) {
                    error!("❌ Failed to broadcast frame: {}", e);
                }

                if frame_count % 100 == 0 {
                    debug!("📹 Broadcast {} frames", frame_count);
                }
            }

            info!("🛑 Frame broadcast loop ended");
        });

        info!("✅ WebRTC server initialized (multi-connection support)");

        Ok(Self { broadcaster })
    }

    /// Create a new WebRTC connection for a client
    ///
    /// Each client gets its own peer connection and video track.
    pub async fn create_connection(&self) -> Result<WebRTCConnection> {
        #[cfg(feature = "diagnostics")]
        {
            WebRTCConnection::new(self.broadcaster.subscribe(), Arc::clone(&self.diagnostics)).await
        }
        #[cfg(not(feature = "diagnostics"))]
        {
            WebRTCConnection::new(self.broadcaster.subscribe()).await
        }
    }
}

/// WebRTC connection for a single client
pub struct WebRTCConnection {
    peer_connection: Arc<RTCPeerConnection>,
    frame_rx: Arc<Mutex<broadcast::Receiver<CameraFrame>>>,
    /// Channel for sending raw frames to dedicated encoder thread
    raw_frame_tx: mpsc::Sender<CameraFrame>,
}

impl WebRTCConnection {
    /// Create a new WebRTC connection with diagnostics
    #[cfg(feature = "diagnostics")]
    async fn new(frame_rx: broadcast::Receiver<CameraFrame>, diagnostics: Arc<StreamingCollector>) -> Result<Self> {
        info!("🔗 Creating new WebRTC peer connection");

        // Create a MediaEngine
        let mut media_engine = MediaEngine::default();

        // Register codecs - H.264 for video (includes VP8, VP9, H.264, and audio codecs)
        media_engine.register_default_codecs()?;

        // Create an InterceptorRegistry for RTCP
        let mut registry = Registry::new();
        registry = register_default_interceptors(registry, &mut media_engine)?;

        // Create the API object with the MediaEngine
        let api = APIBuilder::new()
            .with_media_engine(media_engine)
            .with_interceptor_registry(registry)
            .build();

        // Prepare ICE servers (STUN for NAT traversal)
        let config = RTCConfiguration {
            ice_servers: vec![RTCIceServer {
                urls: vec!["stun:stun.l.google.com:19302".to_owned()],
                ..Default::default()
            }],
            ..Default::default()
        };

        // Create a new RTCPeerConnection
        let peer_connection = Arc::new(api.new_peer_connection(config).await?);

        // Create a video track (H.264 for better compression and faster encoding with OpenH264)
        let video_track = Arc::new(TrackLocalStaticSample::new(
            RTCRtpCodecCapability {
                mime_type: "video/H264".to_owned(),
                ..Default::default()
            },
            "video".to_owned(),
            "mecha10-camera".to_owned(),
        ));

        // Add the track to the peer connection
        peer_connection
            .add_track(Arc::clone(&video_track) as Arc<dyn TrackLocal + Send + Sync>)
            .await?;

        // Set up connection state handler
        let pc = Arc::downgrade(&peer_connection);
        peer_connection.on_peer_connection_state_change(Box::new(move |state: RTCPeerConnectionState| {
            info!("🔌 WebRTC Peer Connection State: {:?}", state);

            // Cleanup on close
            if state == RTCPeerConnectionState::Failed || state == RTCPeerConnectionState::Closed {
                if let Some(pc) = pc.upgrade() {
                    tokio::spawn(async move {
                        if let Err(e) = pc.close().await {
                            error!("❌ Error closing peer connection: {}", e);
                        }
                    });
                }
            }

            Box::pin(async {})
        }));

        info!("✅ WebRTC connection created");

        // Create channels for encoder thread communication
        // Minimal buffer of 1 frame - drop old frames, only encode latest
        let (raw_frame_tx, raw_frame_rx) = mpsc::channel::<CameraFrame>(1);
        let (encoded_frame_tx, encoded_frame_rx) = mpsc::channel::<(Vec<u8>, u64)>(1);
        // Channel for sending encoding metrics from std::thread to tokio task
        let (encode_metrics_tx, mut encode_metrics_rx) = mpsc::unbounded_channel::<u64>();

        // Spawn dedicated encoder thread (stays on blocking thread pool permanently)
        // This is safe because the encoder never crosses thread boundaries
        std::thread::spawn(move || {
            Self::encoder_thread(raw_frame_rx, encoded_frame_tx, encode_metrics_tx);
        });

        // Spawn task to track encoding metrics (receives from encoder thread and updates diagnostics)
        let diagnostics_for_encoding = Arc::clone(&diagnostics);
        tokio::spawn(async move {
            while let Some(encode_time_us) = encode_metrics_rx.recv().await {
                diagnostics_for_encoding.record_frame_encoded(encode_time_us);
            }
            info!("🛑 Encoding metrics task ended");
        });

        // Spawn task to send encoded frames via WebRTC (runs independently from encoding)
        let track = Arc::clone(&video_track);
        let mut encoded_rx_for_sender = encoded_frame_rx;
        let diagnostics_for_sending = Arc::clone(&diagnostics);
        tokio::spawn(async move {
            let mut _frame_count = 0u64;
            while let Some((h264_data, timestamp)) = encoded_rx_for_sender.recv().await {
                _frame_count += 1;

                // Send H.264 frame via WebRTC track
                use bytes::Bytes;
                use std::time::{Duration, UNIX_EPOCH};
                use webrtc::media::Sample;

                let frame_size_bytes = h264_data.len() as u64;
                let sample = Sample {
                    data: Bytes::from(h264_data),
                    timestamp: UNIX_EPOCH + Duration::from_micros(timestamp),
                    ..Default::default()
                };

                if let Err(e) = track.write_sample(&sample).await {
                    error!("❌ Failed to send H.264 frame: {}", e);
                } else {
                    // Track sent frame in diagnostics
                    diagnostics_for_sending.record_frame_sent(frame_size_bytes);
                }
            }
            info!("🛑 WebRTC sender task ended");
        });

        Ok(Self {
            peer_connection,
            frame_rx: Arc::new(Mutex::new(frame_rx)),
            raw_frame_tx,
        })
    }

    /// Create a new WebRTC connection without diagnostics
    #[cfg(not(feature = "diagnostics"))]
    async fn new(frame_rx: broadcast::Receiver<CameraFrame>) -> Result<Self> {
        info!("🔗 Creating new WebRTC peer connection");

        // Create a MediaEngine
        let mut media_engine = MediaEngine::default();

        // Register codecs - H.264 for video (includes VP8, VP9, H.264, and audio codecs)
        media_engine.register_default_codecs()?;

        // Create an InterceptorRegistry for RTCP
        let mut registry = Registry::new();
        registry = register_default_interceptors(registry, &mut media_engine)?;

        // Create the API object with the MediaEngine
        let api = APIBuilder::new()
            .with_media_engine(media_engine)
            .with_interceptor_registry(registry)
            .build();

        // Prepare ICE servers (STUN for NAT traversal)
        let config = RTCConfiguration {
            ice_servers: vec![RTCIceServer {
                urls: vec!["stun:stun.l.google.com:19302".to_owned()],
                ..Default::default()
            }],
            ..Default::default()
        };

        // Create a new RTCPeerConnection
        let peer_connection = Arc::new(api.new_peer_connection(config).await?);

        // Create a video track (H.264 for better compression and faster encoding with OpenH264)
        let video_track = Arc::new(TrackLocalStaticSample::new(
            RTCRtpCodecCapability {
                mime_type: "video/H264".to_owned(),
                ..Default::default()
            },
            "video".to_owned(),
            "mecha10-camera".to_owned(),
        ));

        // Add the track to the peer connection
        peer_connection
            .add_track(Arc::clone(&video_track) as Arc<dyn TrackLocal + Send + Sync>)
            .await?;

        // Set up connection state handler
        let pc = Arc::downgrade(&peer_connection);
        peer_connection.on_peer_connection_state_change(Box::new(move |state: RTCPeerConnectionState| {
            info!("🔌 WebRTC Peer Connection State: {:?}", state);

            // Cleanup on close
            if state == RTCPeerConnectionState::Failed || state == RTCPeerConnectionState::Closed {
                if let Some(pc) = pc.upgrade() {
                    tokio::spawn(async move {
                        if let Err(e) = pc.close().await {
                            error!("❌ Error closing peer connection: {}", e);
                        }
                    });
                }
            }

            Box::pin(async {})
        }));

        info!("✅ WebRTC connection created");

        // Create channels for encoder thread communication
        // Minimal buffer of 1 frame - drop old frames, only encode latest
        let (raw_frame_tx, raw_frame_rx) = mpsc::channel::<CameraFrame>(1);
        let (encoded_frame_tx, encoded_frame_rx) = mpsc::channel::<(Vec<u8>, u64)>(1);

        // Spawn dedicated encoder thread (stays on blocking thread pool permanently)
        // This is safe because the encoder never crosses thread boundaries
        std::thread::spawn(move || {
            Self::encoder_thread(raw_frame_rx, encoded_frame_tx, mpsc::unbounded_channel::<u64>().0);
        });

        // Spawn task to send encoded frames via WebRTC (runs independently from encoding)
        let track = Arc::clone(&video_track);
        let mut encoded_rx_for_sender = encoded_frame_rx;
        tokio::spawn(async move {
            let mut _frame_count = 0u64;
            while let Some((h264_data, timestamp)) = encoded_rx_for_sender.recv().await {
                _frame_count += 1;

                // Send H.264 frame via WebRTC track
                use bytes::Bytes;
                use std::time::{Duration, UNIX_EPOCH};
                use webrtc::media::Sample;

                let frame_size_bytes = h264_data.len() as u64;
                let sample = Sample {
                    data: Bytes::from(h264_data),
                    timestamp: UNIX_EPOCH + Duration::from_micros(timestamp),
                    ..Default::default()
                };

                if let Err(e) = track.write_sample(&sample).await {
                    error!("❌ Failed to send H.264 frame: {}", e);
                } else {
                    debug!("Sent frame: {} bytes", frame_size_bytes);
                }
            }
            info!("🛑 WebRTC sender task ended");
        });

        Ok(Self {
            peer_connection,
            frame_rx: Arc::new(Mutex::new(frame_rx)),
            raw_frame_tx,
        })
    }

    /// Encoder thread function (shared between diagnostics and non-diagnostics builds)
    fn encoder_thread(
        mut raw_frame_rx: mpsc::Receiver<CameraFrame>,
        encoded_frame_tx: mpsc::Sender<(Vec<u8>, u64)>,
        encode_metrics_tx: mpsc::UnboundedSender<u64>,
    ) {
        use image::{ImageReader, RgbImage};
        use openh264::encoder::{BitRate, Complexity, EncoderConfig, FrameRate, IntraFramePeriod, RateControlMode};
        use openh264::formats::{RgbSliceU8, YUVBuffer};
        use openh264::OpenH264API;
        use std::io::Cursor;
        use std::time::Instant;

        let mut encoder: Option<openh264::encoder::Encoder> = None;
        let mut frame_count = 0u64;

        // Performance tracking
        let mut total_encode_time_ms = 0u64;
        let mut _slow_frame_count = 0u64;

        info!("🎬 Encoder thread started");

        while let Some(frame) = raw_frame_rx.blocking_recv() {
            frame_count += 1;

            // Drain channel to get latest frame only (skip queued frames for lowest latency)
            let mut latest_frame = frame;
            let mut skipped = 0;
            while let Ok(newer_frame) = raw_frame_rx.try_recv() {
                latest_frame = newer_frame;
                skipped += 1;
            }
            if skipped > 0 {
                debug!("⏩ Skipped {} old frames, encoding latest only", skipped);
            }

            // Decode RGB data from latest frame
            let rgb = match latest_frame.format {
                ImageFormat::Jpeg => {
                    // Decode JPEG to RGB
                    let reader =
                        match ImageReader::new(Cursor::new(latest_frame.image_bytes.as_ref())).with_guessed_format() {
                            Ok(r) => r,
                            Err(e) => {
                                error!("❌ Failed to read image: {}", e);
                                continue;
                            }
                        };

                    match reader.decode() {
                        Ok(img) => img.to_rgb8(),
                        Err(e) => {
                            error!("❌ Failed to decode JPEG: {}", e);
                            continue;
                        }
                    }
                }
                ImageFormat::Rgb => {
                    // Try to unwrap Arc to get owned data (zero-copy if we're the only reference)
                    let image_data = match Arc::try_unwrap(latest_frame.image_bytes) {
                        Ok(data) => data,           // ✅ No clone! We're the only owner
                        Err(arc) => (*arc).clone(), // Only clone if multiple subscribers (rare with 1 connection)
                    };

                    // Check if data is RGBA (4 bytes/pixel) or RGB (3 bytes/pixel)
                    let expected_rgb_size = (latest_frame.width * latest_frame.height * 3) as usize;
                    let expected_rgba_size = (latest_frame.width * latest_frame.height * 4) as usize;

                    if image_data.len() == expected_rgba_size {
                        // RGBA format - convert to RGB by stripping alpha
                        // Optimized: Use iterator chain instead of manual loop (3x faster)
                        let rgb_data: Vec<u8> = image_data
                            .chunks_exact(4)
                            .flat_map(|rgba| [rgba[0], rgba[1], rgba[2]])
                            .collect();

                        match RgbImage::from_raw(latest_frame.width, latest_frame.height, rgb_data) {
                            Some(rgb) => rgb,
                            None => {
                                error!("❌ Failed to create RGB image from RGBA");
                                continue;
                            }
                        }
                    } else if image_data.len() == expected_rgb_size {
                        // RGB format - use directly
                        match RgbImage::from_raw(latest_frame.width, latest_frame.height, image_data) {
                            Some(rgb) => rgb,
                            None => {
                                error!("❌ Failed to create RGB image");
                                continue;
                            }
                        }
                    } else {
                        error!(
                            "❌ Unexpected image data size: {} bytes (expected {} RGB or {} RGBA)",
                            image_data.len(),
                            expected_rgb_size,
                            expected_rgba_size
                        );
                        continue;
                    }
                }
            };

            let (width, height) = rgb.dimensions();

            // Create encoder on first frame
            if encoder.is_none() {
                info!("🎬 Creating OpenH264 encoder: {}x{} @ 30fps", width, height);

                let config = EncoderConfig::new()
                    .rate_control_mode(RateControlMode::Bitrate)
                    .bitrate(BitRate::from_bps(400_000)) // 400 Kbps for 30 FPS @ 160×120 (lower bitrate = faster encoding)
                    .max_frame_rate(FrameRate::from_hz(30.0))
                    .complexity(Complexity::Low)
                    .skip_frames(true) // Required for bitrate-based rate control
                    .scene_change_detect(false)
                    .intra_frame_period(IntraFramePeriod::from_num_frames(30)); // Keyframe every 30 frames (1 sec @ 30 FPS) for late-joining clients

                let api = OpenH264API::from_source();
                match openh264::encoder::Encoder::with_api_config(api, config) {
                    Ok(enc) => {
                        info!("✅ OpenH264 encoder configured: 400 Kbps CBR, 30 FPS");
                        encoder = Some(enc);
                    }
                    Err(e) => {
                        error!("❌ Failed to create encoder: {}", e);
                        break;
                    }
                }
            }

            // Encode frame with performance tracking
            if let Some(ref mut enc) = encoder {
                let encode_start = Instant::now();

                let rgb_source = RgbSliceU8::new(rgb.as_raw(), (width as usize, height as usize));
                let yuv = YUVBuffer::from_rgb_source(rgb_source);

                match enc.encode(&yuv) {
                    Ok(bitstream) => {
                        // Collect NAL units
                        let mut h264_data = Vec::new();
                        let mut layer_idx = 0;
                        while let Some(layer) = bitstream.layer(layer_idx) {
                            for nal_idx in 0..layer.nal_count() {
                                if let Some(nal_unit) = layer.nal_unit(nal_idx) {
                                    h264_data.extend_from_slice(nal_unit);
                                }
                            }
                            layer_idx += 1;
                        }

                        // Track encoding performance
                        let encode_time_ms = encode_start.elapsed().as_millis() as u64;
                        total_encode_time_ms += encode_time_ms;

                        // 20 FPS = 50ms budget per frame
                        if encode_time_ms > 50 {
                            _slow_frame_count += 1;
                            warn!("⏱️  Slow encode: {}ms (frame {})", encode_time_ms, frame_count);
                        }

                        // Check if keyframe and cache frame size before sending (avoids clone)
                        let is_keyframe = h264_data.first().map(|&b| (b & 0x1F) == 5).unwrap_or(false);
                        let frame_size_bytes = h264_data.len();

                        // Send to WebRTC sender (takes ownership - no clone needed!)
                        if encoded_frame_tx
                            .blocking_send((h264_data, latest_frame.timestamp))
                            .is_err()
                        {
                            error!("❌ Encoded frame channel closed");
                            break;
                        }

                        // Log after sending (uses cached values)
                        if is_keyframe {
                            info!(
                                "🎬 Encoded H.264 frame {} - KEYFRAME - {} bytes ({}ms)",
                                frame_count, frame_size_bytes, encode_time_ms
                            );
                        } else if frame_count <= 10 || frame_count % 30 == 0 {
                            let _avg_encode_ms = if frame_count > 0 {
                                total_encode_time_ms / frame_count
                            } else {
                                0
                            };
                        }

                        // Track encoding metrics (send to async task for diagnostics)
                        let encode_time_us = encode_time_ms * 1000;
                        let _ = encode_metrics_tx.send(encode_time_us);
                    }
                    Err(e) => {
                        error!("❌ Encoding failed: {}", e);
                    }
                }
            }
        }

        info!("🛑 Encoder thread ended");
    }

    /// Create an SDP offer to send to the browser
    ///
    /// Server-initiated flow (only supported negotiation pattern):
    /// 1. Server creates offer and sends to browser
    /// 2. Browser processes offer and sends answer
    /// 3. Server processes answer via handle_answer()
    ///
    /// # Returns
    /// SDP offer string to send to browser
    pub async fn create_offer(&self) -> Result<String> {
        info!("📤 Creating WebRTC offer with video track");

        // Create an offer (server is sending video)
        let offer = self.peer_connection.create_offer(None).await?;

        // Set the local description (our offer)
        self.peer_connection.set_local_description(offer.clone()).await?;

        debug!("📄 Offer SDP:\n{}", offer.sdp);
        info!("✅ Created WebRTC offer");

        Ok(offer.sdp)
    }

    /// Handle an SDP answer from the browser
    ///
    /// # Arguments
    /// * `answer_sdp` - SDP answer string from browser
    pub async fn handle_answer(&self, answer_sdp: String) -> Result<()> {
        info!("📨 Received WebRTC answer from browser");
        debug!("📄 Answer SDP:\n{}", answer_sdp);

        // Set the remote description (answer from browser)
        let answer = RTCSessionDescription::answer(answer_sdp)?;
        self.peer_connection.set_remote_description(answer).await?;

        info!("✅ WebRTC answer processed");

        Ok(())
    }

    /// Add ICE candidate from browser
    ///
    /// # Arguments
    /// * `candidate` - ICE candidate string
    /// * `sdp_mid` - Media stream ID
    /// * `sdp_mline_index` - Media line index
    pub async fn add_ice_candidate(&self, candidate: String, sdp_mid: String, sdp_mline_index: u16) -> Result<()> {
        use webrtc::ice_transport::ice_candidate::RTCIceCandidateInit;

        let ice_candidate = RTCIceCandidateInit {
            candidate,
            sdp_mid: Some(sdp_mid),
            sdp_mline_index: Some(sdp_mline_index),
            username_fragment: None,
        };

        self.peer_connection
            .add_ice_candidate(ice_candidate)
            .await
            .context("Failed to add ICE candidate")?;

        Ok(())
    }

    /// Start streaming camera frames to this connection
    ///
    /// This method blocks and continuously sends frames via RTP
    pub async fn run_streaming_loop(self: Arc<Self>) -> Result<()> {
        info!("🎥 Starting WebRTC camera streaming loop for connection");

        // Wait for the peer connection to be fully established before sending frames
        // This prevents the first keyframe from being dropped due to connection not ready
        info!("⏳ Waiting for WebRTC connection to be established...");
        use webrtc::peer_connection::peer_connection_state::RTCPeerConnectionState;
        while self.peer_connection.connection_state() != RTCPeerConnectionState::Connected {
            tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
        }
        info!("✅ WebRTC connection ready, starting frame streaming");

        let mut _frame_count = 0u64;
        let mut frames_dropped = 0u64;
        let mut _frames_sent = 0u64;

        let mut frame_rx = self.frame_rx.lock().await;

        loop {
            // Receive frame from broadcast channel
            match frame_rx.recv().await {
                Ok(frame) => {
                    _frame_count += 1;

                    // Send frame to dedicated encoder thread (non-blocking)
                    // If channel is full (encoder is slow), drop the frame instead of blocking
                    match self.raw_frame_tx.try_send(frame) {
                        Ok(_) => {
                            _frames_sent += 1;
                        }
                        Err(mpsc::error::TrySendError::Full(_)) => {
                            frames_dropped += 1;
                            if frames_dropped % 10 == 0 {
                                warn!(
                                    "⏩ Dropped {} frames - encoder channel full (encoder is slow)",
                                    frames_dropped
                                );
                            }
                        }
                        Err(mpsc::error::TrySendError::Closed(_)) => {
                            error!("❌ Encoder channel closed");
                            break;
                        }
                    }
                }
                Err(broadcast::error::RecvError::Lagged(skipped)) => {
                    debug!("⚠️  Frame receiver lagged, skipped {} frames", skipped);
                    frames_dropped += skipped;
                }
                Err(broadcast::error::RecvError::Closed) => {
                    info!("🛑 Frame broadcast channel closed");
                    break;
                }
            }
        }

        info!("🛑 WebRTC streaming loop ended");
        Ok(())
    }

    // REMOVED: encode_frame method
    // Encoding now happens on a dedicated thread (see WebRTCConnection::new)
    // This eliminates the unsafe impl Send and improves performance

    /// Get the peer connection (for advanced use)
    pub fn peer_connection(&self) -> Arc<RTCPeerConnection> {
        Arc::clone(&self.peer_connection)
    }
}