rvoip-rtp-core 0.2.5

//! RTP Session Management
//!
//! This module provides functionality for managing RTP sessions, including
//! configuration, packet sending/receiving, and jitter buffer management.

mod scheduling;
mod stream;

pub use scheduling::{RtpScheduler, RtpSchedulerStats};
pub use stream::{RtpStream, RtpStreamStats};

use bytes::{Bytes, BytesMut};
use dashmap::DashMap;
use rand::Rng;
use std::net::SocketAddr;
use std::sync::atomic::{AtomicU16, Ordering};
use std::sync::Arc;
use std::time::Duration;
use tokio::net::UdpSocket;
use tokio::sync::{broadcast, mpsc};
use tokio::task::JoinHandle;
use tracing::{debug, error, info, trace, warn};

use crate::error::Error;
use crate::packet::{RtpHeader, RtpPacket};
use crate::transport::{
    RtpTransport, RtpTransportBufferConfig, RtpTransportConfig, UdpRtpTransport,
};
use crate::{Result, RtpSsrc, RtpTimestamp};

#[cfg(feature = "memory-diagnostics")]
fn spawn_memory_tracked<F>(kind: &'static str, future: F) -> JoinHandle<F::Output>
where
    F: std::future::Future + Send + 'static,
    F::Output: Send + 'static,
{
    rvoip_infra_common::memory_diagnostics::spawn_tracked(kind, future)
}

#[cfg(not(feature = "memory-diagnostics"))]
fn spawn_memory_tracked<F>(_: &'static str, future: F) -> JoinHandle<F::Output>
where
    F: std::future::Future + Send + 'static,
    F::Output: Send + 'static,
{
    tokio::spawn(future)
}

/// Bounded queue depth for per-session RTP send/event channels.
///
/// RTP is real-time traffic; keeping many seconds of packet backlog per call
/// hides overload and retains packet payloads. At 20 ms packets, 64 entries is
/// roughly 1.3 seconds of headroom for one stream.
pub const RTP_SESSION_CHANNEL_CAPACITY: usize = 64;

/// Small best-effort queue for the legacy polling receive API.
///
/// Media-core consumes RTP packets through the event broadcast path, so this
/// queue must not become an unbounded duplicate packet buffer when nobody calls
/// [`RtpSession::receive_packet`].
pub const RTP_SESSION_RECEIVE_QUEUE_CAPACITY: usize = 32;

/// RTP session queue sizing.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RtpSessionBufferConfig {
    /// Bounded sender queue capacity in RTP packets.
    pub sender_channel_capacity: usize,
    /// Bounded legacy polling receive queue capacity in RTP packets.
    pub receiver_channel_capacity: usize,
    /// Broadcast ring capacity for RTP session events.
    pub event_channel_capacity: usize,
}

impl Default for RtpSessionBufferConfig {
    fn default() -> Self {
        Self {
            sender_channel_capacity: RTP_SESSION_CHANNEL_CAPACITY,
            receiver_channel_capacity: RTP_SESSION_RECEIVE_QUEUE_CAPACITY,
            event_channel_capacity: RTP_SESSION_CHANNEL_CAPACITY,
        }
    }
}

/// Stats for an RTP session
#[derive(Debug, Clone, Default)]
pub struct RtpSessionStats {
    /// Total packets sent
    pub packets_sent: u64,

    /// Total packets received
    pub packets_received: u64,

    /// Total bytes sent
    pub bytes_sent: u64,

    /// Total bytes received
    pub bytes_received: u64,

    /// Packets lost (based on sequence numbers)
    pub packets_lost: u64,

    /// Duplicate packets received
    pub packets_duplicated: u64,

    /// Out-of-order packets received
    pub packets_out_of_order: u64,

    /// Packets discarded by jitter buffer (too old)
    pub packets_discarded_by_jitter: u64,

    /// Current jitter estimate (in milliseconds)
    pub jitter_ms: f64,

    /// Remote address of the most recent packet
    pub remote_addr: Option<SocketAddr>,
}

/// Snapshot of bounded queue occupancy inside an RTP session.
#[derive(Debug, Clone, Copy, Default)]
pub struct RtpSessionQueueDiagnostics {
    /// Packets waiting to be sent by the RTP send task.
    pub sender_queue_packets: usize,
    /// Configured sender queue capacity.
    pub sender_capacity_packets: usize,
    /// Packets waiting in the receive queue for explicit `receive_packet` users.
    pub receiver_queue_packets: usize,
    /// Configured receiver queue capacity.
    pub receiver_capacity_packets: usize,
    /// Events retained in the broadcast ring.
    pub event_queue_events: usize,
    /// Current subscribers to the event broadcast ring.
    pub event_receiver_count: usize,
    #[cfg(feature = "memory-diagnostics")]
    /// Current SSRC stream entries retained by this session.
    pub stream_count: usize,
}

/// RTP session configuration options
#[derive(Debug, Clone)]
pub struct RtpSessionConfig {
    /// Local address to bind to
    pub local_addr: SocketAddr,

    /// Remote address to send packets to
    pub remote_addr: Option<SocketAddr>,

    /// SSRC to use for sending packets
    pub ssrc: Option<RtpSsrc>,

    /// Payload type
    pub payload_type: u8,

    /// Clock rate for the payload type (needed for jitter buffer)
    pub clock_rate: u32,

    /// Jitter buffer size in packets
    pub jitter_buffer_size: Option<usize>,

    /// Maximum packet age in the jitter buffer (ms)
    pub max_packet_age_ms: Option<u32>,

    /// Enable jitter buffer
    pub enable_jitter_buffer: bool,

    /// RTP session queue and reusable send-buffer sizing.
    pub session_buffer_config: RtpSessionBufferConfig,

    /// UDP transport buffer sizing used when the session creates its transport.
    pub transport_buffer_config: RtpTransportBufferConfig,
}

impl Default for RtpSessionConfig {
    fn default() -> Self {
        Self {
            local_addr: "0.0.0.0:0".parse().unwrap(),
            remote_addr: None,
            ssrc: None,
            payload_type: 0,
            clock_rate: 8000, // Default for most audio codecs (8kHz)
            jitter_buffer_size: Some(50),
            max_packet_age_ms: Some(200),
            enable_jitter_buffer: true,
            session_buffer_config: RtpSessionBufferConfig::default(),
            transport_buffer_config: RtpTransportBufferConfig::default(),
        }
    }
}

/// Lock-free handle for sending RTP packets through an existing
/// [`RtpSession`] without touching the outer `Arc<Mutex<RtpSession>>`.
///
/// Cheap to clone (3 Arcs + 2 small scalars). Issued by
/// [`RtpSession::send_handle`]; multiple handles for the same session
/// stay in sync because they share the same sequence atomic.
#[derive(Clone)]
pub struct RtpSendHandle {
    sender: mpsc::Sender<RtpPacket>,
    ssrc: RtpSsrc,
    sequence: Arc<AtomicU16>,
    default_payload_type: u8,
}

impl RtpSendHandle {
    /// Send an RTP packet with the session's default payload type.
    pub async fn send_packet(
        &self,
        timestamp: RtpTimestamp,
        payload: Bytes,
        marker: bool,
    ) -> Result<()> {
        self.send_packet_with_pt(timestamp, payload, marker, self.default_payload_type)
            .await
    }

    /// Send an RTP packet overriding the configured payload type
    /// (e.g. RFC 4733 telephone-event PT 101).
    pub async fn send_packet_with_pt(
        &self,
        timestamp: RtpTimestamp,
        payload: Bytes,
        marker: bool,
        payload_type: u8,
    ) -> Result<()> {
        let sequence = self.sequence.fetch_add(1, Ordering::Relaxed);
        let mut header = RtpHeader::new(payload_type, sequence, timestamp, self.ssrc);
        header.marker = marker;
        let packet = RtpPacket::new(header, payload);
        self.sender
            .send(packet)
            .await
            .map_err(|_| Error::SessionError("Failed to send packet".to_string()))
    }

    /// Get the session's SSRC (immutable post-construction).
    pub fn ssrc(&self) -> RtpSsrc {
        self.ssrc
    }
}

/// Events emitted by the RTP session
#[derive(Debug, Clone)]
pub enum RtpSessionEvent {
    /// New packet received
    PacketReceived(RtpPacket),

    /// Error in the session
    Error(Error),

    /// BYE RTCP packet received (a party is leaving the session)
    Bye {
        /// SSRC of the source that sent the BYE
        ssrc: RtpSsrc,

        /// Optional reason text
        reason: Option<String>,
    },

    /// New stream detected with a specific SSRC
    /// This event is emitted as soon as the first packet for a new SSRC is received,
    /// even if the packet is being held in a jitter buffer.
    NewStreamDetected {
        /// SSRC of the new stream
        ssrc: RtpSsrc,
    },

    /// RTCP Sender Report received
    RtcpSenderReport {
        /// SSRC of the sender
        ssrc: RtpSsrc,

        /// NTP timestamp
        ntp_timestamp: crate::packet::rtcp::NtpTimestamp,

        /// RTP timestamp
        rtp_timestamp: RtpTimestamp,

        /// Packet count
        packet_count: u32,

        /// Octet count
        octet_count: u32,

        /// Report blocks
        report_blocks: Vec<crate::packet::rtcp::RtcpReportBlock>,
    },

    /// RTCP Receiver Report received
    RtcpReceiverReport {
        /// SSRC of the receiver
        ssrc: RtpSsrc,

        /// Report blocks
        report_blocks: Vec<crate::packet::rtcp::RtcpReportBlock>,
    },

    /// RFC 4733 telephone-event (DTMF / fax / modem tone) received.
    /// Forwarded verbatim from the transport-level `RtpEvent::DtmfEvent`.
    /// Consumers should forward the digit up to the application only on
    /// the frame where `end_of_event == true` — RFC 4733 §2.5.1.3
    /// requires three final retransmissions so the last three frames
    /// of each tone all set the `E` bit — and dedup on `(ssrc, timestamp)`
    /// which uniquely identifies a tone.
    DtmfReceived {
        /// Event code (0-15 for DTMF).
        event: u8,
        /// End-of-event `E` bit.
        end_of_event: bool,
        /// -dBm0 volume (0-63).
        volume: u8,
        /// Duration in RTP timestamp units.
        duration: u16,
        /// RTP packet timestamp (dedup key for retransmits).
        timestamp: u32,
        /// SSRC that sent the event.
        ssrc: RtpSsrc,
    },
}

/// RTP session for sending and receiving RTP packets
///
/// This class manages an RTP session, including sending and receiving packets,
/// jitter buffer management, and demultiplexing of multiple streams.
///
/// # SSRC Demultiplexing
///
/// An RTP session can receive packets from multiple sources, each identified by
/// a unique Synchronization Source identifier (SSRC). This implementation
/// automatically demultiplexes incoming packets based on their SSRC:
///
/// 1. When a packet arrives, its SSRC is extracted
/// 2. If this is the first packet from this SSRC, a new stream is created
/// 3. The packet is processed by the appropriate stream, which handles:
///    - Sequence number tracking
///    - Jitter calculation
///    - Duplicate detection
///    - Packet reordering (via jitter buffer)
///
/// Each stream maintains its own statistics and state. You can access information
/// about individual streams using the `get_stream()`, `get_all_streams()`, and
/// `stream_count()` methods.
///
/// This approach aligns with RFC 3550 Section 8.2, which describes how to handle
/// multiple sources in a single RTP session.
pub struct RtpSession {
    /// Session configuration
    config: RtpSessionConfig,

    /// SSRC for this session
    ssrc: RtpSsrc,

    /// Transport for sending/receiving packets
    transport: Arc<dyn RtpTransport>,

    /// Map of received streams by SSRC. `DashMap` so the per-packet
    /// demultiplex hot path (`session/mod.rs:620`+) doesn't serialise
    /// every receive through a single mutex, and so `get_stream` /
    /// `stream_count` readers don't block the demux task.
    streams: Arc<DashMap<RtpSsrc, RtpStream>>,

    /// Packet scheduler for sending packets
    scheduler: Option<RtpScheduler>,

    /// Channel for receiving packets
    receiver: mpsc::Receiver<RtpPacket>,

    /// Channel for sending packets
    sender: mpsc::Sender<RtpPacket>,

    /// Whether received RTP packets should also be mirrored into the legacy
    /// polling receive queue.
    receive_queue_enabled: bool,

    /// Event broadcaster
    event_tx: broadcast::Sender<RtpSessionEvent>,

    /// Receiving task handle
    recv_task: Option<JoinHandle<()>>,

    /// Sending task handle
    send_task: Option<JoinHandle<()>>,

    /// Session statistics. `parking_lot::Mutex` because every guard is
    /// CPU-only (counter updates, snapshot reads); the std variant
    /// added avoidable lock-acquire overhead on the send/recv hot
    /// paths and forced everything to unwrap poison.
    stats: Arc<parking_lot::Mutex<RtpSessionStats>>,

    /// Media synchronization context
    media_sync: Option<Arc<std::sync::RwLock<crate::sync::MediaSync>>>,

    /// Whether the session is active
    active: bool,

    /// RTCP report generator
    rtcp_generator: Option<crate::stats::reports::RtcpReportGenerator>,

    /// RTCP sender task
    rtcp_task: Option<JoinHandle<()>>,

    /// Session bandwidth (bits per second)
    bandwidth_bps: u32,

    #[cfg(feature = "memory-diagnostics")]
    _memory_guard: rvoip_infra_common::memory_diagnostics::ObjectGuard,
    #[cfg(feature = "memory-diagnostics")]
    _sender_channel_guard: rvoip_infra_common::memory_diagnostics::ObjectGuard,
    #[cfg(feature = "memory-diagnostics")]
    _receiver_channel_guard: rvoip_infra_common::memory_diagnostics::ObjectGuard,
    #[cfg(feature = "memory-diagnostics")]
    _event_channel_guard: rvoip_infra_common::memory_diagnostics::ObjectGuard,
}

impl RtpSession {
    /// Create a new RTP session
    pub async fn new(config: RtpSessionConfig) -> Result<Self> {
        Self::new_with_receive_queue(config, true).await
    }

    /// Create a new RTP session for event-driven consumers.
    ///
    /// Packets are still emitted through [`RtpSessionEvent::PacketReceived`],
    /// but they are not duplicated into the polling queue used by
    /// [`RtpSession::receive_packet`].
    pub async fn new_event_driven(config: RtpSessionConfig) -> Result<Self> {
        Self::new_with_receive_queue(config, false).await
    }

    async fn new_with_receive_queue(
        config: RtpSessionConfig,
        receive_queue_enabled: bool,
    ) -> Result<Self> {
        let session_buffer_config = config.session_buffer_config;
        let transport_buffer_config = config.transport_buffer_config;

        // Generate SSRC if not provided
        let ssrc = config.ssrc.unwrap_or_else(|| {
            let mut rng = rand::thread_rng();
            rng.gen::<u32>()
        });

        // Create transport config - respect provided ports!
        let transport_config = RtpTransportConfig {
            local_rtp_addr: config.local_addr,
            local_rtcp_addr: None, // RTCP on same port for now
            symmetric_rtp: true,
            rtcp_mux: true, // Enable RTCP multiplexing by default
            session_id: Some(format!("rtp-session-{}", ssrc)),
            // Don't allocate a new port - use the one provided in config
            use_port_allocator: false,
            buffer_config: transport_buffer_config,
        };

        // Create UDP transport
        let transport = Arc::new(UdpRtpTransport::new(transport_config).await?);

        // Create channels for internal communication.
        let (sender_tx, sender_rx) =
            mpsc::channel(session_buffer_config.sender_channel_capacity.max(1));
        let (receiver_tx, receiver_rx) =
            mpsc::channel(session_buffer_config.receiver_channel_capacity.max(1));
        let (event_tx, _) = broadcast::channel(session_buffer_config.event_channel_capacity.max(1));

        // Create scheduler if needed
        let scheduler = Some(RtpScheduler::new(
            config.clock_rate,
            rand::thread_rng().gen::<u16>(), // Random starting sequence
            rand::thread_rng().gen::<u32>(), // Random starting timestamp
        ));

        // Create RTCP report generator
        let hostname = hostname::get().unwrap_or_else(|_| "unknown".into());
        let hostname_str = hostname.to_string_lossy();
        let cname = format!(
            "{}@{}",
            std::env::var("USER").unwrap_or_else(|_| "user".to_string()),
            hostname_str
        );
        let rtcp_generator = crate::stats::reports::RtcpReportGenerator::new(ssrc, cname);

        let mut session = Self {
            config,
            ssrc,
            transport,
            streams: Arc::new(DashMap::new()),
            scheduler,
            receiver: receiver_rx,
            sender: sender_tx,
            receive_queue_enabled,
            event_tx,
            recv_task: None,
            send_task: None,
            stats: Arc::new(parking_lot::Mutex::new(RtpSessionStats::default())),
            media_sync: None,
            active: false,
            rtcp_generator: Some(rtcp_generator),
            rtcp_task: None,
            bandwidth_bps: 64000, // Default bandwidth: 64 kbps
            #[cfg(feature = "memory-diagnostics")]
            _memory_guard: rvoip_infra_common::memory_diagnostics::ObjectGuard::new(
                "rtp_core.rtp_session",
                std::mem::size_of::<Self>(),
            ),
            #[cfg(feature = "memory-diagnostics")]
            _sender_channel_guard: rvoip_infra_common::memory_diagnostics::ObjectGuard::new(
                "rtp_core.rtp_session.sender_channel_capacity",
                session_buffer_config.sender_channel_capacity * std::mem::size_of::<RtpPacket>(),
            ),
            #[cfg(feature = "memory-diagnostics")]
            _receiver_channel_guard: rvoip_infra_common::memory_diagnostics::ObjectGuard::new(
                "rtp_core.rtp_session.receiver_channel_capacity",
                session_buffer_config.receiver_channel_capacity * std::mem::size_of::<RtpPacket>(),
            ),
            #[cfg(feature = "memory-diagnostics")]
            _event_channel_guard: rvoip_infra_common::memory_diagnostics::ObjectGuard::new(
                "rtp_core.rtp_session.event_broadcast_capacity",
                session_buffer_config.event_channel_capacity
                    * std::mem::size_of::<RtpSessionEvent>(),
            ),
        };

        // Start the session
        session.start(sender_rx, receiver_tx).await?;

        Ok(session)
    }

    /// Start the session tasks
    async fn start(
        &mut self,
        mut sender_rx: mpsc::Receiver<RtpPacket>,
        receiver_tx: mpsc::Sender<RtpPacket>,
    ) -> Result<()> {
        if self.active {
            return Ok(());
        }

        let transport = self.transport.clone();
        let stats_send = self.stats.clone();
        let stats_recv = self.stats.clone();
        let remote_addr = self.config.remote_addr;
        let event_tx_send = self.event_tx.clone();
        let event_tx_recv = self.event_tx.clone();
        let clock_rate = self.config.clock_rate;
        let _payload_type = self.config.payload_type;
        let ssrc = self.ssrc;
        let streams_map = self.streams.clone();
        let _jitter_buffer_enabled = self.config.enable_jitter_buffer;
        let _jitter_size = self.config.jitter_buffer_size.unwrap_or(50);
        let _max_age_ms = self.config.max_packet_age_ms.unwrap_or(200);
        let receive_queue_enabled = self.receive_queue_enabled;

        let media_sync = self.media_sync.clone();

        // If we have a remote address, set it on the transport
        if let Some(addr) = remote_addr {
            // Set the remote RTP address on the UDP transport
            if let Some(t) = transport.as_any().downcast_ref::<UdpRtpTransport>() {
                t.set_remote_rtp_addr(addr).await;
            }
        }

        // Prepare the scheduler's sequence state, but do not start its
        // millisecond polling task. The current send paths route directly to
        // `sender_tx` and only need the shared sequence atomic; no production
        // code uses the scheduler queue. Starting one 1 ms timer per call was
        // measurable CPU load under SIPp fan-out.
        if let Some(scheduler) = &mut self.scheduler {
            let sender_tx = self.sender.clone();
            scheduler.set_sender(sender_tx);

            // Set appropriate timestamp increment based on packet interval
            let interval_ms = 20; // Default 20ms packet interval
            let samples_per_packet = (clock_rate as f64 * (interval_ms as f64 / 1000.0)) as u32;
            scheduler.set_interval(interval_ms, samples_per_packet);
        }

        // Start sending task
        let send_transport = transport.clone();
        let send_task = spawn_memory_tracked("rtp_core.rtp_session.send_task", async move {
            let mut last_remote_addr = remote_addr;
            let mut rtp_send_buffer = BytesMut::with_capacity(crate::DEFAULT_MAX_PACKET_SIZE);

            while let Some(packet) = sender_rx.recv().await {
                // Always try to get the current remote address from transport first
                let dest =
                    if let Some(t) = send_transport.as_any().downcast_ref::<UdpRtpTransport>() {
                        // Check transport for current remote address
                        match t.remote_rtp_addr().await {
                            Some(addr) => {
                                // Update our cached value
                                last_remote_addr = Some(addr);
                                addr
                            }
                            None => {
                                // Fall back to cached value if transport doesn't have one
                                if let Some(addr) = last_remote_addr {
                                    addr
                                } else {
                                    // No destination address, can't send
                                    warn!("No destination address for RTP packet, dropping");
                                    continue;
                                }
                            }
                        }
                    } else {
                        // Not a UDP transport, use cached value
                        if let Some(addr) = last_remote_addr {
                            addr
                        } else {
                            // No destination address, can't send
                            warn!("No destination address for RTP packet, dropping");
                            continue;
                        }
                    };

                // Send the packet
                debug!(
                    "Sending RTP packet to {} (seq={}, timestamp={})",
                    dest, packet.header.sequence_number, packet.header.timestamp
                );

                let send_result =
                    if let Some(t) = send_transport.as_any().downcast_ref::<UdpRtpTransport>() {
                        t.send_rtp_with_buffer(&packet, dest, &mut rtp_send_buffer)
                            .await
                    } else {
                        send_transport.send_rtp(&packet, dest).await
                    };

                if let Err(e) = send_result {
                    error!("Failed to send RTP packet: {}", e);

                    // Broadcast error event
                    let _ = event_tx_send.send(RtpSessionEvent::Error(e));
                    continue;
                }

                debug!("Successfully sent RTP packet to {}", dest);

                // Update stats
                {
                    let mut session_stats = stats_send.lock();
                    session_stats.packets_sent += 1;
                    session_stats.bytes_sent += packet.size() as u64;
                }
            }
        });

        // Start receiving task
        let recv_transport = transport.clone();

        // Subscribe to transport events to handle RTCP packets
        let mut transport_events = recv_transport.subscribe();

        let recv_task = spawn_memory_tracked("rtp_core.rtp_session.recv_task", async move {
            // IMPORTANT: Only handle events from transport, no direct packet reception
            // to avoid race conditions where two tasks read from the same socket
            loop {
                match transport_events.recv().await {
                    Ok(crate::traits::RtpEvent::RtcpReceived { data, source: _ }) => {
                        // Try to parse the RTCP packet
                        if let Ok(rtcp_packet) = crate::packet::rtcp::RtcpPacket::parse(&data) {
                            // Handle the RTCP packet based on its type
                            match rtcp_packet {
                                crate::packet::rtcp::RtcpPacket::Goodbye(bye) => {
                                    // Extract the SSRC and reason
                                    if !bye.sources.is_empty() {
                                        let source_ssrc = bye.sources[0];

                                        // Broadcast BYE event
                                        let _ = event_tx_recv.send(RtpSessionEvent::Bye {
                                            ssrc: source_ssrc,
                                            reason: bye.reason,
                                        });

                                        info!("Received RTCP BYE from SSRC={:08x}", source_ssrc);
                                    }
                                }
                                crate::packet::rtcp::RtcpPacket::SenderReport(sr) => {
                                    // Process sender report
                                    let report_ssrc = sr.ssrc;

                                    debug!("Received RTCP SR from SSRC={:08x}", report_ssrc);

                                    // Update stream statistics if this stream exists
                                    if let Some(mut stream) = streams_map.get_mut(&report_ssrc) {
                                        // Update the stream's RTCP SR info
                                        // This will be used for calculating round-trip time
                                        stream.update_last_sr_info(
                                            sr.ntp_timestamp.to_u32(),
                                            std::time::Instant::now(),
                                        );

                                        debug!(
                                            "Updated RTCP SR info for stream SSRC={:08x}",
                                            report_ssrc
                                        );
                                    }

                                    // If media sync is enabled, update it
                                    if let Some(sync) = &media_sync {
                                        if let Ok(mut media_sync) = sync.write() {
                                            // Update synchronization data
                                            media_sync.update_from_sr(
                                                report_ssrc,
                                                sr.ntp_timestamp,
                                                sr.rtp_timestamp,
                                            );
                                        }
                                    }

                                    // Emit SR event for external processing
                                    let _ = event_tx_recv.send(RtpSessionEvent::RtcpSenderReport {
                                        ssrc: report_ssrc,
                                        ntp_timestamp: sr.ntp_timestamp,
                                        rtp_timestamp: sr.rtp_timestamp,
                                        packet_count: sr.sender_packet_count,
                                        octet_count: sr.sender_octet_count,
                                        report_blocks: sr.report_blocks,
                                    });
                                }
                                crate::packet::rtcp::RtcpPacket::ReceiverReport(rr) => {
                                    // Process receiver report
                                    let report_ssrc = rr.ssrc;

                                    debug!(
                                        "Received RTCP RR from SSRC={:08x} with {} report blocks",
                                        report_ssrc,
                                        rr.report_blocks.len()
                                    );

                                    // If there's a report block about our SSRC, process it
                                    for block in &rr.report_blocks {
                                        if block.ssrc == ssrc {
                                            debug!(
                                                "Processing report block about our SSRC={:08x}",
                                                ssrc
                                            );

                                            // Update session stats with packet loss info
                                            {
                                                let mut stats = stats_recv.lock();
                                                stats.packets_lost = block.cumulative_lost as u64;

                                                // Calculate packet loss percentage
                                                let fraction_lost =
                                                    block.fraction_lost as f64 / 256.0;
                                                debug!(
                                                    "Packet loss: {}% (fraction={})",
                                                    fraction_lost * 100.0,
                                                    block.fraction_lost
                                                );
                                            }
                                        }
                                    }

                                    // Emit RR event for external processing
                                    let _ =
                                        event_tx_recv.send(RtpSessionEvent::RtcpReceiverReport {
                                            ssrc: report_ssrc,
                                            report_blocks: rr.report_blocks,
                                        });
                                }
                                // Handle other RTCP packet types as needed
                                _ => {
                                    // For now, we're just logging other packet types
                                    trace!("Received RTCP packet: {:?}", rtcp_packet);
                                }
                            }
                        } else {
                            warn!("Failed to parse RTCP packet");
                        }
                    }
                    Ok(crate::traits::RtpEvent::MediaReceived {
                        payload_type,
                        sequence_number,
                        timestamp,
                        payload,
                        source,
                        ssrc: ssrc_from_event,
                        marker,
                        ..
                    }) => {
                        // Handle RTP packets received via transport events
                        // This is the ONLY path for RTP packets to avoid race conditions

                        // Reconstruct minimal RTP header for processing
                        let header = RtpHeader {
                            version: 2,
                            padding: false,
                            extension: false,
                            cc: 0,
                            marker,
                            payload_type,
                            sequence_number,
                            timestamp,
                            ssrc,
                            csrc: vec![],
                            extensions: None,
                        };

                        let packet = RtpPacket {
                            header,
                            payload: payload.clone(),
                        };

                        // Update stats
                        {
                            let mut session_stats = stats_recv.lock();
                            session_stats.packets_received += 1;
                            session_stats.bytes_received += payload.len() as u64 + 12; // payload + header
                            session_stats.remote_addr = Some(source);
                        }

                        // Use the SSRC from the event
                        let packet_ssrc = ssrc_from_event;

                        // Get or create the stream for this SSRC. The
                        // `entry` runs the closure exactly once per
                        // first insert, so `created` flips iff this
                        // packet's SSRC has never been seen — that's
                        // also the signal for the `NewStreamDetected`
                        // event downstream. The shard guard is dropped
                        // before we forward the packet.
                        let (is_new_stream, output_packet) = {
                            let mut created = false;
                            {
                                let _entry = streams_map.entry(packet_ssrc).or_insert_with(|| {
                                    created = true;
                                    info!("New RTP stream detected with SSRC={:08x}", packet_ssrc);
                                    RtpStream::new(packet_ssrc, clock_rate)
                                });
                            }
                            (created, Some(packet.clone()))
                        };

                        // If this is a new stream, emit the NewStreamDetected event
                        if is_new_stream {
                            let _ = event_tx_recv
                                .send(RtpSessionEvent::NewStreamDetected { ssrc: packet_ssrc });
                        }

                        // Forward the packet
                        if let Some(output) = output_packet {
                            if receive_queue_enabled {
                                match receiver_tx.try_send(output.clone()) {
                                    Ok(()) => {}
                                    Err(mpsc::error::TrySendError::Full(_)) => {
                                        trace!(
                                            "RTP receive polling queue full; dropping duplicate packet"
                                        );
                                    }
                                    Err(mpsc::error::TrySendError::Closed(_)) => {
                                        error!(
                                            "Failed to forward RTP packet to receiver: channel closed"
                                        );
                                    }
                                }
                            }

                            // Broadcast packet received event
                            let _ = event_tx_recv.send(RtpSessionEvent::PacketReceived(output));
                        }
                    }
                    Ok(crate::traits::RtpEvent::Error(e)) => {
                        error!("Transport error: {}", e);
                        let _ = event_tx_recv.send(RtpSessionEvent::Error(e));
                    }
                    Ok(crate::traits::RtpEvent::DtmfEvent {
                        event,
                        end_of_event,
                        volume,
                        duration,
                        timestamp,
                        ssrc,
                        ..
                    }) => {
                        // RFC 4733: forward as a typed session event so
                        // media-core's RTP handler can bubble the digit
                        // up to session-core without re-parsing the
                        // 4-byte body.
                        let _ = event_tx_recv.send(RtpSessionEvent::DtmfReceived {
                            event,
                            end_of_event,
                            volume,
                            duration,
                            timestamp,
                            ssrc,
                        });
                    }
                    Err(e) => {
                        debug!("Transport event channel error: {}", e);
                    }
                }
            }
        });

        // Start RTCP sending task if we have a remote address and report generator
        if let (Some(remote_addr), Some(mut rtcp_generator)) =
            (self.config.remote_addr, self.rtcp_generator.take())
        {
            let transport = self.transport.clone();
            let ssrc = self.ssrc;
            let event_tx = self.event_tx.clone();
            let stats = self.stats.clone();
            let active_state = Arc::new(tokio::sync::Mutex::new(true));
            let _active_state_clone = active_state.clone();
            let bandwidth = self.bandwidth_bps;

            // Set bandwidth in the generator
            rtcp_generator.set_bandwidth(bandwidth);

            // Start the RTCP task
            let rtcp_task = spawn_memory_tracked("rtp_core.rtp_session.rtcp_task", async move {
                debug!("RTCP scheduling task started");

                // Initial interval calculation
                let mut interval = rtcp_generator.calculate_interval();
                debug!("Initial RTCP interval: {:?}", interval);

                while *active_state.lock().await {
                    // Wait for the calculated interval
                    tokio::time::sleep(interval).await;

                    // Check if we should continue
                    if !*active_state.lock().await {
                        break;
                    }

                    // Update RTP statistics before sending the report
                    {
                        let session_stats = stats.lock();
                        rtcp_generator.update_sent_stats(
                            session_stats.packets_sent as u32,
                            session_stats.bytes_sent as u32,
                        );

                        // Log the current stats for debugging
                        debug!(
                            "Current stats for RTCP report: packets={}, bytes={}",
                            session_stats.packets_sent, session_stats.bytes_sent
                        );
                    }

                    // Send an RTCP report regardless of should_send_report logic for this example
                    // We'll send a compound packet with SR and SDES
                    debug!("Sending RTCP report");

                    // Generate sender report
                    let rtp_timestamp = std::time::SystemTime::now()
                        .duration_since(std::time::UNIX_EPOCH)
                        .unwrap_or_default()
                        .as_millis() as u32;

                    let sr = rtcp_generator.generate_sender_report(rtp_timestamp);
                    let sdes = rtcp_generator.generate_sdes();

                    // Create compound packet
                    let mut compound = crate::packet::rtcp::RtcpCompoundPacket::new_with_sr(sr);
                    compound.add_sdes(sdes);

                    // Send the compound packet
                    if let Ok(data) = compound.serialize() {
                        if let Err(e) = transport.send_rtcp_bytes(&data, remote_addr).await {
                            warn!("Failed to send RTCP compound packet: {}", e);
                        } else {
                            info!("Sent RTCP compound packet of {} bytes", data.len());

                            // Emit SR event
                            if let Some(sr) = compound.get_sr() {
                                let _ = event_tx.send(RtpSessionEvent::RtcpSenderReport {
                                    ssrc,
                                    ntp_timestamp: sr.ntp_timestamp,
                                    rtp_timestamp: sr.rtp_timestamp,
                                    packet_count: sr.sender_packet_count,
                                    octet_count: sr.sender_octet_count,
                                    report_blocks: sr.report_blocks.clone(),
                                });
                            }
                        }
                    }

                    // Recalculate interval for next report
                    interval = rtcp_generator.calculate_interval();
                    debug!("Next RTCP report in {:?}", interval);
                }

                debug!("RTCP scheduling task ended");
            });

            self.rtcp_task = Some(rtcp_task);
        }

        self.recv_task = Some(recv_task);
        self.send_task = Some(send_task);
        self.active = true;

        info!("Started RTP session with SSRC={:08x}", ssrc);
        Ok(())
    }

    /// Send an RTP packet with payload. Now `&self` — sequence
    /// numbers are managed by an atomic shared with the scheduler,
    /// and the `sender` mpsc clone is intrinsically `Send + Sync`,
    /// so this no longer requires exclusive borrow. Lets concurrent
    /// callers (audio TX, DTMF transmitter, bridge forwarder) send
    /// without serialising on `Arc<Mutex<RtpSession>>`.
    pub async fn send_packet(
        &self,
        timestamp: RtpTimestamp,
        payload: Bytes,
        marker: bool,
    ) -> Result<()> {
        self.send_packet_with_pt(timestamp, payload, marker, self.config.payload_type)
            .await
    }

    /// Send an RTP packet overriding the configured payload type.
    ///
    /// Needed for RFC 4733 telephone-event (DTMF) transmission — the
    /// session's `config.payload_type` is the audio codec PT (0/8/etc),
    /// but DTMF rides on a distinct PT (typically 101). All other
    /// fields (SSRC, marker, timestamp) follow the same rules as
    /// [`send_packet`](Self::send_packet).
    pub async fn send_packet_with_pt(
        &self,
        timestamp: RtpTimestamp,
        payload: Bytes,
        marker: bool,
        payload_type: u8,
    ) -> Result<()> {
        // The whole point of this method is that the caller controls
        // PT + timestamp explicitly — RFC 4733 telephone-event needs
        // every packet of a tone to share the start timestamp, and
        // the scheduler's `schedule_packet` would overwrite it with
        // its audio-rate cursor. So we bypass the scheduler's
        // queueing path and route directly to the sender channel.
        // Sequence numbers still come from the scheduler (when
        // present) so DTMF + audio share the seq-number space the
        // peer expects.
        let sequence = self
            .scheduler
            .as_ref()
            .map(|s| s.next_sequence())
            .unwrap_or(0);
        let mut header = RtpHeader::new(payload_type, sequence, timestamp, self.ssrc);
        header.marker = marker;
        let packet = RtpPacket::new(header, payload);

        self.sender
            .send(packet)
            .await
            .map_err(|_| Error::SessionError("Failed to send packet".to_string()))
    }

    /// Get a lock-free send handle for this session.
    ///
    /// `RtpSendHandle` is `Send + Sync + Clone` and bypasses the
    /// outer `Arc<Mutex<RtpSession>>` that wraps this session in
    /// media-core. It shares the same sequence atomic as the
    /// scheduler, so the wire-side sees one monotonic seq number
    /// space across both the audio TX path and any scheduler /
    /// `send_packet` call.
    pub fn send_handle(&self) -> Option<RtpSendHandle> {
        let scheduler = self.scheduler.as_ref()?;
        Some(RtpSendHandle {
            sender: self.sender.clone(),
            ssrc: self.ssrc,
            sequence: scheduler.sequence_handle(),
            default_payload_type: self.config.payload_type,
        })
    }

    /// Receive an RTP packet
    pub async fn receive_packet(&mut self) -> Result<RtpPacket> {
        self.receiver
            .recv()
            .await
            .ok_or_else(|| Error::SessionError("Receiver channel closed".to_string()))
    }

    /// Get the session statistics
    pub fn get_stats(&self) -> RtpSessionStats {
        self.stats.lock().clone()
    }

    /// Get current bounded-queue occupancy for leak/perf diagnostics.
    pub fn queue_diagnostics(&self) -> RtpSessionQueueDiagnostics {
        let sender_capacity_packets = self.sender.max_capacity();
        let (receiver_queue_packets, receiver_capacity_packets) = if self.receive_queue_enabled {
            (self.receiver.len(), self.receiver.max_capacity())
        } else {
            (0, 0)
        };
        RtpSessionQueueDiagnostics {
            sender_queue_packets: sender_capacity_packets.saturating_sub(self.sender.capacity()),
            sender_capacity_packets,
            receiver_queue_packets,
            receiver_capacity_packets,
            event_queue_events: self.event_tx.len(),
            event_receiver_count: self.event_tx.receiver_count(),
            #[cfg(feature = "memory-diagnostics")]
            stream_count: self.streams.len(),
        }
    }

    /// Set the remote address
    pub async fn set_remote_addr(&mut self, addr: SocketAddr) {
        self.config.remote_addr = Some(addr);

        // Update stats with remote address
        {
            let mut stats = self.stats.lock();
            stats.remote_addr = Some(addr);
        }

        // Update the transport's remote address
        if let Some(t) = self.transport.as_any().downcast_ref::<UdpRtpTransport>() {
            t.set_remote_rtp_addr(addr).await;
        }
    }

    /// Get the local address
    pub fn local_addr(&self) -> Result<SocketAddr> {
        self.transport.local_rtp_addr()
    }

    /// Get the transport
    pub fn transport(&self) -> Arc<dyn RtpTransport> {
        self.transport.clone()
    }

    /// Close the session and clean up resources
    pub async fn close(&mut self) -> Result<()> {
        // Send BYE packet if we have a remote address
        if let Some(remote_addr) = self.config.remote_addr {
            // Create BYE packet
            let bye = crate::packet::rtcp::RtcpGoodbye::new_with_reason(
                self.ssrc,
                "Session closed".to_string(),
            );

            // Create RTCP packet
            let rtcp_packet = crate::packet::rtcp::RtcpPacket::Goodbye(bye);

            // Serialize and send
            match rtcp_packet.serialize() {
                Ok(data) => {
                    // Send using transport (through RTCP port if available)
                    if let Err(e) = self.transport.send_rtcp_bytes(&data, remote_addr).await {
                        warn!("Failed to send RTCP BYE: {}", e);
                    }
                }
                Err(e) => {
                    warn!("Failed to serialize RTCP BYE: {}", e);
                }
            }
        }

        // Stop the scheduler if running
        if let Some(scheduler) = &mut self.scheduler {
            scheduler.stop().await;
        }

        // Stop the receive task
        if let Some(handle) = self.recv_task.take() {
            handle.abort();
            let _ = handle.await;
        }

        // Stop the send task
        if let Some(handle) = self.send_task.take() {
            handle.abort();
            let _ = handle.await;
        }

        // Stop the RTCP task
        if let Some(handle) = self.rtcp_task.take() {
            handle.abort();
            let _ = handle.await;
        }

        // Close the transport
        let _ = self.transport.close().await;

        self.active = false;
        info!("Closed RTP session with SSRC={:08x}", self.ssrc);

        Ok(())
    }

    /// Get the current timestamp
    pub fn get_timestamp(&self) -> RtpTimestamp {
        if let Some(scheduler) = &self.scheduler {
            scheduler.get_timestamp()
        } else {
            // Generate based on uptime if no scheduler
            let now = std::time::SystemTime::now();
            let since_epoch = now
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap_or_else(|_| Duration::from_secs(0));

            let secs = since_epoch.as_secs();
            let nanos = since_epoch.subsec_nanos();

            // Convert to timestamp units (samples)
            let timestamp_secs = secs * (self.config.clock_rate as u64);
            let timestamp_fraction =
                ((nanos as u64) * (self.config.clock_rate as u64)) / 1_000_000_000;

            (timestamp_secs + timestamp_fraction) as u32
        }
    }

    /// Current RTP timestamp cursor — the timestamp the next audio
    /// packet would carry. Coherent with the audio stream's SSRC per
    /// RFC 4733 §2.1: telephone-event packets share the start
    /// timestamp of the surrounding audio so receivers can align
    /// tones with the audio they overlay.
    ///
    /// The implementation derives the timestamp from wall-clock at
    /// the configured clock rate rather than reading the scheduler's
    /// internal `self.timestamp` field directly. This matters because:
    ///
    /// - When audio packets are flowing through the scheduler at the
    ///   audio rate, wall-clock and scheduler cursor stay in lockstep
    ///   (both advance at `clock_rate` Hz), so the returned value is
    ///   audio-anchored as RFC 4733 expects.
    /// - When no audio is flowing (e.g. the streampeer/dtmf example,
    ///   which exercises only RTP-control with PT 101 and never
    ///   pushes a PCMU audio source), the scheduler's `self.timestamp`
    ///   is frozen. A frozen timestamp would collapse successive DTMF
    ///   tones into one `(peer, ssrc, ts)` dedup key at the receiver,
    ///   silently dropping every digit after the first. Wall-clock
    ///   keeps successive tones distinct unconditionally.
    pub fn current_timestamp(&self) -> RtpTimestamp {
        let now = std::time::SystemTime::now();
        let since_epoch = now
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_else(|_| Duration::from_secs(0));
        let secs = since_epoch.as_secs();
        let nanos = since_epoch.subsec_nanos();
        let timestamp_secs = secs * (self.config.clock_rate as u64);
        let timestamp_fraction = ((nanos as u64) * (self.config.clock_rate as u64)) / 1_000_000_000;
        (timestamp_secs + timestamp_fraction) as u32
    }

    /// Get the SSRC of this session
    pub fn get_ssrc(&self) -> RtpSsrc {
        self.ssrc
    }

    /// Subscribe to session events
    pub fn subscribe(&self) -> broadcast::Receiver<RtpSessionEvent> {
        self.event_tx.subscribe()
    }

    /// Get the current payload type
    pub fn get_payload_type(&self) -> u8 {
        self.config.payload_type
    }

    /// Set the payload type
    pub fn set_payload_type(&mut self, payload_type: u8) {
        self.config.payload_type = payload_type;
    }

    /// Get a stream by SSRC, if it exists
    pub async fn get_stream(&self, ssrc: RtpSsrc) -> Option<RtpStreamStats> {
        self.streams.get(&ssrc).map(|stream| stream.get_stats())
    }

    /// Get a list of all current streams
    pub async fn get_all_streams(&self) -> Vec<RtpStreamStats> {
        self.streams
            .iter()
            .map(|entry| entry.value().get_stats())
            .collect()
    }

    /// Get the number of active streams
    pub async fn stream_count(&self) -> usize {
        self.streams.len()
    }

    /// Get a list of all SSRCs known to this session
    ///
    /// This returns all SSRCs that have been seen, even if their streams
    /// haven't released any packets from their jitter buffers yet.
    pub async fn get_all_ssrcs(&self) -> Vec<RtpSsrc> {
        self.streams.iter().map(|entry| *entry.key()).collect()
    }

    /// Force creation of a stream for a specific SSRC
    ///
    /// This is useful when we want to ensure a stream exists for an SSRC
    /// even if no packets have been received yet.
    pub async fn create_stream_for_ssrc(&mut self, ssrc: RtpSsrc) -> bool {
        // Check if this SSRC already exists. The contains_key + insert
        // pair has a benign race (two callers may both decide "new" and
        // race the insert), but we only need a stable per-SSRC entry —
        // DashMap's `entry()` arbitrates.
        if self.streams.contains_key(&ssrc) {
            debug!("Stream for SSRC={:08x} already exists", ssrc);
            return false;
        }

        // Create the stream
        info!("Manually creating new RTP stream for SSRC={:08x}", ssrc);
        let stream = if self.config.enable_jitter_buffer {
            debug!("Creating stream with jitter buffer for SSRC={:08x}", ssrc);
            RtpStream::with_jitter_buffer(
                ssrc,
                self.config.clock_rate,
                self.config.jitter_buffer_size.unwrap_or(50),
                self.config.max_packet_age_ms.unwrap_or(200) as u64,
            )
        } else {
            debug!(
                "Creating stream without jitter buffer for SSRC={:08x}",
                ssrc
            );
            RtpStream::new(ssrc, self.config.clock_rate)
        };

        // The contains_key check above is racy w.r.t. the recv hot
        // path also inserting on first packet; `entry()` arbitrates.
        // The closure runs only on first insert, so `closure_ran`
        // tells us whether *we* created the entry or lost the race.
        let mut closure_ran = false;
        {
            let _entry = self.streams.entry(ssrc).or_insert_with(|| {
                closure_ran = true;
                stream
            });
        }
        if !closure_ran {
            return false;
        }

        // Emit the new stream event
        debug!("Emitting NewStreamDetected event for SSRC={:08x}", ssrc);
        let _ = self
            .event_tx
            .send(RtpSessionEvent::NewStreamDetected { ssrc });

        true
    }

    /// Send an RTCP BYE packet to notify that we're leaving the session
    ///
    /// This can be used to notify other participants that we're leaving the session
    /// without closing the entire RtpSession. The BYE packet includes our SSRC and
    /// an optional reason string.
    ///
    /// Returns an error if serialization fails or if there's no remote address configured.
    pub async fn send_bye(&self, reason: Option<String>) -> Result<()> {
        // Check if we have a remote address
        let remote_addr = match self.config.remote_addr {
            Some(addr) => addr,
            None => {
                return Err(Error::SessionError(
                    "No remote address configured".to_string(),
                ))
            }
        };

        // Create BYE packet
        let bye = crate::packet::rtcp::RtcpGoodbye::new_with_reason(
            self.ssrc,
            reason.unwrap_or_else(|| "Session terminated".to_string()),
        );

        // Create RTCP packet
        let rtcp_packet = crate::packet::rtcp::RtcpPacket::Goodbye(bye);

        // Serialize and send
        match rtcp_packet.serialize() {
            Ok(data) => {
                // Send using transport
                self.transport.send_rtcp_bytes(&data, remote_addr).await
            }
            Err(e) => Err(Error::SerializationError(format!(
                "Failed to serialize RTCP BYE: {}",
                e
            ))),
        }
    }

    /// Send an RTCP Sender Report (SR) packet
    ///
    /// A Sender Report contains:
    /// - Our SSRC
    /// - Current NTP and RTP timestamps
    /// - Packet and octet counts
    /// - Optional report blocks with reception statistics about other sources
    ///
    /// This method generates an SR based on the current session statistics, which is useful
    /// for providing quality metrics to other participants.
    ///
    /// Returns an error if serialization fails or if there's no remote address configured.
    pub async fn send_sender_report(&self) -> Result<()> {
        // Check if we have a remote address
        let remote_addr = match self.config.remote_addr {
            Some(addr) => addr,
            None => {
                return Err(Error::SessionError(
                    "No remote address configured".to_string(),
                ))
            }
        };

        // Get session stats
        let session_stats = self.stats.lock().clone();

        // Create a new SR packet
        let mut sr = crate::packet::rtcp::RtcpSenderReport::new(self.ssrc);

        // Set current NTP timestamp
        sr.ntp_timestamp = crate::packet::rtcp::NtpTimestamp::now();

        // Set current RTP timestamp (convert from NTP time)
        sr.rtp_timestamp = self.get_timestamp();

        // Set packet and octet count from session stats
        sr.sender_packet_count = session_stats.packets_sent as u32;
        sr.sender_octet_count = session_stats.bytes_sent as u32;

        // Add report blocks for active streams (remote SSRCs we're receiving from)
        // Up to 31 streams per RTCP packet.
        for entry in self.streams.iter().take(31) {
            let ssrc = *entry.key();
            let stream_stats = entry.value().get_stats();

            // Create a report block for this source
            let mut block = crate::packet::rtcp::RtcpReportBlock::new(ssrc);

            // Set statistics
            let expected_packets = stream_stats.highest_seq - stream_stats.first_seq + 1;
            let (fraction_lost, cumulative_lost) =
                block.calculate_packet_loss(expected_packets, stream_stats.received);

            block.fraction_lost = fraction_lost;
            block.cumulative_lost = cumulative_lost as u32;
            block.highest_seq = stream_stats.highest_seq;
            block.jitter = stream_stats.jitter;

            // TODO: Set last_sr and delay_since_last_sr when we process incoming SRs

            // Add the block to the SR
            sr.add_report_block(block);
        }

        // **FIX: Update our own MediaSync context with the SR data we're sending**
        // This ensures our own timing data flows into MediaSync for API access
        if let Some(media_sync) = &self.media_sync {
            if let Ok(mut sync) = media_sync.write() {
                sync.update_from_sr(self.ssrc, sr.ntp_timestamp, sr.rtp_timestamp);
                debug!(
                    "Updated MediaSync with our own SR: SSRC={:08x}, NTP={:?}, RTP={}",
                    self.ssrc, sr.ntp_timestamp, sr.rtp_timestamp
                );
            }
        }

        // Create RTCP packet
        let rtcp_packet = crate::packet::rtcp::RtcpPacket::SenderReport(sr);

        // Serialize and send
        match rtcp_packet.serialize() {
            Ok(data) => self.transport.send_rtcp_bytes(&data, remote_addr).await,
            Err(e) => Err(Error::SerializationError(format!(
                "Failed to serialize RTCP SR: {}",
                e
            ))),
        }
    }

    /// Send an RTCP Receiver Report (RR) packet
    ///
    /// A Receiver Report contains:
    /// - Our SSRC
    /// - Report blocks with reception statistics about other sources
    ///
    /// This method generates an RR based on the current stream statistics, which is useful
    /// for providing quality metrics to other participants when we're receiving but not sending.
    ///
    /// Returns an error if serialization fails or if there's no remote address configured.
    pub async fn send_receiver_report(&self) -> Result<()> {
        // Check if we have a remote address
        let remote_addr = match self.config.remote_addr {
            Some(addr) => addr,
            None => {
                return Err(Error::SessionError(
                    "No remote address configured".to_string(),
                ))
            }
        };

        // Create a new RR packet
        let mut rr = crate::packet::rtcp::RtcpReceiverReport::new(self.ssrc);

        // Add report blocks for active streams (remote SSRCs we're receiving from)
        // Up to 31 streams per RTCP packet.
        for entry in self.streams.iter().take(31) {
            let ssrc = *entry.key();
            let stream_stats = entry.value().get_stats();

            // Create a report block for this source
            let mut block = crate::packet::rtcp::RtcpReportBlock::new(ssrc);

            // Set statistics
            let expected_packets = stream_stats.highest_seq - stream_stats.first_seq + 1;
            let (fraction_lost, cumulative_lost) =
                block.calculate_packet_loss(expected_packets, stream_stats.received);

            block.fraction_lost = fraction_lost;
            block.cumulative_lost = cumulative_lost as u32;
            block.highest_seq = stream_stats.highest_seq;
            block.jitter = stream_stats.jitter;

            // TODO: Set last_sr and delay_since_last_sr when we process incoming SRs

            // Add the block to the RR
            rr.add_report_block(block);
        }

        // Create RTCP packet
        let rtcp_packet = crate::packet::rtcp::RtcpPacket::ReceiverReport(rr);

        // Serialize and send
        match rtcp_packet.serialize() {
            Ok(data) => self.transport.send_rtcp_bytes(&data, remote_addr).await,
            Err(e) => Err(Error::SerializationError(format!(
                "Failed to serialize RTCP RR: {}",
                e
            ))),
        }
    }

    /// Enable media synchronization
    pub fn enable_media_sync(&mut self) -> Arc<std::sync::RwLock<crate::sync::MediaSync>> {
        let sync = Arc::new(std::sync::RwLock::new(crate::sync::MediaSync::new()));
        self.media_sync = Some(sync.clone());

        // Register our stream
        if let Ok(mut media_sync) = sync.write() {
            media_sync.register_stream(self.ssrc, self.config.clock_rate);
        }

        sync
    }

    /// Get the media synchronization context
    pub fn media_sync(&self) -> Option<Arc<std::sync::RwLock<crate::sync::MediaSync>>> {
        self.media_sync.clone()
    }

    /// Set the session bandwidth in bits per second
    ///
    /// This affects the RTCP report interval calculation.
    /// Higher bandwidth means more frequent RTCP packets.
    pub fn set_bandwidth(&mut self, bandwidth_bps: u32) {
        self.bandwidth_bps = bandwidth_bps;
    }

    /// Create a sender handle for this session
    ///
    /// This creates a lightweight handle that can be used to send RTP packets
    /// from another thread. This is useful when you need to send packets
    /// but don't want to clone the entire session.
    pub fn create_sender_handle(&self) -> RtpSessionSender {
        RtpSessionSender {
            sender: self.sender.clone(),
            ssrc: self.ssrc,
            payload_type: self.config.payload_type,
            clock_rate: self.config.clock_rate,
        }
    }

    /// Get the UDP socket handle from the transport
    ///
    /// This method is used to access the underlying UDP socket when needed for
    /// other protocols that need to share the same socket (e.g., DTLS).
    pub async fn get_socket_handle(&self) -> Result<Arc<UdpSocket>> {
        // Try to get the socket from the UdpRtpTransport
        if let Some(t) = self.transport.as_any().downcast_ref::<UdpRtpTransport>() {
            // Clone and return the RTP socket using the public method
            let socket = t.get_socket();
            return Ok(socket);
        }

        // If we get here, the transport is not UdpRtpTransport
        Err(Error::Transport(
            "Transport is not a UDP transport".to_string(),
        ))
    }
}

/// A lightweight sender handle for an RTP session
///
/// This handle can be used to send RTP packets to the session
/// from another thread without having to clone the entire session.
#[derive(Clone)]
#[allow(dead_code)] // retained (liveness/Drop hold or reserved); not read
pub struct RtpSessionSender {
    /// Channel for sending packets
    sender: mpsc::Sender<RtpPacket>,

    /// SSRC for this session
    ssrc: RtpSsrc,

    /// Payload type
    payload_type: u8,

    /// Clock rate for the payload type
    #[allow(dead_code)] // retained (liveness/Drop hold or reserved); not read
    clock_rate: u32,
}

impl RtpSessionSender {
    /// Send an RTP packet with payload
    pub async fn send_packet(
        &self,
        timestamp: RtpTimestamp,
        payload: Bytes,
        marker: bool,
    ) -> Result<()> {
        // Create RTP header
        let mut header = RtpHeader::new(
            self.payload_type,
            0, // Sequence number will be set by scheduler
            timestamp,
            self.ssrc,
        );

        // Set marker bit if needed
        header.marker = marker;

        // Create packet
        let packet = RtpPacket::new(header, payload);

        // Send the packet
        self.sender
            .send(packet)
            .await
            .map_err(|_| Error::SessionError("Failed to send packet".to_string()))
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn default_session_buffer_config_preserves_channel_capacities() {
        let config = RtpSessionConfig::default();

        assert_eq!(
            config.session_buffer_config.sender_channel_capacity,
            RTP_SESSION_CHANNEL_CAPACITY
        );
        assert_eq!(
            config.session_buffer_config.receiver_channel_capacity,
            RTP_SESSION_RECEIVE_QUEUE_CAPACITY
        );
        assert_eq!(
            config.session_buffer_config.event_channel_capacity,
            RTP_SESSION_CHANNEL_CAPACITY
        );
        assert_eq!(
            config.transport_buffer_config,
            RtpTransportBufferConfig::default()
        );
    }
}