rvoip-rtp-core 0.2.3

//! Default implementation for the server transport
//!
//! This file contains the implementation of the MediaTransportServer trait
//! through the DefaultMediaTransportServer struct.

use std::collections::HashMap;
use std::net::{SocketAddr, UdpSocket};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;

use async_trait::async_trait;
use dashmap::DashMap;
use tokio::sync::{broadcast, Mutex, RwLock};
use tokio::task::JoinHandle;
use tracing::{debug, error, info, warn};
use uuid::Uuid;

use crate::api::client::transport::{RtcpStats, VoipMetrics};
use crate::api::common::config::{SecurityInfo, SecurityMode};
use crate::api::common::error::MediaTransportError;
use crate::api::common::events::MediaEventCallback;
use crate::api::common::extension::ExtensionFormat;
use crate::api::common::frame::MediaFrame;
use crate::api::common::stats::MediaStats;
use crate::api::server::config::ServerConfig;
use crate::api::server::security::ServerSecurityContext;
use crate::api::server::transport::{ClientInfo, HeaderExtension, MediaTransportServer};
use crate::srtp::crypto::SrtpCryptoKey;
use crate::srtp::{SrtpContext, SRTP_AES128_CM_SHA1_80};
use crate::transport::{RtpTransport, RtpTransportConfig, SecurityRtpTransport, UdpRtpTransport};
use crate::{CsrcManager, CsrcMapping, RtpCsrc, RtpSsrc};

use super::core::ClientConnection;

/// Default implementation of the MediaTransportServer
pub struct DefaultMediaTransportServer {
    /// Session identifier
    id: String,

    /// Configuration
    config: ServerConfig,

    /// Whether the server is running
    running: Arc<RwLock<bool>>,

    /// Main socket for the server
    main_socket: Arc<RwLock<Option<Arc<dyn RtpTransport>>>>,

    /// Socket
    socket: Arc<RwLock<Option<UdpSocket>>>,

    /// Main transport
    transport: Arc<RwLock<Option<Arc<dyn RtpTransport>>>>,

    /// Client sockets
    client_sockets: Arc<RwLock<HashMap<String, UdpSocket>>>,

    /// Client transports
    client_transports: Arc<RwLock<HashMap<String, Arc<dyn RtpTransport>>>>,

    /// Security context for DTLS/SRTP
    security_context: Arc<RwLock<Option<Arc<dyn ServerSecurityContext + Send + Sync>>>>,

    /// Client connections. `DashMap` so per-frame egress / RTCP /
    /// security helpers don't serialise behind a single async writer
    /// lock — see Phase C22. Every consumer reaches into one shard
    /// rather than blocking the whole map. The helper modules
    /// (`frame.rs`, `connection.rs`, `ssrc/demux.rs`, `rtcp/*`,
    /// `stats/*`) take this by reference and use `entry()` /
    /// `get()` / `get_mut()` directly; no shard guard is held
    /// across `.await`.
    clients: Arc<DashMap<String, ClientConnection>>,

    /// Transport event callbacks
    event_callbacks: Arc<RwLock<Vec<MediaEventCallback>>>,

    /// Client connected callbacks
    client_connected_callbacks: Arc<RwLock<Vec<Box<dyn Fn(ClientInfo) + Send + Sync>>>>,

    /// Client disconnected callbacks
    client_disconnected_callbacks: Arc<RwLock<Vec<Box<dyn Fn(ClientInfo) + Send + Sync>>>>,

    /// SSRC demultiplexing enabled flag. AtomicBool because it's read
    /// on every per-frame send (`send_frame_to`) and per-packet receive
    /// dispatch (event loop in `start()`); the RwLock added a
    /// scheduler dip per frame for a single boolean.
    ssrc_demultiplexing_enabled: Arc<AtomicBool>,

    /// CSRC management status. AtomicBool for the same reason as
    /// [`Self::ssrc_demultiplexing_enabled`] — read on every per-frame
    /// send + broadcast path.
    csrc_management_enabled: Arc<AtomicBool>,

    /// CSRC manager
    csrc_manager: Arc<parking_lot::RwLock<CsrcManager>>,

    /// Header extension format
    header_extension_format: Arc<RwLock<ExtensionFormat>>,

    /// Header extensions enabled flag
    header_extensions_enabled: Arc<RwLock<bool>>,

    /// Header extension ID-to-URI mappings
    header_extension_mappings: Arc<RwLock<HashMap<u8, String>>>,

    /// Pending header extensions to be added to outgoing packets
    pending_extensions: Arc<RwLock<HashMap<String, Vec<HeaderExtension>>>>,

    /// Header extensions received from clients
    received_extensions: Arc<RwLock<HashMap<String, Vec<HeaderExtension>>>>,

    /// Frame sender for broadcast
    frame_sender: broadcast::Sender<(String, MediaFrame)>,

    /// Event handling task
    event_task: Arc<Mutex<Option<JoinHandle<()>>>>,
}

// Custom implementation of Clone for DefaultMediaTransportServer
impl Clone for DefaultMediaTransportServer {
    fn clone(&self) -> Self {
        Self {
            id: self.id.clone(),
            config: self.config.clone(),
            running: self.running.clone(),
            main_socket: self.main_socket.clone(),
            socket: self.socket.clone(),
            transport: self.transport.clone(),
            client_sockets: self.client_sockets.clone(),
            client_transports: self.client_transports.clone(),
            security_context: self.security_context.clone(),
            clients: self.clients.clone(),
            event_callbacks: self.event_callbacks.clone(),
            client_connected_callbacks: self.client_connected_callbacks.clone(),
            client_disconnected_callbacks: self.client_disconnected_callbacks.clone(),
            ssrc_demultiplexing_enabled: self.ssrc_demultiplexing_enabled.clone(),
            csrc_management_enabled: self.csrc_management_enabled.clone(),
            csrc_manager: self.csrc_manager.clone(),
            header_extension_format: self.header_extension_format.clone(),
            header_extensions_enabled: self.header_extensions_enabled.clone(),
            header_extension_mappings: self.header_extension_mappings.clone(),
            pending_extensions: self.pending_extensions.clone(),
            received_extensions: self.received_extensions.clone(),
            frame_sender: self.frame_sender.clone(),
            event_task: self.event_task.clone(),
        }
    }
}

impl DefaultMediaTransportServer {
    /// Create a new DefaultMediaTransportServer
    pub async fn new(config: ServerConfig) -> Result<Self, MediaTransportError> {
        // Extract configuration values
        let csrc_management_enabled = config.csrc_management_enabled;
        let header_extensions_enabled = config.header_extensions_enabled;
        let header_extension_format = config.header_extension_format;
        let ssrc_demultiplexing_enabled = config.ssrc_demultiplexing_enabled.unwrap_or(false); // Read from config, default to false if not specified

        // Create broadcast channel for frames.
        let (sender, _) = broadcast::channel(config.frame_channel_capacity.max(1));

        Ok(Self {
            id: format!("media-server-main-{}", Uuid::new_v4()),
            config,
            running: Arc::new(RwLock::new(false)),
            main_socket: Arc::new(RwLock::new(None)),
            socket: Arc::new(RwLock::new(None)),
            transport: Arc::new(RwLock::new(None)),
            client_sockets: Arc::new(RwLock::new(HashMap::new())),
            client_transports: Arc::new(RwLock::new(HashMap::new())),
            security_context: Arc::new(RwLock::new(None)),
            clients: Arc::new(DashMap::new()),
            event_callbacks: Arc::new(RwLock::new(Vec::new())),
            client_connected_callbacks: Arc::new(RwLock::new(Vec::new())),
            client_disconnected_callbacks: Arc::new(RwLock::new(Vec::new())),
            ssrc_demultiplexing_enabled: Arc::new(AtomicBool::new(ssrc_demultiplexing_enabled)),
            csrc_management_enabled: Arc::new(AtomicBool::new(csrc_management_enabled)),
            csrc_manager: Arc::new(parking_lot::RwLock::new(CsrcManager::new())),
            header_extension_format: Arc::new(RwLock::new(header_extension_format)),
            header_extensions_enabled: Arc::new(RwLock::new(header_extensions_enabled)),
            header_extension_mappings: Arc::new(RwLock::new(HashMap::new())),
            pending_extensions: Arc::new(RwLock::new(HashMap::new())),
            received_extensions: Arc::new(RwLock::new(HashMap::new())),
            frame_sender: sender,
            event_task: Arc::new(Mutex::new(None)),
        })
    }

    /// Get the frame type based on payload type
    pub fn get_frame_type_from_payload_type(
        &self,
        payload_type: u8,
    ) -> crate::api::common::frame::MediaFrameType {
        super::util::get_frame_type_from_payload_type(payload_type)
    }

    /// Create a new instance of ServerMetrics
    pub async fn get_server_metrics(
        &self,
        server_start_time: std::time::Duration,
    ) -> Result<super::stats::ServerMetrics, MediaTransportError> {
        // Get current media stats
        let media_stats = self.get_stats().await?;

        // Create server metrics
        super::stats::get_server_metrics(&self.clients, &media_stats, server_start_time).await
    }
}

#[async_trait]
impl MediaTransportServer for DefaultMediaTransportServer {
    /// Start the server
    ///
    /// This binds to the configured address and starts listening for
    /// incoming client connections.
    async fn start(&self) -> Result<(), MediaTransportError> {
        // Check if already running
        {
            let running = self.running.read().await;
            if *running {
                return Ok(());
            }
        }

        // Set running flag
        {
            let mut running = self.running.write().await;
            *running = true;
        }

        // Initialize security if needed
        super::security::init_security_if_needed(&self.config, &self.security_context).await?;

        // Create main transport
        let transport_config = RtpTransportConfig {
            local_rtp_addr: self.config.local_address,
            local_rtcp_addr: None,
            symmetric_rtp: true,
            rtcp_mux: self.config.rtcp_mux,
            session_id: Some(format!("media-server-main-{}", Uuid::new_v4())),
            use_port_allocator: false, // Changed from true to false to use exact specified port
            buffer_config: self.config.rtp_transport_buffer_config,
        };

        debug!(
            "Creating main transport with config: {:?}",
            transport_config
        );

        // Create base UDP transport
        let udp_transport = UdpRtpTransport::new(transport_config).await.map_err(|e| {
            MediaTransportError::Transport(format!("Failed to create UDP transport: {}", e))
        })?;

        // Create SecurityRtpTransport wrapper
        let transport = SecurityRtpTransport::new(Arc::new(udp_transport), true)
            .await
            .map_err(|e| {
                MediaTransportError::Transport(format!(
                    "Failed to create SecurityRtpTransport: {}",
                    e
                ))
            })?;

        // Note: We don't start the UDP receiver here since SecurityRtpTransport handles packet reception

        let transport_arc: Arc<dyn RtpTransport> = Arc::new(transport);

        // Wire SRTP context if security is enabled
        let security_guard = self.security_context.read().await;
        if let Some(_security_ctx) = security_guard.as_ref() {
            // For now, we'll extract the SRTP key directly from the config
            // In a full implementation, we'd access it through security_ctx.get_config()
            if self.config.security_config.security_mode == SecurityMode::Srtp {
                if let Some(srtp_key) = &self.config.security_config.srtp_key {
                    debug!("Configuring server SRTP context with pre-shared keys");
                    debug!(
                        "Creating server SRTP context from {} byte key",
                        srtp_key.len()
                    );

                    // Extract key and salt (first 16 bytes = key, next 14 bytes = salt)
                    if srtp_key.len() >= 30 {
                        let key = srtp_key[0..16].to_vec();
                        let salt = srtp_key[16..30].to_vec();

                        debug!(
                            "Creating server SrtpCryptoKey with key: {} bytes, salt: {} bytes",
                            key.len(),
                            salt.len()
                        );
                        let crypto_key = SrtpCryptoKey::new(key, salt);

                        match SrtpContext::new(SRTP_AES128_CM_SHA1_80, crypto_key) {
                            Ok(srtp_context) => {
                                debug!("Successfully created server SRTP context");

                                // Set SRTP context on security transport
                                if let Some(sec_transport) = transport_arc
                                    .as_any()
                                    .downcast_ref::<SecurityRtpTransport>()
                                {
                                    sec_transport.set_srtp_context(srtp_context).await;
                                    info!(
                                        "SRTP context successfully configured on server transport"
                                    );
                                } else {
                                    warn!("Failed to downcast to SecurityRtpTransport for server SRTP context setting");
                                }
                            }
                            Err(e) => {
                                error!("Failed to create server SRTP context: {}", e);
                                return Err(MediaTransportError::Security(format!(
                                    "Server SRTP context creation failed: {}",
                                    e
                                )));
                            }
                        }
                    } else {
                        error!(
                            "Server SRTP key too short: {} bytes (expected at least 30)",
                            srtp_key.len()
                        );
                        return Err(MediaTransportError::Security(
                            "Server SRTP key too short".to_string(),
                        ));
                    }
                } else {
                    warn!("SRTP mode enabled on server but no key provided");
                }
            }
        }
        drop(security_guard);

        // Store the socket in a thread-safe way using RwLock
        {
            let mut main_socket = self.main_socket.write().await;
            *main_socket = Some(transport_arc.clone());
        }

        // Subscribe to transport events
        let mut transport_events = transport_arc.subscribe();
        let clients_clone = self.clients.clone();
        let sender_clone = self.frame_sender.clone();
        let ssrc_demux_enabled_clone = self.ssrc_demultiplexing_enabled.clone();
        let session_buffer_config = self.config.rtp_session_buffer_config;
        let transport_buffer_config = self.config.rtp_transport_buffer_config;

        let task = tokio::spawn(async move {
            debug!("Started transport event task");
            while let Ok(event) = transport_events.recv().await {
                match event {
                    crate::traits::RtpEvent::MediaReceived {
                        source,
                        payload,
                        payload_type,
                        sequence_number: _,
                        timestamp,
                        marker,
                        ssrc,
                    } => {
                        // Debug output to help diagnose issues
                        debug!(
                            "RtpEvent::MediaReceived from {} - SSRC={:08x}, payload size={}",
                            source,
                            ssrc,
                            payload.len()
                        );

                        // Smart detection: if the payload is small and looks like text/media data rather than RTP packet,
                        // it's likely already processed by SecurityRtpTransport
                        let is_processed_payload = payload.len() < 100
                            && (payload.len() < 12 || // Too small for RTP header
                             (payload.len() >= 1 && payload[0] != 0x80)); // First byte doesn't look like RTP version 2

                        let (frame, final_ssrc) = if is_processed_payload {
                            // This is already processed payload from SecurityRtpTransport - use event SSRC
                            debug!("Detected processed payload from SecurityRtpTransport - using event SSRC={:08x}", ssrc);

                            // Use the RTP header information already provided in the event
                            let frame = crate::api::common::frame::MediaFrame {
                                frame_type: crate::api::common::frame::MediaFrameType::Audio, // Default to Audio, media-core handles frame type
                                data: payload.clone(),
                                timestamp,
                                sequence: 0, // SecurityRtpTransport doesn't provide sequence in event
                                marker,
                                payload_type,
                                ssrc, // Use the SSRC from the event!
                                csrcs: Vec::new(),
                            };

                            // Log payload data if it's text (helps with debugging)
                            if let Ok(text) = std::str::from_utf8(&payload) {
                                debug!("Processed payload (text): {}", text);
                            }

                            // Use the event SSRC directly
                            (frame, ssrc)
                        } else {
                            // This looks like a raw RTP packet - parse it normally
                            debug!("Attempting to parse raw RTP packet from payload...");
                            let packet_result =
                                crate::packet::RtpPacket::parse_from_bytes(payload.clone());

                            match packet_result {
                                Ok(packet) => {
                                    // Successfully parsed the packet, use its header information
                                    debug!("Successfully parsed RTP packet with SSRC={:08x}, seq={}, ts={}, PT={}",
                                           packet.header.ssrc, packet.header.sequence_number,
                                           packet.header.timestamp, packet.header.payload_type);

                                    // Log payload data if it's text (helps with debugging)
                                    if let Ok(text) = std::str::from_utf8(&packet.payload) {
                                        debug!("Raw packet payload (text): {}", text);
                                    }

                                    let frame = crate::api::common::frame::MediaFrame {
                                        frame_type:
                                            crate::api::common::frame::MediaFrameType::Audio, // Default to Audio, media-core handles frame type
                                        data: packet.payload,
                                        timestamp: packet.header.timestamp,
                                        sequence: packet.header.sequence_number,
                                        marker: packet.header.marker,
                                        payload_type: packet.header.payload_type,
                                        ssrc: packet.header.ssrc,
                                        csrcs: packet.header.csrc.clone(),
                                    };

                                    (frame, packet.header.ssrc)
                                }
                                Err(e) => {
                                    // If we couldn't parse the packet, create a MediaFrame with the available info
                                    debug!(
                                        "Couldn't parse RTP packet, using available info: {}",
                                        e
                                    );

                                    // Attempt to dump the first few bytes for debugging
                                    if payload.len() >= 12 {
                                        let mut header_bytes = [0u8; 12];
                                        header_bytes.copy_from_slice(&payload[0..12]);
                                        debug!("First 12 bytes of payload: {:?}", header_bytes);
                                    }

                                    // Generate a random sequence number and SSRC as fallback
                                    let sequence_number = rand::random::<u16>();
                                    let ssrc = rand::random::<u32>();

                                    let frame = crate::api::common::frame::MediaFrame {
                                        frame_type:
                                            crate::api::common::frame::MediaFrameType::Audio, // Default to Audio, media-core handles frame type
                                        data: payload.clone(),
                                        timestamp,
                                        sequence: sequence_number,
                                        marker,
                                        payload_type,
                                        ssrc,
                                        csrcs: Vec::new(), // No CSRCs in fallback case
                                    };
                                    (frame, ssrc)
                                }
                            }
                        };

                        // Check if SSRC demultiplexing is enabled
                        let ssrc_demux_enabled = ssrc_demux_enabled_clone.load(Ordering::Relaxed);
                        debug!(
                            "SSRC demultiplexing enabled: {}, handling packet with SSRC={:08x}",
                            ssrc_demux_enabled, final_ssrc
                        );

                        // Check if we already have a client for this address.
                        // DashMap iter scope ends with `.find()` returning.
                        let existing_id = clients_clone
                            .iter()
                            .find(|e| e.value().address == source)
                            .map(|e| e.value().id.clone());
                        let client_exists = existing_id.is_some();
                        let client_id =
                            existing_id.unwrap_or_else(|| format!("pending-{}", source));

                        // Forward directly to broadcast channel
                        // This ensures frames are available immediately via the receive_frame method
                        match sender_clone.send((client_id.clone(), frame)) {
                            Ok(receivers) => {
                                debug!("Directly forwarded frame to {} receivers for client {} (SSRC={:08x})",
                                        receivers, client_id, final_ssrc);
                            }
                            Err(e) => {
                                debug!("No receivers for direct frame forwarding: {}", e);
                            }
                        }

                        // Handle client creation if new
                        if !client_exists {
                            // New client - handle it
                            debug!("New client detected at {}, handling...", source);
                            let client_result = super::core::handle_client_static(
                                source,
                                &clients_clone,
                                &sender_clone,
                                session_buffer_config,
                                transport_buffer_config,
                            )
                            .await;

                            match client_result {
                                Ok(client_id) => debug!(
                                    "Successfully handled new client {} from {}",
                                    client_id, source
                                ),
                                Err(e) => {
                                    error!("Failed to handle new client from {}: {}", source, e)
                                }
                            }
                        } else {
                            debug!(
                                "Existing client from {}, no new client creation needed",
                                source
                            );
                        }
                    }
                    crate::traits::RtpEvent::RtcpReceived { source, .. } => {
                        debug!("RtpEvent::RtcpReceived from {}", source);
                    }
                    crate::traits::RtpEvent::DtmfEvent {
                        source,
                        event,
                        end_of_event,
                        timestamp: _,
                        ..
                    } => {
                        // RFC 4733 DTMF is surfaced on the transport
                        // broadcast channel directly; the server-side
                        // transport doesn't synthesize a MediaFrame.
                        debug!(
                            "RtpEvent::DtmfEvent from {} event={} end={}",
                            source, event, end_of_event
                        );
                    }
                    crate::traits::RtpEvent::Error(e) => {
                        error!("Transport error: {}", e);
                    }
                }
            }
            debug!("Transport event task ended");
        });

        // Store task handle
        let mut event_task = self.event_task.lock().await;
        *event_task = Some(task);

        Ok(())
    }

    /// Stop the server
    ///
    /// This stops listening for new connections and disconnects all
    /// existing clients.
    async fn stop(&self) -> Result<(), MediaTransportError> {
        // Check if we're running
        {
            let running = self.running.read().await;
            if !*running {
                debug!("Server already stopped, no action needed");
                return Ok(());
            }
        }

        // Set not running
        {
            let mut running = self.running.write().await;
            *running = false;
        }

        debug!("Stopping server - aborting event task");

        // Stop event task first to prevent new client connections
        {
            let mut event_task = self.event_task.lock().await;
            if let Some(task) = event_task.take() {
                debug!("Aborting main event task");
                task.abort();
                // Try to wait for the task to finish (with timeout)
                match tokio::time::timeout(Duration::from_millis(100), task).await {
                    Ok(_) => debug!("Event task terminated gracefully"),
                    Err(_) => debug!("Event task termination timed out"),
                }
            } else {
                debug!("No event task to abort");
            }
        }

        debug!("Disconnecting all clients");

        // Drain the DashMap into an owned Vec so we can run the
        // per-client async tear-down (`session.close().await`,
        // `security.close().await`, callback dispatch) without
        // holding any shard guard across `.await`. `iter()` /
        // `get_mut()` on DashMap return shard locks that would
        // serialise other shard work — draining avoids that
        // entirely.
        let drained: Vec<(String, ClientConnection)> = self
            .clients
            .iter()
            .map(|entry| entry.key().clone())
            .collect::<Vec<_>>()
            .into_iter()
            .filter_map(|k| self.clients.remove(&k))
            .collect();
        debug!("Disconnecting {} clients", drained.len());

        for (id, mut client) in drained {
            debug!("Disconnecting client {}", id);

            // Abort task
            if let Some(task) = client.task.take() {
                debug!("Aborting client task for {}", id);
                task.abort();
                // Try to wait for the task to finish (with timeout)
                match tokio::time::timeout(Duration::from_millis(100), task).await {
                    Ok(_) => debug!("Client task for {} terminated gracefully", id),
                    Err(_) => debug!("Client task termination for {} timed out", id),
                }
            }

            // Close session
            debug!("Closing session for client {}", id);
            {
                let mut session = client.session.lock().await;
                if let Err(e) = session.close().await {
                    warn!("Error closing client session {}: {}", id, e);
                }
            }

            // Close security context if present
            if let Some(security_ctx) = &client.security {
                debug!("Closing security for client {}", id);
                if let Err(e) = security_ctx.close().await {
                    warn!("Error closing client security {}: {}", id, e);
                }
            }

            // Mark as disconnected
            client.connected = false;

            // Notify callbacks
            debug!("Notifying disconnection callbacks for client {}", id);
            let callbacks_guard = self.client_disconnected_callbacks.read().await;
            let client_info = ClientInfo {
                id: client.id.clone(),
                address: client.address,
                secure: client.security.is_some(),
                security_info: None,
                connected: false,
            };

            for callback in &*callbacks_guard {
                callback(client_info.clone());
            }
            drop(callbacks_guard);
        }
        debug!("Disconnect-all complete");

        // Close main socket if available
        debug!("Closing main socket");
        let mut main_socket = self.main_socket.write().await;
        if let Some(socket) = main_socket.take() {
            debug!("Closing main transport socket");
            if let Err(e) = socket.close().await {
                warn!("Error closing main socket: {}", e);
            }
        } else {
            debug!("No main socket to close");
        }

        // Ensure we release broadcast channel resources
        debug!("Ensuring broadcast channel resources are released");
        // Create a temporary receiver and then immediately drop it to avoid any lingering receivers
        {
            let _temp_receiver = self.frame_sender.subscribe();
            // Immediately drop the receiver
        }

        debug!("Server stopped successfully");
        Ok(())
    }

    /// Receive a media frame from any client
    async fn receive_frame(&self) -> Result<(String, MediaFrame), MediaTransportError> {
        super::core::receive_frame(&self.frame_sender).await
    }

    /// Get a persistent frame receiver for receiving multiple frames
    fn get_frame_receiver(&self) -> broadcast::Receiver<(String, MediaFrame)> {
        self.frame_sender.subscribe()
    }

    /// Get the local address currently bound to
    async fn get_local_address(&self) -> Result<SocketAddr, MediaTransportError> {
        let main_socket = self.main_socket.read().await;
        if let Some(socket) = main_socket.as_ref() {
            match socket.local_rtp_addr() {
                Ok(addr) => Ok(addr),
                Err(e) => Err(MediaTransportError::Transport(format!(
                    "Failed to get local address: {}",
                    e
                ))),
            }
        } else {
            Err(MediaTransportError::Transport(
                "No socket bound yet. Start server first.".to_string(),
            ))
        }
    }

    /// Send a media frame to a specific client
    async fn send_frame_to(
        &self,
        client_id: &str,
        frame: MediaFrame,
    ) -> Result<(), MediaTransportError> {
        super::core::send_frame_to(
            client_id,
            frame,
            &self.clients,
            &self.ssrc_demultiplexing_enabled,
            &self.csrc_management_enabled,
            &self.csrc_manager,
            &self.main_socket,
        )
        .await
    }

    /// Broadcast a media frame to all connected clients
    async fn broadcast_frame(&self, frame: MediaFrame) -> Result<(), MediaTransportError> {
        super::core::broadcast_frame(
            frame,
            &self.clients,
            &self.csrc_management_enabled,
            &self.csrc_manager,
            &self.main_socket,
        )
        .await
    }

    /// Get a list of connected clients
    async fn get_clients(&self) -> Result<Vec<ClientInfo>, MediaTransportError> {
        super::core::get_clients_info(&self.clients, &self.config).await
    }

    /// Disconnect a specific client
    async fn disconnect_client(&self, client_id: &str) -> Result<(), MediaTransportError> {
        super::core::disconnect_client(
            client_id,
            &self.clients,
            &self.client_disconnected_callbacks,
        )
        .await
    }

    /// Register a callback for transport events
    async fn on_event(&self, callback: MediaEventCallback) -> Result<(), MediaTransportError> {
        super::core::on_event(callback, &self.event_callbacks).await
    }

    /// Register a callback for client connection events
    async fn on_client_connected(
        &self,
        callback: Box<dyn Fn(ClientInfo) + Send + Sync>,
    ) -> Result<(), MediaTransportError> {
        super::core::on_client_connected(callback, &self.client_connected_callbacks).await
    }

    /// Register a callback for client disconnection events
    async fn on_client_disconnected(
        &self,
        callback: Box<dyn Fn(ClientInfo) + Send + Sync>,
    ) -> Result<(), MediaTransportError> {
        super::core::on_client_disconnected(callback, &self.client_disconnected_callbacks).await
    }

    /// Get statistics for all clients
    async fn get_stats(&self) -> Result<MediaStats, MediaTransportError> {
        // Stats aggregation moved to media-core
        Ok(MediaStats::default())
    }

    /// Get statistics for a specific client
    async fn get_client_stats(&self, _client_id: &str) -> Result<MediaStats, MediaTransportError> {
        // Stats aggregation moved to media-core
        Ok(MediaStats::default())
    }

    /// Get security information for SDP exchange
    async fn get_security_info(&self) -> Result<SecurityInfo, MediaTransportError> {
        super::security::get_security_info(&self.config, &self.security_context).await
    }

    async fn send_rtcp_receiver_report(&self) -> Result<(), MediaTransportError> {
        super::rtcp::send_rtcp_receiver_report(&self.clients).await
    }

    async fn send_rtcp_sender_report(&self) -> Result<(), MediaTransportError> {
        super::rtcp::send_rtcp_sender_report(&self.clients).await
    }

    async fn send_rtcp_receiver_report_to_client(
        &self,
        client_id: &str,
    ) -> Result<(), MediaTransportError> {
        super::rtcp::send_rtcp_receiver_report_to_client(client_id, &self.clients).await
    }

    async fn send_rtcp_sender_report_to_client(
        &self,
        client_id: &str,
    ) -> Result<(), MediaTransportError> {
        super::rtcp::send_rtcp_sender_report_to_client(client_id, &self.clients).await
    }

    async fn get_rtcp_stats(&self) -> Result<RtcpStats, MediaTransportError> {
        super::rtcp::get_rtcp_stats(&self.clients).await
    }

    async fn get_client_rtcp_stats(
        &self,
        client_id: &str,
    ) -> Result<RtcpStats, MediaTransportError> {
        super::rtcp::get_client_rtcp_stats(client_id, &self.clients).await
    }

    async fn set_rtcp_interval(&self, interval: Duration) -> Result<(), MediaTransportError> {
        super::rtcp::set_rtcp_interval(&self.clients, interval).await
    }

    async fn send_rtcp_app(&self, name: &str, data: Vec<u8>) -> Result<(), MediaTransportError> {
        super::rtcp::send_rtcp_app(&self.clients, &self.main_socket, name, data).await
    }

    async fn send_rtcp_app_to_client(
        &self,
        client_id: &str,
        name: &str,
        data: Vec<u8>,
    ) -> Result<(), MediaTransportError> {
        super::rtcp::send_rtcp_app_to_client(
            client_id,
            &self.clients,
            &self.main_socket,
            name,
            data,
        )
        .await
    }

    async fn send_rtcp_bye(&self, reason: Option<String>) -> Result<(), MediaTransportError> {
        super::rtcp::send_rtcp_bye(&self.clients, &self.main_socket, reason).await
    }

    async fn send_rtcp_bye_to_client(
        &self,
        client_id: &str,
        reason: Option<String>,
    ) -> Result<(), MediaTransportError> {
        super::rtcp::send_rtcp_bye_to_client(client_id, &self.clients, &self.main_socket, reason)
            .await
    }

    async fn send_rtcp_xr_voip_metrics(
        &self,
        metrics: VoipMetrics,
    ) -> Result<(), MediaTransportError> {
        super::rtcp::send_rtcp_xr_voip_metrics(&self.clients, &self.main_socket, metrics).await
    }

    async fn send_rtcp_xr_voip_metrics_to_client(
        &self,
        client_id: &str,
        metrics: VoipMetrics,
    ) -> Result<(), MediaTransportError> {
        super::rtcp::send_rtcp_xr_voip_metrics_to_client(
            client_id,
            &self.clients,
            &self.main_socket,
            metrics,
        )
        .await
    }

    async fn is_csrc_management_enabled(&self) -> Result<bool, MediaTransportError> {
        Ok(self.csrc_management_enabled.load(Ordering::Relaxed))
    }

    async fn enable_csrc_management(&self) -> Result<bool, MediaTransportError> {
        // CAS so re-enable is a no-op observable via the return value.
        if self
            .csrc_management_enabled
            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
            .is_err()
        {
            return Ok(true);
        }
        debug!("Enabled CSRC management on server");
        Ok(true)
    }

    async fn add_csrc_mapping(&self, _mapping: CsrcMapping) -> Result<(), MediaTransportError> {
        // CSRC management moved to media-core
        Ok(())
    }

    async fn add_simple_csrc_mapping(
        &self,
        _original_ssrc: RtpSsrc,
        _csrc: RtpCsrc,
    ) -> Result<(), MediaTransportError> {
        // CSRC management moved to media-core
        Ok(())
    }

    async fn remove_csrc_mapping_by_ssrc(
        &self,
        _original_ssrc: RtpSsrc,
    ) -> Result<Option<CsrcMapping>, MediaTransportError> {
        // CSRC management moved to media-core
        Ok(None)
    }

    async fn get_csrc_mapping_by_ssrc(
        &self,
        _original_ssrc: RtpSsrc,
    ) -> Result<Option<CsrcMapping>, MediaTransportError> {
        // CSRC management moved to media-core
        Ok(None)
    }

    async fn get_all_csrc_mappings(&self) -> Result<Vec<CsrcMapping>, MediaTransportError> {
        // CSRC management moved to media-core
        Ok(Vec::new())
    }

    async fn get_active_csrcs(
        &self,
        _active_ssrcs: &[RtpSsrc],
    ) -> Result<Vec<RtpCsrc>, MediaTransportError> {
        // CSRC management moved to media-core
        Ok(Vec::new())
    }

    async fn is_header_extensions_enabled(&self) -> Result<bool, MediaTransportError> {
        Ok(*self.header_extensions_enabled.read().await)
    }

    async fn enable_header_extensions(
        &self,
        format: ExtensionFormat,
    ) -> Result<bool, MediaTransportError> {
        // Check if already enabled
        if *self.header_extensions_enabled.read().await {
            return Ok(true);
        }

        // Set format and enabled flag
        *self.header_extension_format.write().await = format;
        *self.header_extensions_enabled.write().await = true;

        debug!("Enabled header extensions with format: {:?}", format);
        Ok(true)
    }

    async fn configure_header_extension(
        &self,
        _id: u8,
        _uri: String,
    ) -> Result<(), MediaTransportError> {
        // Header extension configuration moved to media-core
        Ok(())
    }

    async fn configure_header_extensions(
        &self,
        _mappings: HashMap<u8, String>,
    ) -> Result<(), MediaTransportError> {
        // Header extension configuration moved to media-core
        Ok(())
    }

    async fn add_header_extension_for_client(
        &self,
        _client_id: &str,
        _extension: HeaderExtension,
    ) -> Result<(), MediaTransportError> {
        // Header extension handling moved to media-core
        Ok(())
    }

    async fn add_header_extension_for_all_clients(
        &self,
        _extension: HeaderExtension,
    ) -> Result<(), MediaTransportError> {
        // Header extension handling moved to media-core
        Ok(())
    }

    async fn add_audio_level_extension_for_client(
        &self,
        _client_id: &str,
        _voice_activity: bool,
        _level: u8,
    ) -> Result<(), MediaTransportError> {
        // Audio level extension handling moved to media-core
        Ok(())
    }

    async fn add_audio_level_extension_for_all_clients(
        &self,
        _voice_activity: bool,
        _level: u8,
    ) -> Result<(), MediaTransportError> {
        // Audio level extension handling moved to media-core
        Ok(())
    }

    async fn add_video_orientation_extension_for_client(
        &self,
        _client_id: &str,
        _camera_front_facing: bool,
        _camera_flipped: bool,
        _rotation: u16,
    ) -> Result<(), MediaTransportError> {
        todo!("Implement add_video_orientation_extension_for_client")
    }

    async fn add_video_orientation_extension_for_all_clients(
        &self,
        _camera_front_facing: bool,
        _camera_flipped: bool,
        _rotation: u16,
    ) -> Result<(), MediaTransportError> {
        todo!("Implement add_video_orientation_extension_for_all_clients")
    }

    async fn add_transport_cc_extension_for_client(
        &self,
        _client_id: &str,
        _sequence_number: u16,
    ) -> Result<(), MediaTransportError> {
        todo!("Implement add_transport_cc_extension_for_client")
    }

    async fn add_transport_cc_extension_for_all_clients(
        &self,
        _sequence_number: u16,
    ) -> Result<(), MediaTransportError> {
        todo!("Implement add_transport_cc_extension_for_all_clients")
    }

    async fn get_received_header_extensions(
        &self,
        _client_id: &str,
    ) -> Result<Vec<HeaderExtension>, MediaTransportError> {
        // Header extension handling moved to media-core
        Ok(Vec::new())
    }

    async fn get_received_audio_level(
        &self,
        _client_id: &str,
    ) -> Result<Option<(bool, u8)>, MediaTransportError> {
        // Audio level extension handling moved to media-core
        Ok(None)
    }

    async fn get_received_video_orientation(
        &self,
        _client_id: &str,
    ) -> Result<Option<(bool, bool, u16)>, MediaTransportError> {
        todo!("Implement get_received_video_orientation")
    }

    async fn get_received_transport_cc(
        &self,
        _client_id: &str,
    ) -> Result<Option<u16>, MediaTransportError> {
        todo!("Implement get_received_transport_cc")
    }

    async fn is_ssrc_demultiplexing_enabled(&self) -> Result<bool, MediaTransportError> {
        super::ssrc::is_ssrc_demultiplexing_enabled(&self.ssrc_demultiplexing_enabled).await
    }

    async fn enable_ssrc_demultiplexing(&self) -> Result<bool, MediaTransportError> {
        super::ssrc::enable_ssrc_demultiplexing(&self.ssrc_demultiplexing_enabled).await
    }

    async fn register_client_ssrc(
        &self,
        client_id: &str,
        ssrc: u32,
    ) -> Result<bool, MediaTransportError> {
        super::ssrc::register_client_ssrc(
            client_id,
            ssrc,
            &self.ssrc_demultiplexing_enabled,
            &self.clients,
        )
        .await
    }

    async fn get_client_ssrcs(&self, client_id: &str) -> Result<Vec<u32>, MediaTransportError> {
        super::ssrc::get_client_ssrcs(client_id, &self.clients).await
    }

    async fn update_csrc_cname(
        &self,
        _original_ssrc: RtpSsrc,
        _cname: String,
    ) -> Result<bool, MediaTransportError> {
        // CSRC management moved to media-core
        Ok(false)
    }

    async fn update_csrc_display_name(
        &self,
        _original_ssrc: RtpSsrc,
        _name: String,
    ) -> Result<bool, MediaTransportError> {
        // CSRC management moved to media-core
        Ok(false)
    }
}