rvoip-rtp-core 0.2.2

//! Transmit buffer for outbound RTP packets
//!
//! This module provides a high-performance transmit buffer with:
//! - Prioritization of different packet types
//! - Congestion control using RTCP feedback
//! - Buffer limits to prevent memory exhaustion
//! - Efficient packet scheduling

use std::collections::{BTreeMap, VecDeque};
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::{Notify, Semaphore};
use tokio::time::sleep;

use tracing::{debug, trace, warn};

use crate::packet::{rtcp::RtcpPacket, RtpPacket};
use crate::RtpSsrc;

use super::{BufferPool, GlobalBufferManager};

/// Default transmit buffer capacity
pub const DEFAULT_TRANSMIT_BUFFER_CAPACITY: usize = 1000;

/// Default congestion window size (packets)
pub const DEFAULT_CONGESTION_WINDOW: usize = 64;

/// Default minimum retransmission timeout
pub const DEFAULT_MIN_RTO_MS: u64 = 70;

/// Default maximum retransmission timeout
pub const DEFAULT_MAX_RTO_MS: u64 = 1000;

/// Default initial retransmission timeout
pub const DEFAULT_INITIAL_RTO_MS: u64 = 200;

/// Default maximum burst size (packets)
pub const DEFAULT_MAX_BURST: usize = 16;

/// Packet priority levels
#[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord)]
pub enum PacketPriority {
    /// Critical control packets (RTCP BYE, SR, RR)
    Control = 0,

    /// High priority media (e.g., I-frames in video)
    High = 1,

    /// Normal priority media (typical packets)
    Normal = 2,

    /// Low priority (can be dropped first)
    Low = 3,
}

impl Default for PacketPriority {
    fn default() -> Self {
        Self::Normal
    }
}

/// Packet in the transmit queue
#[allow(dead_code)] // retained (liveness/Drop hold or reserved); not read
struct QueuedPacket {
    /// The RTP packet to transmit
    packet: RtpPacket,

    /// When the packet was queued
    queue_time: Instant,

    /// Priority of this packet
    #[allow(dead_code)] // retained (liveness/Drop hold or reserved); not read
    priority: PacketPriority,

    /// Whether this is a retransmission
    #[allow(dead_code)] // retained (liveness/Drop hold or reserved); not read
    is_retransmission: bool,

    /// Metadata for the packet
    metadata: Option<PacketMetadata>,
}

/// Metadata for packet tracking
#[derive(Debug, Clone)]
pub struct PacketMetadata {
    /// Timestamp when packet was first sent
    pub first_send_time: Option<Instant>,

    /// Number of transmission attempts
    pub transmit_count: u32,

    /// Whether the packet was acknowledged
    pub acknowledged: bool,

    /// Last time the packet was sent
    pub last_send_time: Option<Instant>,
}

/// Congestion control state
#[allow(dead_code)] // retained (liveness/Drop hold or reserved); not read
struct CongestionState {
    /// Current congestion window size (packets)
    cwnd: usize,

    /// Slow start threshold
    ssthresh: usize,

    /// Current retransmission timeout (ms)
    rto_ms: u64,

    /// Smoothed round-trip time (ms)
    srtt_ms: Option<f64>,

    /// RTT variation (ms)
    rttvar_ms: Option<f64>,

    /// Last time window was reduced
    #[allow(dead_code)] // retained (liveness/Drop hold or reserved); not read
    last_window_reduction: Instant,

    /// Current estimate of network bandwidth (bps)
    #[allow(dead_code)] // retained (liveness/Drop hold or reserved); not read
    estimated_bps: u64,

    /// Packets in flight
    in_flight: usize,

    /// Lost packets detected
    #[allow(dead_code)] // retained (liveness/Drop hold or reserved); not read
    lost_packets: u64,

    /// Total packets sent
    total_sent: u64,

    /// Last sequence number sent
    last_seq_sent: u16,

    /// Whether we're in slow start
    in_slow_start: bool,
}

impl Default for CongestionState {
    fn default() -> Self {
        Self {
            cwnd: DEFAULT_CONGESTION_WINDOW,
            ssthresh: usize::MAX,
            rto_ms: DEFAULT_INITIAL_RTO_MS,
            srtt_ms: None,
            rttvar_ms: None,
            last_window_reduction: Instant::now(),
            estimated_bps: 1_000_000, // 1 Mbps initial guess
            in_flight: 0,
            lost_packets: 0,
            total_sent: 0,
            last_seq_sent: 0,
            in_slow_start: true,
        }
    }
}

/// Transmit buffer configuration
#[derive(Debug, Clone)]
pub struct TransmitBufferConfig {
    /// Maximum packets to buffer for transmission
    pub max_packets: usize,

    /// Initial congestion window size (packets)
    pub initial_cwnd: usize,

    /// Minimum retransmission timeout (ms)
    pub min_rto_ms: u64,

    /// Maximum retransmission timeout (ms)
    pub max_rto_ms: u64,

    /// Initial retransmission timeout (ms)
    pub initial_rto_ms: u64,

    /// Whether to enable congestion control
    pub congestion_control_enabled: bool,

    /// Maximum burst size (packets)
    pub max_burst: usize,

    /// Whether to prioritize packets
    pub prioritize_packets: bool,

    /// Maximum packet age before dropping (ms)
    pub max_packet_age_ms: u64,

    /// Clock rate for timestamp calculations
    pub clock_rate: u32,
}

impl Default for TransmitBufferConfig {
    fn default() -> Self {
        Self {
            max_packets: DEFAULT_TRANSMIT_BUFFER_CAPACITY,
            initial_cwnd: DEFAULT_CONGESTION_WINDOW,
            min_rto_ms: DEFAULT_MIN_RTO_MS,
            max_rto_ms: DEFAULT_MAX_RTO_MS,
            initial_rto_ms: DEFAULT_INITIAL_RTO_MS,
            congestion_control_enabled: true,
            max_burst: DEFAULT_MAX_BURST,
            prioritize_packets: true,
            max_packet_age_ms: 1000, // 1 second
            clock_rate: 8000,        // Default for audio
        }
    }
}

/// Statistics for the transmit buffer
#[derive(Debug, Clone)]
pub struct TransmitBufferStats {
    /// Current number of packets in the queue
    pub queued_packets: usize,

    /// Total packets sent
    pub packets_sent: u64,

    /// Packets dropped due to queue overflow
    pub packets_dropped_overflow: u64,

    /// Packets dropped due to age
    pub packets_dropped_aged: u64,

    /// Packets retransmitted
    pub packets_retransmitted: u64,

    /// Current congestion window size
    pub cwnd: usize,

    /// Current retransmission timeout (ms)
    pub rto_ms: u64,

    /// Current estimated RTT (ms)
    pub srtt_ms: Option<f64>,

    /// Estimated bandwidth (bps)
    pub estimated_bps: u64,

    /// Packets currently in flight
    pub in_flight: usize,

    /// Packet loss rate (0.0-1.0)
    pub loss_rate: f64,

    /// Buffer fullness percentage (0.0-1.0)
    pub buffer_fullness: f32,

    /// Access for the API layer to these stats
    pub packets_queued: usize,

    /// Total number of packets that have been dropped
    pub packets_dropped: u64,
}

impl Default for TransmitBufferStats {
    fn default() -> Self {
        Self {
            queued_packets: 0,
            packets_sent: 0,
            packets_dropped_overflow: 0,
            packets_dropped_aged: 0,
            packets_retransmitted: 0,
            cwnd: DEFAULT_CONGESTION_WINDOW,
            rto_ms: DEFAULT_INITIAL_RTO_MS,
            srtt_ms: None,
            estimated_bps: 1_000_000, // 1 Mbps initial guess
            in_flight: 0,
            loss_rate: 0.0,
            buffer_fullness: 0.0,
            packets_queued: 0,
            packets_dropped: 0,
        }
    }
}

/// High-performance transmit buffer for RTP packets
///
/// This implementation provides:
/// - Efficient packet queuing and sending
/// - Congestion control and bandwidth estimation
/// - Packet prioritization
/// - RTCP-based feedback handling
/// - Memory management with global limits
#[allow(dead_code)] // retained (liveness/Drop hold or reserved); not read
pub struct TransmitBuffer {
    /// Configuration
    config: TransmitBufferConfig,

    /// Priority queues for packets
    /// Lower priority value = higher priority (will be sent first)
    queues: BTreeMap<PacketPriority, VecDeque<QueuedPacket>>,

    /// Congestion control state
    congestion: CongestionState,

    /// Statistics
    stats: TransmitBufferStats,

    /// Reference to global buffer manager
    buffer_manager: Option<Arc<GlobalBufferManager>>,

    /// Notification for new packets
    packet_notify: Arc<Notify>,

    /// Sequence -> packet metadata map for tracking
    packet_tracking: BTreeMap<u16, PacketMetadata>,

    /// Permits for congestion window
    cwnd_semaphore: Arc<Semaphore>,

    /// Buffer for efficient packet ordering
    packet_buffer: Option<Arc<BufferPool>>,

    /// SSRC for this transmit buffer
    #[allow(dead_code)] // retained (liveness/Drop hold or reserved); not read
    ssrc: RtpSsrc,

    /// Last packet send time
    last_send_time: Option<Instant>,

    /// Pacing state
    pacing_interval_us: u64,
}

impl TransmitBuffer {
    /// Create a new transmit buffer
    pub fn new(ssrc: RtpSsrc, config: TransmitBufferConfig) -> Self {
        let mut queues = BTreeMap::new();

        // Initialize priority queues
        queues.insert(PacketPriority::Control, VecDeque::with_capacity(100));
        queues.insert(
            PacketPriority::High,
            VecDeque::with_capacity(config.max_packets / 4),
        );
        queues.insert(
            PacketPriority::Normal,
            VecDeque::with_capacity(config.max_packets / 2),
        );
        queues.insert(
            PacketPriority::Low,
            VecDeque::with_capacity(config.max_packets / 4),
        );

        // Initialize congestion state
        let mut congestion = CongestionState::default();
        congestion.cwnd = config.initial_cwnd;
        congestion.rto_ms = config.initial_rto_ms;

        let stats = TransmitBufferStats {
            queued_packets: 0,
            packets_sent: 0,
            packets_dropped_overflow: 0,
            packets_dropped_aged: 0,
            packets_retransmitted: 0,
            cwnd: config.initial_cwnd,
            rto_ms: config.initial_rto_ms,
            srtt_ms: None,
            estimated_bps: 1_000_000, // 1 Mbps initial guess
            in_flight: 0,
            loss_rate: 0.0,
            buffer_fullness: 0.0,
            packets_queued: 0,
            packets_dropped: 0,
        };

        // Create congestion window semaphore
        let cwnd_semaphore = Arc::new(Semaphore::new(config.initial_cwnd));

        Self {
            config,
            queues,
            congestion,
            stats,
            buffer_manager: None,
            packet_notify: Arc::new(Notify::new()),
            packet_tracking: BTreeMap::new(),
            cwnd_semaphore,
            packet_buffer: None,
            ssrc,
            last_send_time: None,
            pacing_interval_us: 0,
        }
    }

    /// Create a new transmit buffer with global buffer management
    pub fn with_buffer_manager(
        ssrc: RtpSsrc,
        config: TransmitBufferConfig,
        buffer_manager: Arc<GlobalBufferManager>,
        packet_buffer: Arc<BufferPool>,
    ) -> Self {
        let mut buffer = Self::new(ssrc, config);
        buffer.buffer_manager = Some(buffer_manager);
        buffer.packet_buffer = Some(packet_buffer);
        buffer
    }

    /// Queue a packet for transmission
    ///
    /// Returns true if the packet was queued, false if it was dropped.
    pub async fn queue_packet(&mut self, packet: RtpPacket, priority: PacketPriority) -> bool {
        // Check if buffer has reached maximum capacity
        let total_packets = self.total_queued_packets();

        // Fast reject if at max capacity and priority isn't high enough to drop other packets
        if total_packets >= self.config.max_packets {
            // For normal or low priority packets, drop if we're at capacity
            // High priority packets can try to make room by dropping lower priority ones
            if priority != PacketPriority::High {
                // Normal or low priority when buffer is full - drop immediately
                trace!(
                    "Buffer full ({}/{}), dropping {:?} priority packet with seq={}",
                    total_packets,
                    self.config.max_packets,
                    priority,
                    packet.header.sequence_number
                );
                self.stats.packets_dropped_overflow += 1;
                return false;
            } else {
                // High priority packet - try to drop something to make room
                if !self.drop_low_priority_packets() {
                    // If there's nothing to drop, drop this packet too
                    trace!("Buffer full, nowhere to make room for high priority packet");
                    self.stats.packets_dropped_overflow += 1;
                    return false;
                }
            }
        }

        // At this point we have room (or made room) in the buffer

        // Create packet metadata for tracking
        let metadata = PacketMetadata {
            first_send_time: None,
            transmit_count: 0,
            acknowledged: false,
            last_send_time: None,
        };

        // Create queued packet entry
        let queued = QueuedPacket {
            packet,
            queue_time: Instant::now(),
            priority,
            is_retransmission: false,
            metadata: Some(metadata),
        };

        // Get or create queue for this priority
        let queue = self
            .queues
            .entry(priority)
            .or_insert_with(|| VecDeque::new());

        // Add to queue
        queue.push_back(queued);

        // Update stats
        self.stats.queued_packets = self.total_queued_packets();

        // Notify waiters
        self.packet_notify.notify_one();

        true
    }

    /// Schedule a packet retransmission
    ///
    /// Returns true if the retransmission was scheduled, false otherwise.
    pub async fn schedule_retransmission(&mut self, seq: u16) -> bool {
        // Check if we're tracking this packet
        if let Some(metadata) = self.packet_tracking.get_mut(&seq) {
            // Only retransmit if not already acknowledged
            if !metadata.acknowledged {
                // Find the packet to retransmit (we may still have it)
                // Look through all queues in order of priority
                for (_priority, queue) in self.queues.iter() {
                    for queued_packet in queue {
                        if queued_packet.packet.header.sequence_number == seq {
                            // Packet is already queued, just update stats
                            trace!("Packet seq={} already queued for retransmission", seq);
                            return true;
                        }
                    }
                }

                // We don't have the packet in queue, need to rebuild it
                // This would require access to the original data
                // In a real implementation, we'd need to cache the packets
                // or have the application rebuild it

                // For now, just log it
                warn!(
                    "Requested retransmission for seq={} but packet not available",
                    seq
                );

                // Update stats
                self.stats.packets_retransmitted += 1;

                return false;
            }
        }

        false
    }

    /// Get the next packet to send
    ///
    /// This handles congestion control if enabled.
    pub async fn get_next_packet(&mut self) -> Option<RtpPacket> {
        // Nothing to send
        if self.total_queued_packets() == 0 {
            return None;
        }

        // If congestion control is not enabled, bypass window checks
        if !self.config.congestion_control_enabled {
            return self.get_packet_without_congestion_control().await;
        }

        // Skip this check if we have no packets in flight yet
        if self.congestion.in_flight > 0 {
            // Check if window is already full
            if self.congestion.in_flight >= self.congestion.cwnd {
                trace!(
                    "Congestion window full ({}/{}), not sending new packets",
                    self.congestion.in_flight,
                    self.congestion.cwnd
                );
                return None;
            }
        }

        // Try to acquire a congestion window permit
        match self.cwnd_semaphore.try_acquire() {
            Ok(permit) => {
                // Permit acquired, we can send
                drop(permit);
            }
            Err(_) => {
                // No permits available, congestion window is full
                trace!("No congestion permits available, not sending");
                return None;
            }
        }

        // Apply pacing if needed
        if self.pacing_interval_us > 0 {
            self.apply_pacing().await;
        }

        // Get highest priority packet
        let packet_option = self.dequeue_highest_priority_packet();

        if let Some(_packet) = &packet_option {
            // Update in-flight count
            self.congestion.in_flight += 1;
            self.congestion.total_sent += 1;

            // Update stats
            self.stats.in_flight = self.congestion.in_flight;
            self.stats.packets_sent += 1;

            // Update last send time
            self.last_send_time = Some(Instant::now());
        }

        packet_option
    }

    /// Wait until either a packet is available or timeout occurs
    ///
    /// Returns true if a packet is available, false if timeout occurred.
    pub async fn wait_for_packet(&self, timeout: Duration) -> bool {
        // If we already have packets and congestion window permits, return immediately
        if self.total_queued_packets() > 0 && self.cwnd_semaphore.available_permits() > 0 {
            return true;
        }

        // Wait for notification with timeout
        let notify = self.packet_notify.clone();
        tokio::select! {
            _ = notify.notified() => true,
            _ = tokio::time::sleep(timeout) => false,
        }
    }

    /// Process RTCP feedback packets to update congestion control
    pub fn process_rtcp_feedback(&mut self, rtcp: &RtcpPacket) {
        // Simplified implementation that doesn't rely on specific RTCP methods
        match rtcp {
            RtcpPacket::ReceiverReport(_) => {
                // In a real implementation, we would extract RTT and loss data
                // from the receiver report. For this example, we'll simulate it.

                // Simulate RTT measurement
                let rtt_ms = 50.0; // Assume 50ms RTT
                self.update_rtt_estimate(rtt_ms);

                // Simulate packet loss information
                self.stats.loss_rate = 0.01; // 1% loss rate

                // Update congestion window
                self.update_congestion_window(None);
            }
            RtcpPacket::SenderReport(_) => {
                // Nothing specific to do for sender reports
            }
            _ => {
                // Other RTCP packet types not handled
            }
        }
    }

    /// Signal a packet has been acknowledged (e.g., via RTCP)
    pub fn acknowledge_packet(&mut self, seq: u16) {
        if let Some(metadata) = self.packet_tracking.get_mut(&seq) {
            metadata.acknowledged = true;

            // Update in-flight count if needed
            if self.congestion.in_flight > 0 {
                self.congestion.in_flight -= 1;
                self.stats.in_flight = self.congestion.in_flight;
            }

            // Release a congestion window permit
            self.cwnd_semaphore.add_permits(1);

            // Adjust congestion window for successful transmission
            if self.config.congestion_control_enabled {
                self.update_congestion_window(Some(seq));
            }
        }
    }

    /// Update RTT estimate using RFC 6298 algorithm
    fn update_rtt_estimate(&mut self, rtt_ms: f64) {
        if let (Some(srtt), Some(rttvar)) = (self.congestion.srtt_ms, self.congestion.rttvar_ms) {
            // Update RTTVAR and SRTT
            // RTTVAR = (1 - beta) * RTTVAR + beta * |SRTT - R|
            // SRTT = (1 - alpha) * SRTT + alpha * R
            // where alpha = 1/8 and beta = 1/4

            let alpha = 0.125;
            let beta = 0.25;

            let new_rttvar = (1.0 - beta) * rttvar + beta * (srtt - rtt_ms).abs();
            let new_srtt = (1.0 - alpha) * srtt + alpha * rtt_ms;

            self.congestion.rttvar_ms = Some(new_rttvar);
            self.congestion.srtt_ms = Some(new_srtt);

            // Update RTO based on RFC 6298 formula: RTO = SRTT + 4 * RTTVAR
            let new_rto = new_srtt + 4.0 * new_rttvar;

            // Clamp RTO to min/max values
            let clamped_rto = (new_rto.round() as u64)
                .max(self.config.min_rto_ms)
                .min(self.config.max_rto_ms);

            self.congestion.rto_ms = clamped_rto;
        } else {
            // First RTT measurement
            let srtt = rtt_ms;
            let rttvar = rtt_ms / 2.0;

            self.congestion.srtt_ms = Some(srtt);
            self.congestion.rttvar_ms = Some(rttvar);

            // Initial RTO = SRTT + 4 * RTTVAR
            let new_rto = srtt + 4.0 * rttvar;

            // Clamp RTO to min/max values
            let clamped_rto = (new_rto.round() as u64)
                .max(self.config.min_rto_ms)
                .min(self.config.max_rto_ms);

            self.congestion.rto_ms = clamped_rto;
        }

        // Update stats
        self.stats.srtt_ms = self.congestion.srtt_ms;
        self.stats.rto_ms = self.congestion.rto_ms;
    }

    /// Update congestion window after successful transmission
    fn update_congestion_window(&mut self, seq: Option<u16>) {
        if self.congestion.in_slow_start {
            // In slow start, grow window exponentially
            // Increase by 1 for each ACK
            let new_cwnd = self.congestion.cwnd + 1;

            // Check if we should exit slow start
            if new_cwnd >= self.congestion.ssthresh {
                self.congestion.in_slow_start = false;
            }

            self.congestion.cwnd = new_cwnd;
        } else {
            // In congestion avoidance, grow window linearly
            // Increase by 1/cwnd for each ACK, so cwnd += 1 after a full window
            // We approximate this by adding 1 every cwnd packets
            if let Some(ack_seq) = seq {
                let cwnd = self.congestion.cwnd;
                if ack_seq % (cwnd as u16) == 0 {
                    self.congestion.cwnd += 1;
                }
            }
        }

        // Update pacing
        self.update_pacing();

        // Update stats
        self.stats.cwnd = self.congestion.cwnd;
    }

    /// Calculate and update pacing interval
    fn update_pacing(&mut self) {
        // Rough approximation: pace packets evenly across the RTT or a minimum interval
        if let Some(srtt_ms) = self.congestion.srtt_ms {
            // Convert to microseconds for precision
            let srtt_us = (srtt_ms * 1000.0) as u64;

            // Minimum sensible interval (100 microseconds)
            const MIN_INTERVAL_US: u64 = 100;

            // Calculate pacing interval: RTT / cwnd
            let interval = if self.congestion.cwnd > 0 {
                srtt_us / self.congestion.cwnd as u64
            } else {
                srtt_us
            };

            // Use a reasonable minimum
            self.pacing_interval_us = interval.max(MIN_INTERVAL_US);
        } else {
            // No RTT estimate yet, use a reasonable default
            self.pacing_interval_us = 1000; // 1ms
        }
    }

    /// Apply pacing to avoid bursts
    async fn apply_pacing(&mut self) {
        if self.pacing_interval_us == 0 {
            return;
        }

        if let Some(last_send_time) = self.last_send_time {
            let now = Instant::now();
            let elapsed_us = now.duration_since(last_send_time).as_micros() as u64;

            if elapsed_us < self.pacing_interval_us {
                // Need to wait for pacing
                let wait_us = self.pacing_interval_us - elapsed_us;

                if wait_us > 100 {
                    // Only wait if it's significant (>100µs)
                    sleep(Duration::from_micros(wait_us)).await;
                }
            }
        }

        // Update last send time
        self.last_send_time = Some(Instant::now());
    }

    /// Total number of packets queued across all priorities
    fn total_queued_packets(&self) -> usize {
        self.queues.values().map(|q| q.len()).sum()
    }

    /// Try to drop low priority packets to make room for higher priority ones
    ///
    /// Returns true if packets were dropped, false otherwise.
    fn drop_low_priority_packets(&mut self) -> bool {
        let mut dropped = false;

        // Try low priority first
        if let Some(queue) = self.queues.get_mut(&PacketPriority::Low) {
            if !queue.is_empty() {
                queue.pop_front();
                self.stats.packets_dropped_overflow += 1;
                self.stats.queued_packets = self.total_queued_packets();
                dropped = true;
                trace!("Dropped low priority packet to make room");
                return dropped;
            }
        }

        // If we couldn't drop low priority, try normal priority
        if let Some(queue) = self.queues.get_mut(&PacketPriority::Normal) {
            if !queue.is_empty() {
                queue.pop_front();
                self.stats.packets_dropped_overflow += 1;
                self.stats.queued_packets = self.total_queued_packets();
                dropped = true;
                trace!("Dropped normal priority packet to make room");
                return dropped;
            }
        }

        // Don't drop high priority or control packets
        trace!("No low/normal priority packets available to drop");
        dropped
    }

    /// Dequeue the highest priority packet
    fn dequeue_highest_priority_packet(&mut self) -> Option<RtpPacket> {
        // Try each queue in priority order
        for priority in [
            PacketPriority::Control,
            PacketPriority::High,
            PacketPriority::Normal,
            PacketPriority::Low,
        ] {
            if let Some(queue) = self.queues.get_mut(&priority) {
                if !queue.is_empty() {
                    let queued_packet = queue.pop_front().unwrap();

                    // Update stats
                    self.stats.queued_packets = self.total_queued_packets();

                    // Update metadata and tracking
                    if let Some(mut metadata) = queued_packet.metadata {
                        let now = Instant::now();
                        let seq = queued_packet.packet.header.sequence_number;

                        if metadata.first_send_time.is_none() {
                            metadata.first_send_time = Some(now);
                        }

                        metadata.transmit_count += 1;
                        metadata.last_send_time = Some(now);

                        // Track the packet
                        self.packet_tracking.insert(seq, metadata);

                        // Update last sequence sent
                        self.congestion.last_seq_sent = seq;
                    }

                    return Some(queued_packet.packet);
                }
            }
        }

        None
    }

    /// Get a packet without congestion control
    ///
    /// This is used when congestion control is disabled.
    async fn get_packet_without_congestion_control(&mut self) -> Option<RtpPacket> {
        // Get highest priority packet
        let packet_option = self.dequeue_highest_priority_packet();

        if packet_option.is_some() {
            // Update stats
            self.stats.packets_sent += 1;

            // Update last send time
            self.last_send_time = Some(Instant::now());
        }

        packet_option
    }

    /// Purge expired packets from the queue
    ///
    /// This is useful to periodically call to avoid memory leaks.
    pub fn purge_expired_packets(&mut self) {
        let now = Instant::now();
        let max_age = Duration::from_millis(self.config.max_packet_age_ms);

        // Check each queue
        for queue in self.queues.values_mut() {
            // Remove packets that are too old
            let mut i = 0;
            while i < queue.len() {
                if now.duration_since(queue[i].queue_time) > max_age {
                    queue.remove(i);
                    self.stats.packets_dropped_aged += 1;
                } else {
                    i += 1;
                }
            }
        }

        // Update stats
        self.stats.queued_packets = self.total_queued_packets();
    }

    /// Clear the transmit buffer
    pub fn clear(&mut self) {
        // Clear all queues
        for queue in self.queues.values_mut() {
            queue.clear();
        }

        // Clear packet tracking
        self.packet_tracking.clear();

        // Reset stats
        self.stats.queued_packets = 0;
    }

    /// Get statistics for the transmit buffer
    pub fn get_stats(&self) -> TransmitBufferStats {
        // Calculate additional stats
        let total_capacity = self.config.max_packets;
        let current_queued = self.total_queued_packets();
        let buffer_fullness = if total_capacity > 0 {
            current_queued as f32 / total_capacity as f32
        } else {
            0.0
        };

        // Update the queued_packets count
        let mut stats = self.stats.clone();
        stats.queued_packets = current_queued;
        stats.buffer_fullness = buffer_fullness;
        stats.packets_queued = current_queued;
        stats.packets_dropped = stats.packets_dropped_overflow + stats.packets_dropped_aged;

        stats
    }

    /// Update the configuration of the transmit buffer
    pub fn update_config(&mut self, config: TransmitBufferConfig) {
        // Store current values for comparison
        let _old_max_packets = self.config.max_packets;
        let old_cwnd = self.config.initial_cwnd;
        let old_cc_enabled = self.config.congestion_control_enabled;

        // Update the configuration
        self.config = config;

        // If the congestion window size changed, update the semaphore
        if old_cwnd != self.config.initial_cwnd {
            // Only if not in active congestion control
            if !old_cc_enabled || !self.config.congestion_control_enabled {
                self.congestion.cwnd = self.config.initial_cwnd;

                // Reset semaphore to current window size minus in-flight packets
                let in_flight = self.congestion.in_flight;
                let available = if in_flight < self.congestion.cwnd {
                    self.congestion.cwnd - in_flight
                } else {
                    0
                };

                // Reset semaphore to new permitted value
                self.cwnd_semaphore = Arc::new(Semaphore::new(available));

                // Update stats
                self.stats.cwnd = self.congestion.cwnd;
            }
        }

        // Update pacing
        self.update_pacing();

        debug!(
            "Updated transmit buffer config: max_packets={}, cwnd={}, cc_enabled={}",
            self.config.max_packets, self.congestion.cwnd, self.config.congestion_control_enabled
        );
    }

    /// Set the priority threshold for a specific buffer fullness level
    ///
    /// When the buffer reaches the specified fullness level (0.0-1.0),
    /// only packets with priority greater than or equal to the threshold
    /// will be transmitted.
    pub fn set_priority_threshold(&mut self, buffer_fullness: f32, priority: PacketPriority) {
        // Store this as a configuration option
        debug!("Setting priority threshold: at {:.1}% fullness, only {:?} or higher priority will be sent",
              buffer_fullness * 100.0, priority);

        // We could implement more sophisticated logic here, like
        // storing multiple thresholds for different fullness levels

        // For this simple implementation, we just note it in the log
        // A real implementation would check buffer fullness in get_next_packet
        // and only return packets above the threshold
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::packet::{RtpHeader, RtpPacket};
    use bytes::Bytes;

    fn create_test_packet(seq: u16, ts: u32, ssrc: RtpSsrc) -> RtpPacket {
        let header = RtpHeader::new(96, seq, ts, ssrc);
        let payload = Bytes::from_static(b"test");
        RtpPacket::new(header, payload)
    }

    #[tokio::test]
    async fn test_basic_queuing() {
        let config = TransmitBufferConfig::default();
        let mut buffer = TransmitBuffer::new(0x12345678, config);

        // Queue some packets
        buffer
            .queue_packet(create_test_packet(1, 0, 0x12345678), PacketPriority::Normal)
            .await;

        buffer
            .queue_packet(create_test_packet(2, 160, 0x12345678), PacketPriority::High)
            .await;

        // Get packets - should come in priority order
        let p1 = buffer.get_next_packet().await;
        let p2 = buffer.get_next_packet().await;

        assert!(p1.is_some());
        assert!(p2.is_some());

        // High priority should be first
        assert_eq!(p1.unwrap().header.sequence_number, 2);
        assert_eq!(p2.unwrap().header.sequence_number, 1);
    }

    #[tokio::test]
    async fn test_buffer_overflow() {
        let config = TransmitBufferConfig {
            max_packets: 2,
            ..Default::default()
        };

        let mut buffer = TransmitBuffer::new(0x12345678, config);

        // Queue up to capacity with normal priority
        assert!(
            buffer
                .queue_packet(create_test_packet(1, 0, 0x12345678), PacketPriority::Normal)
                .await,
            "First packet should be queued"
        );

        assert!(
            buffer
                .queue_packet(
                    create_test_packet(2, 160, 0x12345678),
                    PacketPriority::Normal
                )
                .await,
            "Second packet should be queued"
        );

        // Verify buffer state
        let stats = buffer.get_stats();
        assert_eq!(stats.queued_packets, 2, "Buffer should have 2 packets");

        // Third normal priority packet should be rejected as we're at capacity
        assert!(
            !buffer
                .queue_packet(
                    create_test_packet(3, 320, 0x12345678),
                    PacketPriority::Normal
                )
                .await,
            "Third normal packet should be rejected"
        );

        // High priority packet should be accepted by dropping a normal one
        assert!(
            buffer
                .queue_packet(create_test_packet(3, 320, 0x12345678), PacketPriority::High)
                .await,
            "High priority packet should be accepted"
        );

        // Verify we still have max 2 packets
        let stats = buffer.get_stats();
        assert_eq!(
            stats.queued_packets, 2,
            "Buffer should still have 2 packets"
        );

        // We should now have packets 2 and 3 (high priority)
        let p1 = buffer.get_next_packet().await;
        let p2 = buffer.get_next_packet().await;
        let p3 = buffer.get_next_packet().await;

        assert!(
            p1.is_some(),
            "First packet (high priority) should be available"
        );
        assert!(p2.is_some(), "Second packet should be available");
        assert!(p3.is_none(), "Buffer should be empty after 2 packets");

        // High priority should be first
        assert_eq!(
            p1.unwrap().header.sequence_number,
            3,
            "First packet should be the high priority one"
        );
        assert_eq!(
            p2.unwrap().header.sequence_number,
            2,
            "Second packet should be the remaining normal one"
        );
    }

    #[tokio::test]
    async fn test_congestion_control() {
        let config = TransmitBufferConfig {
            initial_cwnd: 2, // Small window for testing
            congestion_control_enabled: true,
            ..Default::default()
        };

        let mut buffer = TransmitBuffer::new(0x12345678, config);

        // Queue several packets
        for i in 1..=5 {
            assert!(
                buffer
                    .queue_packet(
                        create_test_packet(i, (i as u32) * 160, 0x12345678),
                        PacketPriority::Normal
                    )
                    .await,
                "Packet {} should be queued",
                i
            );
        }

        // Verify all 5 packets are queued
        let stats = buffer.get_stats();
        assert_eq!(
            stats.queued_packets, 5,
            "Buffer should have 5 queued packets"
        );

        // But congestion window should only allow sending 2
        let p1 = buffer.get_next_packet().await;
        assert!(p1.is_some(), "First packet should be available");

        let p2 = buffer.get_next_packet().await;
        assert!(p2.is_some(), "Second packet should be available");

        // Verify in-flight count
        let stats = buffer.get_stats();
        assert_eq!(stats.in_flight, 2, "Should have 2 packets in flight");

        // Third packet should be blocked by congestion window
        let p3 = buffer.get_next_packet().await;
        assert!(
            p3.is_none(),
            "Third packet should be blocked by congestion window"
        );

        // Acknowledge one packet to open a window slot
        buffer.acknowledge_packet(1);

        // Verify in-flight count decreased
        let stats = buffer.get_stats();
        assert_eq!(
            stats.in_flight, 1,
            "Should have 1 packet in flight after ACK"
        );

        // Now we should be able to get another packet
        let p3 = buffer.get_next_packet().await;
        assert!(p3.is_some(), "Third packet should be available after ACK");
        assert_eq!(
            p3.unwrap().header.sequence_number,
            3,
            "Third packet should have seq=3"
        );
    }
}