//! Bounded Tokio UDP transport and the C#-compatible transport value types.

#![allow(clippy::missing_errors_doc)] // Result shapes are fixed by the compatibility map.
#![allow(clippy::needless_pass_by_value)] // Owned arrays and objects preserve mapped signatures.

use crate::packets::{Packet, PacketAckPacket, PacketAckPacketPacketsBlock, PacketType};
use crate::{Error, GridClient, Helpers, Simulator};
use libremetaverse_types::compat::{Array, CancellationToken, CancellationTokenSource, Object};
use std::collections::{BTreeMap, HashMap, HashSet, VecDeque};
use std::fmt;
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
use std::panic::{AssertUnwindSafe, catch_unwind};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::mpsc::{Receiver as AckReceiver, SyncSender as AckSender, sync_channel};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::net::UdpSocket;
use tokio::sync::{mpsc, oneshot};
use tokio::task::JoinHandle;
use tokio::time::{Instant, MissedTickBehavior};

const DEFAULT_DECODE_BUFFER_SIZE: usize = 8 * 1024;
const THROTTLE_PERIOD: Duration = Duration::from_millis(100);
const THROTTLE_MIN_BYTES_PER_PERIOD: usize = 200;
const THROTTLE_BURST_PERIODS: usize = 4;

/// A failure at the native UDP transport boundary.
///
/// The variants deliberately carry no datagram bytes, credentials, endpoint
/// query data, or operating-system error strings.
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
#[non_exhaustive]
pub enum UdpTransportError {
    InvalidConfiguration(&'static str),
    InvalidBuffer,
    MtuExceeded,
    NotRunning,
    AlreadyRunning,
    Backpressure,
    ReliableWindowFull,
    TooManyPeers,
    RuntimeUnavailable,
    Cancelled,
    Socket,
}

impl fmt::Display for UdpTransportError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::InvalidConfiguration(field) => {
                write!(formatter, "invalid UDP transport setting {field}")
            }
            Self::InvalidBuffer => formatter.write_str("invalid UDP packet buffer"),
            Self::MtuExceeded => formatter.write_str("UDP payload exceeds the protocol MTU"),
            Self::NotRunning => formatter.write_str("UDP transport is not running"),
            Self::AlreadyRunning => formatter.write_str("UDP transport is already running"),
            Self::Backpressure => formatter.write_str("UDP transport queue is full"),
            Self::ReliableWindowFull => formatter.write_str("UDP reliable-send window is full"),
            Self::TooManyPeers => formatter.write_str("UDP peer limit is reached"),
            Self::RuntimeUnavailable => {
                formatter.write_str("UDP transport requires a current Tokio runtime")
            }
            Self::Cancelled => formatter.write_str("UDP transport was cancelled"),
            Self::Socket => formatter.write_str("UDP socket operation failed"),
        }
    }
}

impl std::error::Error for UdpTransportError {}

impl From<UdpTransportError> for Error {
    fn from(error: UdpTransportError) -> Self {
        match error {
            UdpTransportError::InvalidConfiguration(_)
            | UdpTransportError::InvalidBuffer
            | UdpTransportError::MtuExceeded => Self::Argument,
            UdpTransportError::Cancelled => Self::Cancelled,
            UdpTransportError::Socket => Self::Socket,
            UdpTransportError::NotRunning
            | UdpTransportError::AlreadyRunning
            | UdpTransportError::Backpressure
            | UdpTransportError::ReliableWindowFull
            | UdpTransportError::TooManyPeers
            | UdpTransportError::RuntimeUnavailable => Self::InvalidOperation,
        }
    }
}

/// C# `UDPPacketBuffer`, with one owned allocation per queued datagram.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct UDPPacketBuffer {
    pub data: Vec<u8>,
    pub data_length: i32,
    pub remote_end_point: SocketAddr,
}

impl UDPPacketBuffer {
    pub const DEFAULT_BUFFER_SIZE: i32 = 4096;

    pub fn new_with_constructor() -> Result<Self, Error> {
        Self::new_with_ip_end_point_int32(
            SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), 0),
            Self::DEFAULT_BUFFER_SIZE,
        )
    }

    pub fn new_with_bytes_int32_ip_end_point_int32(
        buffer: Vec<u8>,
        buffer_size: i32,
        destination: SocketAddr,
        _category: i32,
    ) -> Result<Self, Error> {
        let mut packet = Self::new_with_ip_end_point_int32(destination, buffer_size)?;
        packet.copy_from_slice_with_length(&buffer, buffer_size)?;
        packet.data_length = buffer_size;
        Ok(packet)
    }

    pub fn new_with_ip_end_point(end_point: SocketAddr) -> Result<Self, Error> {
        Self::new_with_ip_end_point_int32(end_point, Self::DEFAULT_BUFFER_SIZE)
    }

    pub fn new_with_ip_end_point_bytes(
        end_point: SocketAddr,
        data: Vec<u8>,
    ) -> Result<Self, Error> {
        Ok(Self {
            data,
            // The C# constructor adopts the supplied array but deliberately
            // leaves the public DataLength field at its zero default.
            data_length: 0,
            remote_end_point: end_point,
        })
    }

    pub fn new_with_ip_end_point_int32(
        end_point: SocketAddr,
        buffer_size: i32,
    ) -> Result<Self, Error> {
        let buffer_size = usize::try_from(buffer_size).map_err(|_| Error::Argument)?;
        let mut data = Vec::new();
        data.try_reserve_exact(buffer_size)
            .map_err(|_| Error::InvalidOperation)?;
        data.resize(buffer_size, 0);
        Ok(Self {
            data,
            data_length: 0,
            remote_end_point: end_point,
        })
    }

    /// Copies a mapped CLR array into this buffer.
    ///
    /// `Object::Bytes` represents a boxed `byte[]`; a flat array of integer
    /// objects is also accepted so the mapped `System.Array` remains useful.
    pub fn copy_from_with_array(&mut self, src: Array) -> Result<(), Error> {
        let bytes = mapped_array_bytes(src)?;
        let length = i32::try_from(bytes.len()).map_err(|_| Error::Argument)?;
        self.copy_from_slice_with_length(&bytes, length)
    }

    pub fn copy_from_with_array_int32(&mut self, src: Array, length: i32) -> Result<(), Error> {
        let bytes = mapped_array_bytes(src)?;
        self.copy_from_slice_with_length(&bytes, length)
    }

    pub fn copy_from_slice(&mut self, src: &[u8]) -> Result<(), Error> {
        let length = i32::try_from(src.len()).map_err(|_| Error::Argument)?;
        self.copy_from_slice_with_length(src, length)
    }

    pub fn copy_from_slice_with_length(&mut self, src: &[u8], length: i32) -> Result<(), Error> {
        let length = usize::try_from(length).map_err(|_| Error::Argument)?;
        if length > src.len() || length > self.data.len() {
            return Err(Error::IndexOutOfRange);
        }
        self.data[..length].copy_from_slice(&src[..length]);
        Ok(())
    }

    pub fn reset_endpoint(&mut self) -> Result<(), Error> {
        self.remote_end_point = match self.remote_end_point.ip() {
            IpAddr::V4(_) => SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), 0),
            IpAddr::V6(_) => SocketAddr::new(IpAddr::V6(Ipv6Addr::UNSPECIFIED), 0),
        };
        Ok(())
    }

    fn payload(&self) -> Result<&[u8], UdpTransportError> {
        let length =
            usize::try_from(self.data_length).map_err(|_| UdpTransportError::InvalidBuffer)?;
        self.data
            .get(..length)
            .ok_or(UdpTransportError::InvalidBuffer)
    }
}

fn mapped_array_bytes(src: Array) -> Result<Vec<u8>, Error> {
    if let [Object::Bytes(bytes)] = src.0.as_slice() {
        return Ok(bytes.clone());
    }
    let mut bytes = Vec::new();
    bytes
        .try_reserve_exact(src.0.len())
        .map_err(|_| Error::InvalidOperation)?;
    for value in src.0 {
        let byte = match value {
            Object::Integer(value) => u8::try_from(value).map_err(|_| Error::Argument)?,
            Object::UInteger(value) => u8::try_from(value).map_err(|_| Error::Argument)?,
            _ => return Err(Error::Argument),
        };
        bytes.push(byte);
    }
    Ok(bytes)
}

struct PacketArchiveState {
    items: Vec<u32>,
    members: HashSet<u32>,
    first: usize,
    next: usize,
}

/// Fixed-size duplicate archive matching `IncomingPacketIDCollection`.
pub struct IncomingPacketIDCollection {
    capacity: usize,
    state: Mutex<PacketArchiveState>,
}

impl IncomingPacketIDCollection {
    pub fn new(capacity: i32) -> Result<Self, Error> {
        let capacity = usize::try_from(capacity).map_err(|_| Error::Argument)?;
        if capacity == 0 {
            return Err(Error::Argument);
        }
        let mut items = Vec::new();
        items
            .try_reserve_exact(capacity)
            .map_err(|_| Error::InvalidOperation)?;
        items.resize(capacity, 0);
        Ok(Self {
            capacity,
            state: Mutex::new(PacketArchiveState {
                items,
                members: HashSet::with_capacity(capacity),
                first: 0,
                next: 0,
            }),
        })
    }

    pub fn try_enqueue(&self, ack: u32) -> bool {
        let mut state = self
            .state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        if !state.members.insert(ack) {
            return false;
        }
        let next = state.next;
        state.items[next] = ack;
        state.next = (next + 1) % self.capacity;
        if state.next == state.first {
            let first = state.first;
            let removed = state.items[first];
            state.members.remove(&removed);
            state.first = (first + 1) % self.capacity;
        }
        true
    }
}

impl fmt::Debug for IncomingPacketIDCollection {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        let state = self
            .state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        formatter
            .debug_struct("IncomingPacketIDCollection")
            .field("capacity", &self.capacity)
            .field("len", &state.members.len())
            .finish()
    }
}

/// Injection boundary used by `AgentThrottle::Set` and the later network
/// manager composition layer.
pub trait AgentThrottleSender: Send + Sync {
    fn send_throttle(
        &self,
        throttle_bytes: &[u8],
        simulator: Option<&Simulator>,
    ) -> Result<(), Error>;
}

/// Exact seven-stream C# throttle values and little-endian wire encoding.
#[derive(Clone)]
pub struct AgentThrottle {
    resend: f32,
    land: f32,
    wind: f32,
    cloud: f32,
    task: f32,
    texture: f32,
    asset: f32,
    sender: Option<Arc<dyn AgentThrottleSender>>,
}

impl AgentThrottle {
    pub fn new_with_grid_client(client: GridClient) -> Result<Self, Error> {
        Ok(Self {
            sender: client.agent_throttle_sender(),
            ..Self::default()
        })
    }

    pub fn new_with_bytes_int32(data: Vec<u8>, pos: i32) -> Result<Self, Error> {
        let pos = usize::try_from(pos).map_err(|_| Error::Argument)?;
        let end = pos.checked_add(28).ok_or(Error::Argument)?;
        let bytes = data.get(pos..end).ok_or(Error::IndexOutOfRange)?;
        let mut value = Self::default();
        value.set_resend(read_f32(bytes, 0)?);
        value.set_land(read_f32(bytes, 4)?);
        value.set_wind(read_f32(bytes, 8)?);
        value.set_cloud(read_f32(bytes, 12)?);
        value.set_task(read_f32(bytes, 16)?);
        value.set_texture(read_f32(bytes, 20)?);
        value.set_asset(read_f32(bytes, 24)?);
        Ok(value)
    }

    #[must_use]
    pub fn with_sender(mut self, sender: Arc<dyn AgentThrottleSender>) -> Self {
        self.sender = Some(sender);
        self
    }

    pub fn set_with_method(&self) -> Result<(), Error> {
        if let Some(sender) = &self.sender {
            sender.send_throttle(&self.to_bytes()?, None)?;
        }
        Ok(())
    }

    pub fn set_with_simulator(&self, simulator: Option<Simulator>) -> Result<(), Error> {
        if let (Some(sender), Some(simulator)) = (&self.sender, simulator.as_ref()) {
            sender.send_throttle(&self.to_bytes()?, Some(simulator))?;
        }
        Ok(())
    }

    pub fn to_bytes(&self) -> Result<Vec<u8>, Error> {
        let mut output = Vec::with_capacity(28);
        for value in [
            self.resend,
            self.land,
            self.wind,
            self.cloud,
            self.task,
            self.texture,
            self.asset,
        ] {
            output.extend_from_slice(&value.to_le_bytes());
        }
        Ok(output)
    }

    #[must_use]
    pub const fn asset(&self) -> f32 {
        self.asset
    }

    pub fn set_asset(&mut self, value: f32) {
        self.asset = value.clamp(10_000.0, 220_000.0);
    }

    #[must_use]
    pub const fn cloud(&self) -> f32 {
        self.cloud
    }

    pub fn set_cloud(&mut self, value: f32) {
        self.cloud = value.clamp(0.0, 34_000.0);
    }

    #[must_use]
    pub const fn land(&self) -> f32 {
        self.land
    }

    pub fn set_land(&mut self, value: f32) {
        self.land = value.clamp(0.0, 170_000.0);
    }

    #[must_use]
    pub const fn resend(&self) -> f32 {
        self.resend
    }

    pub fn set_resend(&mut self, value: f32) {
        self.resend = value.clamp(10_000.0, 150_000.0);
    }

    #[must_use]
    pub const fn task(&self) -> f32 {
        self.task
    }

    pub fn set_task(&mut self, value: f32) {
        self.task = value.clamp(4_000.0, 1_338_000.0);
    }

    #[must_use]
    pub const fn texture(&self) -> f32 {
        self.texture
    }

    pub fn set_texture(&mut self, value: f32) {
        self.texture = value.clamp(4_000.0, 446_000.0);
    }

    #[must_use]
    pub fn total(&self) -> f32 {
        self.resend + self.land + self.wind + self.cloud + self.task + self.texture + self.asset
    }

    pub fn set_total(&mut self, value: f32) {
        self.set_resend(value * 0.1);
        self.set_land(value * 0.52 / 3.0);
        self.set_wind(value * 0.05);
        self.set_cloud(value * 0.05);
        self.set_task(value * 0.704 / 3.0);
        self.set_texture(value * 0.704 / 3.0);
        self.set_asset(value * 0.484 / 3.0);
    }

    #[must_use]
    pub const fn wind(&self) -> f32 {
        self.wind
    }

    pub fn set_wind(&mut self, value: f32) {
        self.wind = value.clamp(0.0, 34_000.0);
    }
}

impl Default for AgentThrottle {
    fn default() -> Self {
        let mut value = Self {
            resend: 0.0,
            land: 0.0,
            wind: 0.0,
            cloud: 0.0,
            task: 0.0,
            texture: 0.0,
            asset: 0.0,
            sender: None,
        };
        value.set_total(1_536_000.0);
        value
    }
}

impl fmt::Debug for AgentThrottle {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("AgentThrottle")
            .field("resend", &self.resend)
            .field("land", &self.land)
            .field("wind", &self.wind)
            .field("cloud", &self.cloud)
            .field("task", &self.task)
            .field("texture", &self.texture)
            .field("asset", &self.asset)
            .finish_non_exhaustive()
    }
}

fn read_f32(bytes: &[u8], offset: usize) -> Result<f32, Error> {
    Ok(f32::from_le_bytes(
        bytes
            .get(offset..offset + 4)
            .ok_or(Error::IndexOutOfRange)?
            .try_into()
            .map_err(|_| Error::IndexOutOfRange)?,
    ))
}

/// Outgoing categories used by the native token buckets.
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
pub enum UdpThrottleCategory {
    Unthrottled,
    Task,
    Texture,
    Asset,
}

impl UdpThrottleCategory {
    #[must_use]
    pub const fn classify(packet_type: PacketType) -> Self {
        match packet_type {
            PacketType::UseCircuitCode
            | PacketType::CompleteAgentMovement
            | PacketType::AgentThrottle
            | PacketType::LogoutRequest
            | PacketType::PacketAck
            | PacketType::StartPingCheck
            | PacketType::CompletePingCheck
            | PacketType::CloseCircuit => Self::Unthrottled,
            PacketType::RequestImage => Self::Texture,
            PacketType::TransferRequest | PacketType::AbortXfer => Self::Asset,
            _ => Self::Task,
        }
    }
}

/// Bounded transport policy. Defaults match the C# settings used by a client
/// simulator, with explicit limits for queues and per-peer reliable state.
#[derive(Clone, Debug)]
pub struct UdpTransportConfig {
    pub receive_queue_capacity: usize,
    pub command_queue_capacity: usize,
    pub write_queue_capacity: usize,
    pub packet_archive_size: usize,
    pub pending_ack_capacity: usize,
    pub max_pending_acks: usize,
    pub reliable_window_capacity: usize,
    pub max_resend_count: u32,
    pub resend_timeout: Duration,
    pub network_tick_interval: Duration,
    pub max_datagram_size: usize,
    pub max_decoded_packet_size: usize,
    pub protocol_mtu: usize,
    pub max_peers: usize,
    pub throttle: AgentThrottle,
}

impl Default for UdpTransportConfig {
    fn default() -> Self {
        Self {
            receive_queue_capacity: 512,
            command_queue_capacity: 512,
            write_queue_capacity: 512,
            packet_archive_size: 1000,
            pending_ack_capacity: 255,
            max_pending_acks: 10,
            reliable_window_capacity: 1024,
            max_resend_count: 3,
            resend_timeout: Duration::from_secs(4),
            network_tick_interval: Duration::from_millis(500),
            max_datagram_size: usize::try_from(UDPPacketBuffer::DEFAULT_BUFFER_SIZE)
                .unwrap_or(4096),
            max_decoded_packet_size: DEFAULT_DECODE_BUFFER_SIZE,
            protocol_mtu: usize::try_from(Packet::MTU).unwrap_or(1200),
            max_peers: 32,
            throttle: AgentThrottle::default(),
        }
    }
}

impl UdpTransportConfig {
    fn validate(&self) -> Result<(), UdpTransportError> {
        for (name, value) in [
            ("receive_queue_capacity", self.receive_queue_capacity),
            ("command_queue_capacity", self.command_queue_capacity),
            ("write_queue_capacity", self.write_queue_capacity),
            ("packet_archive_size", self.packet_archive_size),
            ("pending_ack_capacity", self.pending_ack_capacity),
            ("max_pending_acks", self.max_pending_acks),
            ("reliable_window_capacity", self.reliable_window_capacity),
            ("max_datagram_size", self.max_datagram_size),
            ("max_decoded_packet_size", self.max_decoded_packet_size),
            ("protocol_mtu", self.protocol_mtu),
            ("max_peers", self.max_peers),
        ] {
            if value == 0 {
                return Err(UdpTransportError::InvalidConfiguration(name));
            }
        }
        if self.pending_ack_capacity > usize::from(u8::MAX) {
            return Err(UdpTransportError::InvalidConfiguration(
                "pending_ack_capacity",
            ));
        }
        if self.max_pending_acks > self.pending_ack_capacity {
            return Err(UdpTransportError::InvalidConfiguration("max_pending_acks"));
        }
        if self.protocol_mtu < 10 || self.protocol_mtu > self.max_datagram_size {
            return Err(UdpTransportError::InvalidConfiguration("protocol_mtu"));
        }
        if self.max_decoded_packet_size < self.max_datagram_size {
            return Err(UdpTransportError::InvalidConfiguration(
                "max_decoded_packet_size",
            ));
        }
        if self.resend_timeout.is_zero() {
            return Err(UdpTransportError::InvalidConfiguration("resend_timeout"));
        }
        if self.network_tick_interval.is_zero() {
            return Err(UdpTransportError::InvalidConfiguration(
                "network_tick_interval",
            ));
        }
        Ok(())
    }
}

/// Immutable, redacted snapshot of transport counters.
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub struct UdpTransportStats {
    pub received_datagrams: u64,
    pub received_bytes: u64,
    pub sent_datagrams: u64,
    pub sent_bytes: u64,
    pub dropped_receive_queue: u64,
    pub dropped_send_queue: u64,
    pub malformed_datagrams: u64,
    pub rejected_sources: u64,
    pub duplicate_datagrams: u64,
    pub out_of_order_datagrams: u64,
    pub sequence_gaps: u64,
    pub acknowledgements_received: u64,
    pub acknowledgements_sent: u64,
    pub resent_datagrams: u64,
    pub failed_resends: u64,
    pub socket_errors: u64,
    pub handler_panics: u64,
}

#[derive(Default)]
struct StatsCounters {
    received_datagrams: AtomicU64,
    received_bytes: AtomicU64,
    sent_datagrams: AtomicU64,
    sent_bytes: AtomicU64,
    dropped_receive_queue: AtomicU64,
    dropped_send_queue: AtomicU64,
    malformed_datagrams: AtomicU64,
    rejected_sources: AtomicU64,
    duplicate_datagrams: AtomicU64,
    out_of_order_datagrams: AtomicU64,
    sequence_gaps: AtomicU64,
    acknowledgements_received: AtomicU64,
    acknowledgements_sent: AtomicU64,
    resent_datagrams: AtomicU64,
    failed_resends: AtomicU64,
    socket_errors: AtomicU64,
    handler_panics: AtomicU64,
}

impl StatsCounters {
    fn snapshot(&self) -> UdpTransportStats {
        let load = |counter: &AtomicU64| counter.load(Ordering::Relaxed);
        UdpTransportStats {
            received_datagrams: load(&self.received_datagrams),
            received_bytes: load(&self.received_bytes),
            sent_datagrams: load(&self.sent_datagrams),
            sent_bytes: load(&self.sent_bytes),
            dropped_receive_queue: load(&self.dropped_receive_queue),
            dropped_send_queue: load(&self.dropped_send_queue),
            malformed_datagrams: load(&self.malformed_datagrams),
            rejected_sources: load(&self.rejected_sources),
            duplicate_datagrams: load(&self.duplicate_datagrams),
            out_of_order_datagrams: load(&self.out_of_order_datagrams),
            sequence_gaps: load(&self.sequence_gaps),
            acknowledgements_received: load(&self.acknowledgements_received),
            acknowledgements_sent: load(&self.acknowledgements_sent),
            resent_datagrams: load(&self.resent_datagrams),
            failed_resends: load(&self.failed_resends),
            socket_errors: load(&self.socket_errors),
            handler_panics: load(&self.handler_panics),
        }
    }
}

/// Callback boundary corresponding to the protected methods on C# `UDPBase`.
pub trait UdpPacketHandler: Send + Sync + 'static {
    fn packet_received(&self, _buffer: UDPPacketBuffer) {}
    fn packet_sent(&self, _buffer: UDPPacketBuffer, _bytes_sent: usize) {}
    fn packet_dropped(&self) {}
}

struct NoopPacketHandler;
impl UdpPacketHandler for NoopPacketHandler {}

struct RuntimeTasks {
    cancellation: CancellationTokenSource,
    commands: mpsc::Sender<CoordinatorCommand>,
    handles: Vec<JoinHandle<()>>,
    local_address: SocketAddr,
}

struct UdpInner {
    bind_address: SocketAddr,
    remote_end_point: Option<SocketAddr>,
    config: UdpTransportConfig,
    handler: Arc<dyn UdpPacketHandler>,
    parent_cancellation: CancellationToken,
    running: AtomicBool,
    runtime: Mutex<Option<RuntimeTasks>>,
    stats: Arc<StatsCounters>,
}

impl Drop for UdpInner {
    fn drop(&mut self) {
        if let Some(runtime) = self
            .runtime
            .get_mut()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .take()
        {
            runtime.cancellation.cancel();
            for handle in runtime.handles {
                handle.abort();
            }
        }
    }
}

/// Runtime-owning native implementation of the C# UDP transport slice.
///
/// Construction creates no socket and no runtime. `start` binds a nonblocking
/// socket and spawns tasks on the caller's current Tokio runtime.
#[derive(Clone)]
pub struct UDPBase {
    inner: Arc<UdpInner>,
}

impl UDPBase {
    pub fn client(
        remote_end_point: SocketAddr,
        config: UdpTransportConfig,
        handler: Arc<dyn UdpPacketHandler>,
        parent_cancellation: CancellationToken,
    ) -> Result<Self, UdpTransportError> {
        let bind_address = match remote_end_point.ip() {
            IpAddr::V4(_) => SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), 0),
            IpAddr::V6(_) => SocketAddr::new(IpAddr::V6(Ipv6Addr::UNSPECIFIED), 0),
        };
        Self::new(
            bind_address,
            Some(remote_end_point),
            config,
            handler,
            parent_cancellation,
        )
    }

    pub fn server(
        bind_address: SocketAddr,
        config: UdpTransportConfig,
        handler: Arc<dyn UdpPacketHandler>,
        parent_cancellation: CancellationToken,
    ) -> Result<Self, UdpTransportError> {
        Self::new(bind_address, None, config, handler, parent_cancellation)
    }

    pub fn client_with_defaults(remote_end_point: SocketAddr) -> Result<Self, UdpTransportError> {
        Self::client(
            remote_end_point,
            UdpTransportConfig::default(),
            Arc::new(NoopPacketHandler),
            CancellationToken::default(),
        )
    }

    fn new(
        bind_address: SocketAddr,
        remote_end_point: Option<SocketAddr>,
        config: UdpTransportConfig,
        handler: Arc<dyn UdpPacketHandler>,
        parent_cancellation: CancellationToken,
    ) -> Result<Self, UdpTransportError> {
        config.validate()?;
        Ok(Self {
            inner: Arc::new(UdpInner {
                bind_address,
                remote_end_point,
                config,
                handler,
                parent_cancellation,
                running: AtomicBool::new(false),
                runtime: Mutex::new(None),
                stats: Arc::new(StatsCounters::default()),
            }),
        })
    }

    pub fn start(&self) -> Result<(), Error> {
        self.start_transport().map_err(Into::into)
    }

    pub fn start_transport(&self) -> Result<(), UdpTransportError> {
        let mut runtime = self
            .inner
            .runtime
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        if runtime.is_some() {
            return Ok(());
        }
        if self.inner.parent_cancellation.is_cancellation_requested() {
            return Err(UdpTransportError::Cancelled);
        }
        let handle = tokio::runtime::Handle::try_current()
            .map_err(|_| UdpTransportError::RuntimeUnavailable)?;
        let std_socket = std::net::UdpSocket::bind(self.inner.bind_address)
            .map_err(|_| UdpTransportError::Socket)?;
        std_socket
            .set_nonblocking(true)
            .map_err(|_| UdpTransportError::Socket)?;
        let local_address = std_socket
            .local_addr()
            .map_err(|_| UdpTransportError::Socket)?;
        let socket =
            Arc::new(UdpSocket::from_std(std_socket).map_err(|_| UdpTransportError::Socket)?);
        let cancellation = CancellationTokenSource::new_linked(std::slice::from_ref(
            &self.inner.parent_cancellation,
        ));
        let token = cancellation.token();
        let (raw_sender, raw_receiver) = mpsc::channel(self.inner.config.receive_queue_capacity);
        let (command_sender, command_receiver) =
            mpsc::channel(self.inner.config.command_queue_capacity);
        let (write_sender, write_receiver) = mpsc::channel(self.inner.config.write_queue_capacity);

        let receiver_handle = handle.spawn(receive_loop(
            Arc::clone(&socket),
            raw_sender,
            token.clone(),
            self.inner.config.max_datagram_size,
            Arc::clone(&self.inner.stats),
            Arc::clone(&self.inner.handler),
        ));
        let coordinator_handle = handle.spawn(coordinator_loop(
            raw_receiver,
            command_receiver,
            write_sender,
            token.clone(),
            self.inner.remote_end_point,
            self.inner.config.clone(),
            Arc::clone(&self.inner.stats),
            Arc::clone(&self.inner.handler),
        ));
        let writer_handle = handle.spawn(writer_loop(
            socket,
            write_receiver,
            token,
            self.inner.config.throttle.clone(),
            Arc::clone(&self.inner.stats),
            Arc::clone(&self.inner.handler),
        ));
        *runtime = Some(RuntimeTasks {
            cancellation,
            commands: command_sender,
            handles: vec![receiver_handle, coordinator_handle, writer_handle],
            local_address,
        });
        self.inner.running.store(true, Ordering::Release);
        Ok(())
    }

    pub fn stop(&self) -> Result<(), Error> {
        self.stop_transport();
        Ok(())
    }

    pub fn stop_transport(&self) {
        let runtime = self
            .inner
            .runtime
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .take();
        self.inner.running.store(false, Ordering::Release);
        if let Some(runtime) = runtime {
            runtime.cancellation.cancel();
            for handle in runtime.handles {
                handle.abort();
            }
        }
    }

    pub async fn stop_async(&self) {
        let runtime = self
            .inner
            .runtime
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .take();
        self.inner.running.store(false, Ordering::Release);
        if let Some(runtime) = runtime {
            runtime.cancellation.cancel();
            drop(runtime.commands);
            for handle in runtime.handles {
                let _ = handle.await;
            }
        }
    }

    #[must_use]
    pub fn is_running(&self) -> bool {
        self.inner.running.load(Ordering::Acquire)
    }

    #[must_use]
    pub fn local_address(&self) -> Option<SocketAddr> {
        self.inner
            .runtime
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .as_ref()
            .map(|runtime| runtime.local_address)
    }

    #[must_use]
    pub fn stats(&self) -> UdpTransportStats {
        self.inner.stats.snapshot()
    }

    pub fn async_begin_send(&self, buf: UDPPacketBuffer) -> Result<(), Error> {
        let Some(sender) = self.command_sender() else {
            return Ok(());
        };
        sender
            .try_send(CoordinatorCommand::Raw(buf))
            .map_err(|error| {
                if matches!(error, mpsc::error::TrySendError::Full(_)) {
                    self.inner
                        .stats
                        .dropped_send_queue
                        .fetch_add(1, Ordering::Relaxed);
                    Error::InvalidOperation
                } else {
                    Error::Cancelled
                }
            })
    }

    pub async fn send_packet(
        &self,
        data: Vec<u8>,
        packet_type: PacketType,
        do_zerocode: bool,
        cancellation: CancellationToken,
    ) -> Result<u32, UdpTransportError> {
        let destination = self
            .inner
            .remote_end_point
            .ok_or(UdpTransportError::InvalidBuffer)?;
        self.send_packet_to(data, destination, packet_type, do_zerocode, cancellation)
            .await
    }

    pub async fn send_packet_to(
        &self,
        data: Vec<u8>,
        destination: SocketAddr,
        packet_type: PacketType,
        do_zerocode: bool,
        cancellation: CancellationToken,
    ) -> Result<u32, UdpTransportError> {
        let sender = self.command_sender().ok_or(UdpTransportError::NotRunning)?;
        let (response_sender, response_receiver) = oneshot::channel();
        let command = CoordinatorCommand::Packet {
            data,
            destination,
            packet_type,
            do_zerocode,
            response: response_sender,
            acknowledgement: None,
        };
        tokio::select! {
            result = sender.send(command) => {
                result.map_err(|_| UdpTransportError::Cancelled)?;
            }
            () = cancellation.cancelled() => return Err(UdpTransportError::Cancelled),
            () = self.inner.parent_cancellation.cancelled() => {
                return Err(UdpTransportError::Cancelled);
            }
        }
        tokio::select! {
            result = response_receiver => result.map_err(|_| UdpTransportError::Cancelled)?,
            () = cancellation.cancelled() => Err(UdpTransportError::Cancelled),
            () = self.inner.parent_cancellation.cancelled() => Err(UdpTransportError::Cancelled),
        }
    }

    pub fn try_send_packet(
        &self,
        data: Vec<u8>,
        destination: SocketAddr,
        packet_type: PacketType,
        do_zerocode: bool,
    ) -> Result<oneshot::Receiver<Result<u32, UdpTransportError>>, UdpTransportError> {
        let sender = self.command_sender().ok_or(UdpTransportError::NotRunning)?;
        let (response, receiver) = oneshot::channel();
        sender
            .try_send(CoordinatorCommand::Packet {
                data,
                destination,
                packet_type,
                do_zerocode,
                response,
                acknowledgement: None,
            })
            .map_err(|error| {
                if matches!(error, mpsc::error::TrySendError::Full(_)) {
                    self.inner
                        .stats
                        .dropped_send_queue
                        .fetch_add(1, Ordering::Relaxed);
                    UdpTransportError::Backpressure
                } else {
                    UdpTransportError::Cancelled
                }
            })?;
        Ok(receiver)
    }

    /// Queues a packet and resolves only after its reliable sequence is
    /// acknowledged by the peer. The receiver disconnects if the reliable
    /// packet is dropped or the transport shuts down.
    pub(crate) fn try_send_packet_wait_ack(
        &self,
        data: Vec<u8>,
        destination: SocketAddr,
        packet_type: PacketType,
        do_zerocode: bool,
    ) -> Result<AckReceiver<()>, UdpTransportError> {
        let sender = self.command_sender().ok_or(UdpTransportError::NotRunning)?;
        let (response, _queued) = oneshot::channel();
        let (acknowledgement, receiver) = sync_channel(1);
        sender
            .try_send(CoordinatorCommand::Packet {
                data,
                destination,
                packet_type,
                do_zerocode,
                response,
                acknowledgement: Some(acknowledgement),
            })
            .map_err(|error| {
                if matches!(error, mpsc::error::TrySendError::Full(_)) {
                    self.inner
                        .stats
                        .dropped_send_queue
                        .fetch_add(1, Ordering::Relaxed);
                    UdpTransportError::Backpressure
                } else {
                    UdpTransportError::Cancelled
                }
            })?;
        Ok(receiver)
    }

    /// Queues an immediate flush of acknowledgements pending for `destination`.
    ///
    /// The network manager uses this for the protocol's `OldestUnacked`
    /// request in `StartPingCheck`, matching `Simulator.SendAcks()` without
    /// exposing coordinator state or blocking the packet callback thread.
    pub(crate) fn try_flush_acks(&self, destination: SocketAddr) -> Result<(), UdpTransportError> {
        let sender = self.command_sender().ok_or(UdpTransportError::NotRunning)?;
        sender
            .try_send(CoordinatorCommand::FlushAcks { destination })
            .map_err(|error| {
                if matches!(error, mpsc::error::TrySendError::Full(_)) {
                    self.inner
                        .stats
                        .dropped_send_queue
                        .fetch_add(1, Ordering::Relaxed);
                    UdpTransportError::Backpressure
                } else {
                    UdpTransportError::Cancelled
                }
            })
    }

    pub fn update_throttle(&self, throttle: AgentThrottle) -> Result<(), UdpTransportError> {
        let sender = self.command_sender().ok_or(UdpTransportError::NotRunning)?;
        sender
            .try_send(CoordinatorCommand::UpdateThrottle(throttle))
            .map_err(|error| {
                if matches!(error, mpsc::error::TrySendError::Full(_)) {
                    UdpTransportError::Backpressure
                } else {
                    UdpTransportError::Cancelled
                }
            })
    }

    fn command_sender(&self) -> Option<mpsc::Sender<CoordinatorCommand>> {
        self.inner
            .runtime
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .as_ref()
            .map(|runtime| runtime.commands.clone())
    }
}

impl fmt::Debug for UDPBase {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("UDPBase")
            .field(
                "mode",
                &self.inner.remote_end_point.map_or("server", |_| "client"),
            )
            .field("is_running", &self.is_running())
            .field("stats", &self.stats())
            .finish_non_exhaustive()
    }
}

enum CoordinatorCommand {
    Raw(UDPPacketBuffer),
    Packet {
        data: Vec<u8>,
        destination: SocketAddr,
        packet_type: PacketType,
        do_zerocode: bool,
        response: oneshot::Sender<Result<u32, UdpTransportError>>,
        acknowledgement: Option<AckSender<()>>,
    },
    FlushAcks {
        destination: SocketAddr,
    },
    UpdateThrottle(AgentThrottle),
}

enum WriteCommand {
    Datagram {
        buffer: UDPPacketBuffer,
        category: UdpThrottleCategory,
    },
    UpdateThrottle(AgentThrottle),
}

struct ReliablePacket {
    buffer: UDPPacketBuffer,
    category: UdpThrottleCategory,
    last_sent: Instant,
    resend_count: u32,
    acknowledgement: Option<AckSender<()>>,
}

struct PeerState {
    sequence: u32,
    archive: IncomingPacketIDCollection,
    pending_acks: VecDeque<u32>,
    need_ack: BTreeMap<u32, ReliablePacket>,
    latest_received: Option<u32>,
}

impl PeerState {
    fn new(config: &UdpTransportConfig) -> Result<Self, UdpTransportError> {
        Ok(Self {
            sequence: 0,
            archive: IncomingPacketIDCollection::new(
                i32::try_from(config.packet_archive_size)
                    .map_err(|_| UdpTransportError::InvalidConfiguration("packet_archive_size"))?,
            )
            .map_err(|_| UdpTransportError::InvalidConfiguration("packet_archive_size"))?,
            pending_acks: VecDeque::with_capacity(config.pending_ack_capacity),
            need_ack: BTreeMap::new(),
            latest_received: None,
        })
    }

    fn next_sequence(&mut self) -> u32 {
        // C# increments an Int32 and casts it to UInt32 for the four-byte
        // header. The equivalent wire sequence spans all u32 values and wraps
        // through zero.
        self.sequence = self.sequence.wrapping_add(1);
        self.sequence
    }

    fn observe_sequence(&mut self, sequence: u32, stats: &StatsCounters) {
        let Some(latest) = self.latest_received else {
            self.latest_received = Some(sequence);
            return;
        };
        let expected = latest.wrapping_add(1);
        if sequence == expected {
            self.latest_received = Some(sequence);
            return;
        }
        let forward = sequence.wrapping_sub(latest);
        if forward != 0 && forward <= (u32::MAX / 2) + 1 {
            stats.sequence_gaps.fetch_add(1, Ordering::Relaxed);
            self.latest_received = Some(sequence);
        } else {
            stats.out_of_order_datagrams.fetch_add(1, Ordering::Relaxed);
        }
    }
}

struct CoordinatorState {
    peers: HashMap<SocketAddr, PeerState>,
    remote_end_point: Option<SocketAddr>,
    config: UdpTransportConfig,
    stats: Arc<StatsCounters>,
}

impl CoordinatorState {
    fn canonical_peer(&self, source: SocketAddr) -> Result<SocketAddr, UdpTransportError> {
        if let Some(remote) = self.remote_end_point {
            if remote.ip() != source.ip() {
                return Err(UdpTransportError::InvalidBuffer);
            }
            Ok(remote)
        } else {
            Ok(source)
        }
    }

    fn peer_mut(&mut self, endpoint: SocketAddr) -> Result<&mut PeerState, UdpTransportError> {
        if !self.peers.contains_key(&endpoint) {
            if self.peers.len() >= self.config.max_peers {
                return Err(UdpTransportError::TooManyPeers);
            }
            let peer = PeerState::new(&self.config)?;
            self.peers.insert(endpoint, peer);
        }
        self.peers
            .get_mut(&endpoint)
            .ok_or(UdpTransportError::TooManyPeers)
    }

    fn prepare_packet(
        &mut self,
        data: Vec<u8>,
        destination: SocketAddr,
        packet_type: PacketType,
        do_zerocode: bool,
        acknowledgement: Option<AckSender<()>>,
    ) -> Result<(WriteCommand, u32), UdpTransportError> {
        let mut data = encode_for_transport(data, do_zerocode, self.config.protocol_mtu)?;
        if data.len() < 6 {
            return Err(UdpTransportError::InvalidBuffer);
        }
        let reliable = data[0] & Helpers::MSG_RELIABLE != 0;
        let category = UdpThrottleCategory::classify(packet_type);
        let reliable_capacity = self.config.reliable_window_capacity;
        let protocol_mtu = self.config.protocol_mtu;
        let peer = self.peer_mut(destination)?;
        if reliable && peer.need_ack.len() >= reliable_capacity {
            return Err(UdpTransportError::ReliableWindowFull);
        }
        append_pending_acks(&mut data, &mut peer.pending_acks, protocol_mtu)?;
        let sequence = peer.next_sequence();
        data[1..5].copy_from_slice(&sequence.to_be_bytes());
        let data_length =
            i32::try_from(data.len()).map_err(|_| UdpTransportError::InvalidBuffer)?;
        let buffer = UDPPacketBuffer {
            data,
            data_length,
            remote_end_point: destination,
        };
        if reliable {
            peer.need_ack.insert(
                sequence,
                ReliablePacket {
                    buffer: buffer.clone(),
                    category,
                    last_sent: Instant::now(),
                    resend_count: 0,
                    acknowledgement,
                },
            );
        } else if let Some(acknowledgement) = acknowledgement {
            let _ = acknowledgement.try_send(());
        }
        Ok((WriteCommand::Datagram { buffer, category }, sequence))
    }

    fn take_resends(&mut self) -> Vec<WriteCommand> {
        let now = Instant::now();
        let mut writes = Vec::new();
        for peer in self.peers.values_mut() {
            let mut failed = Vec::new();
            for (&sequence, packet) in &mut peer.need_ack {
                if now.duration_since(packet.last_sent) <= self.config.resend_timeout {
                    continue;
                }
                if packet.resend_count < self.config.max_resend_count {
                    packet.resend_count += 1;
                    packet.last_sent = now;
                    if let Some(flags) = packet.buffer.data.first_mut() {
                        *flags |= Helpers::MSG_RESENT;
                    }
                    self.stats.resent_datagrams.fetch_add(1, Ordering::Relaxed);
                    writes.push(WriteCommand::Datagram {
                        buffer: packet.buffer.clone(),
                        category: packet.category,
                    });
                } else {
                    failed.push(sequence);
                }
            }
            for sequence in failed {
                peer.need_ack.remove(&sequence);
                self.stats.failed_resends.fetch_add(1, Ordering::Relaxed);
            }
        }
        writes
    }
}

fn encode_for_transport(
    mut data: Vec<u8>,
    do_zerocode: bool,
    mtu: usize,
) -> Result<Vec<u8>, UdpTransportError> {
    if data.len() < 6 {
        return Err(UdpTransportError::InvalidBuffer);
    }
    if data.len() > mtu {
        return Err(UdpTransportError::MtuExceeded);
    }
    if !do_zerocode {
        return Ok(data);
    }
    data[0] |= Helpers::MSG_ZEROCODED;
    let mut encoded = vec![0_u8; mtu];
    let source_length = i32::try_from(data.len()).map_err(|_| UdpTransportError::InvalidBuffer)?;
    match crate::packet_wire::zero_encode(Some(&data), source_length, Some(&mut encoded)) {
        Ok(length) => {
            let length = usize::try_from(length).map_err(|_| UdpTransportError::InvalidBuffer)?;
            encoded.truncate(length);
            Ok(encoded)
        }
        Err(Error::IndexOutOfRange) => {
            data[0] &= !Helpers::MSG_ZEROCODED;
            Ok(data)
        }
        Err(_) => Err(UdpTransportError::InvalidBuffer),
    }
}

fn append_pending_acks(
    data: &mut Vec<u8>,
    pending: &mut VecDeque<u32>,
    mtu: usize,
) -> Result<(), UdpTransportError> {
    if data
        .first()
        .is_some_and(|flags| flags & Helpers::MSG_APPENDED_ACKS != 0)
    {
        return Err(UdpTransportError::InvalidBuffer);
    }
    let mut count = 0_u8;
    while data.len().checked_add(5).is_some_and(|length| length < mtu) {
        let Some(ack) = pending.pop_front() else {
            break;
        };
        data.extend_from_slice(&ack.to_be_bytes());
        count = count.saturating_add(1);
        if count == u8::MAX {
            break;
        }
    }
    if count != 0 {
        data.push(count);
        data[0] |= Helpers::MSG_APPENDED_ACKS;
    }
    Ok(())
}

async fn receive_loop(
    socket: Arc<UdpSocket>,
    sender: mpsc::Sender<UDPPacketBuffer>,
    cancellation: CancellationToken,
    max_datagram_size: usize,
    stats: Arc<StatsCounters>,
    handler: Arc<dyn UdpPacketHandler>,
) {
    let mut storage = vec![0_u8; max_datagram_size.saturating_add(1)];
    loop {
        let received = tokio::select! {
            () = cancellation.cancelled() => break,
            result = socket.recv_from(&mut storage) => result,
        };
        let Ok((length, source)) = received else {
            if !cancellation.is_cancellation_requested() {
                stats.socket_errors.fetch_add(1, Ordering::Relaxed);
            }
            break;
        };
        stats.received_datagrams.fetch_add(1, Ordering::Relaxed);
        stats
            .received_bytes
            .fetch_add(u64::try_from(length).unwrap_or(u64::MAX), Ordering::Relaxed);
        if length > max_datagram_size {
            stats.malformed_datagrams.fetch_add(1, Ordering::Relaxed);
            continue;
        }
        let packet = UDPPacketBuffer {
            data: storage[..length].to_vec(),
            data_length: i32::try_from(length).unwrap_or(i32::MAX),
            remote_end_point: source,
        };
        if sender.try_send(packet).is_err() {
            stats.dropped_receive_queue.fetch_add(1, Ordering::Relaxed);
            invoke_handler(&stats, || handler.packet_dropped());
        }
    }
}

#[allow(clippy::too_many_arguments)] // Each argument is one explicit task ownership boundary.
async fn coordinator_loop(
    mut incoming: mpsc::Receiver<UDPPacketBuffer>,
    mut commands: mpsc::Receiver<CoordinatorCommand>,
    writer: mpsc::Sender<WriteCommand>,
    cancellation: CancellationToken,
    remote_end_point: Option<SocketAddr>,
    config: UdpTransportConfig,
    stats: Arc<StatsCounters>,
    handler: Arc<dyn UdpPacketHandler>,
) {
    let mut coordinator = CoordinatorState {
        peers: HashMap::new(),
        remote_end_point,
        config: config.clone(),
        stats: Arc::clone(&stats),
    };
    let mut tick = tokio::time::interval(config.network_tick_interval);
    tick.set_missed_tick_behavior(MissedTickBehavior::Skip);
    tick.tick().await;
    loop {
        tokio::select! {
            biased;
            () = cancellation.cancelled() => break,
            Some(buffer) = incoming.recv() => {
                process_incoming(&mut coordinator, buffer, &writer, &handler).await;
            }
            Some(command) = commands.recv() => {
                process_command(&mut coordinator, command, &writer).await;
            }
            _ = tick.tick() => {
                flush_pending_acks(&mut coordinator, &writer).await;
                for resend in coordinator.take_resends() {
                    if writer.send(resend).await.is_err() {
                        return;
                    }
                }
            }
            else => break,
        }
    }
}

async fn process_command(
    state: &mut CoordinatorState,
    command: CoordinatorCommand,
    writer: &mpsc::Sender<WriteCommand>,
) {
    match command {
        CoordinatorCommand::Raw(buffer) => {
            let valid = buffer.payload().is_ok()
                && usize::try_from(buffer.data_length)
                    .is_ok_and(|length| length <= state.config.max_datagram_size);
            if !valid {
                state
                    .stats
                    .dropped_send_queue
                    .fetch_add(1, Ordering::Relaxed);
                return;
            }
            let _ = writer
                .send(WriteCommand::Datagram {
                    buffer,
                    category: UdpThrottleCategory::Unthrottled,
                })
                .await;
        }
        CoordinatorCommand::Packet {
            data,
            destination,
            packet_type,
            do_zerocode,
            response,
            acknowledgement,
        } => {
            let result =
                state.prepare_packet(data, destination, packet_type, do_zerocode, acknowledgement);
            let result = match result {
                Ok((write, sequence)) => writer
                    .send(write)
                    .await
                    .map(|()| sequence)
                    .map_err(|_| UdpTransportError::Cancelled),
                Err(error) => Err(error),
            };
            let _ = response.send(result);
        }
        CoordinatorCommand::FlushAcks { destination } => {
            send_peer_acks(state, destination, writer).await;
        }
        CoordinatorCommand::UpdateThrottle(throttle) => {
            let _ = writer.send(WriteCommand::UpdateThrottle(throttle)).await;
        }
    }
}

async fn process_incoming(
    state: &mut CoordinatorState,
    buffer: UDPPacketBuffer,
    writer: &mpsc::Sender<WriteCommand>,
    handler: &Arc<dyn UdpPacketHandler>,
) {
    let Ok(endpoint) = state.canonical_peer(buffer.remote_end_point) else {
        state.stats.rejected_sources.fetch_add(1, Ordering::Relaxed);
        return;
    };
    let Ok(payload) = buffer.payload() else {
        state
            .stats
            .malformed_datagrams
            .fetch_add(1, Ordering::Relaxed);
        return;
    };
    let Ok(payload_length) = i32::try_from(payload.len()) else {
        state
            .stats
            .malformed_datagrams
            .fetch_add(1, Ordering::Relaxed);
        return;
    };
    let mut packet_end = payload_length - 1;
    let mut zero_buffer = vec![0_u8; state.config.max_decoded_packet_size];
    let Ok(packet) =
        crate::packet_wire::build_packet_from_bytes(payload, &mut packet_end, &mut zero_buffer)
    else {
        state
            .stats
            .malformed_datagrams
            .fetch_add(1, Ordering::Relaxed);
        return;
    };

    let appended_acks = packet.header.ack_list.clone().unwrap_or_default();
    let mut standalone_acks = Vec::new();
    if packet.type_ == PacketType::PacketAck {
        let mut position = 0;
        if let Ok(ack_packet) =
            PacketAckPacket::new_with_bytes_int32(payload.to_vec(), &mut position)
        {
            standalone_acks.extend(ack_packet.packets.into_iter().map(|block| block.id));
        } else {
            state
                .stats
                .malformed_datagrams
                .fetch_add(1, Ordering::Relaxed);
            return;
        }
    }

    let pending_threshold = state.config.max_pending_acks;
    let pending_capacity = state.config.pending_ack_capacity;
    let counters = Arc::clone(&state.stats);
    let Ok(peer) = state.peer_mut(endpoint) else {
        counters.rejected_sources.fetch_add(1, Ordering::Relaxed);
        return;
    };
    for ack in appended_acks.into_iter().chain(standalone_acks) {
        if let Some(mut reliable) = peer.need_ack.remove(&ack)
            && let Some(acknowledgement) = reliable.acknowledgement.take()
        {
            let _ = acknowledgement.try_send(());
        }
        counters
            .acknowledgements_received
            .fetch_add(1, Ordering::Relaxed);
    }

    if packet.header.reliable {
        if peer.pending_acks.len() < pending_capacity {
            peer.pending_acks.push_back(packet.header.sequence);
        }
        if !peer.archive.try_enqueue(packet.header.sequence) {
            counters.duplicate_datagrams.fetch_add(1, Ordering::Relaxed);
            if peer.pending_acks.len() >= pending_threshold {
                send_peer_acks(state, endpoint, writer).await;
            }
            return;
        }
        peer.observe_sequence(packet.header.sequence, &counters);
    }

    if state
        .peers
        .get(&endpoint)
        .is_some_and(|peer| peer.pending_acks.len() >= pending_threshold)
    {
        send_peer_acks(state, endpoint, writer).await;
    }
    invoke_handler(&state.stats, || handler.packet_received(buffer));
}

async fn flush_pending_acks(state: &mut CoordinatorState, writer: &mpsc::Sender<WriteCommand>) {
    let endpoints: Vec<_> = state
        .peers
        .iter()
        .filter_map(|(endpoint, peer)| (!peer.pending_acks.is_empty()).then_some(*endpoint))
        .collect();
    for endpoint in endpoints {
        send_peer_acks(state, endpoint, writer).await;
    }
}

async fn send_peer_acks(
    state: &mut CoordinatorState,
    endpoint: SocketAddr,
    writer: &mpsc::Sender<WriteCommand>,
) {
    let Some(peer) = state.peers.get_mut(&endpoint) else {
        return;
    };
    if peer.pending_acks.is_empty() {
        return;
    }
    let mut blocks = Vec::with_capacity(peer.pending_acks.len());
    while let Some(id) = peer.pending_acks.pop_front() {
        blocks.push(PacketAckPacketPacketsBlock { id });
    }
    let ack_count = blocks.len();
    let Ok(mut packet) = PacketAckPacket::new_with_constructor() else {
        return;
    };
    packet.packets = blocks;
    let bytes = match packet.to_bytes_with_method() {
        Ok(mut bytes) => {
            // C# `SendAcks` explicitly marks standalone PacketAck packets as
            // unreliable before serialization.
            if let Some(flags) = bytes.first_mut() {
                *flags &= !Helpers::MSG_RELIABLE;
            }
            bytes
        }
        Err(_) => return,
    };
    let Ok((write, _)) = state.prepare_packet(bytes, endpoint, PacketType::PacketAck, false, None)
    else {
        return;
    };
    if writer.send(write).await.is_ok() {
        state.stats.acknowledgements_sent.fetch_add(
            u64::try_from(ack_count).unwrap_or(u64::MAX),
            Ordering::Relaxed,
        );
    }
}

fn invoke_handler(stats: &StatsCounters, callback: impl FnOnce()) {
    if catch_unwind(AssertUnwindSafe(callback)).is_err() {
        stats.handler_panics.fetch_add(1, Ordering::Relaxed);
    }
}

struct TokenBucket {
    tokens_per_period: usize,
    token_limit: usize,
    available: usize,
    last_replenishment: Instant,
}

impl TokenBucket {
    #[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
    fn new(bits_per_second: f32) -> Self {
        let calculated =
            (f64::from(bits_per_second) / 8.0 * THROTTLE_PERIOD.as_secs_f64()).max(0.0) as usize;
        let tokens_per_period = calculated.max(THROTTLE_MIN_BYTES_PER_PERIOD);
        let token_limit = tokens_per_period.saturating_mul(THROTTLE_BURST_PERIODS);
        Self {
            tokens_per_period,
            token_limit,
            available: token_limit,
            last_replenishment: Instant::now(),
        }
    }

    fn replenish(&mut self, now: Instant) {
        let elapsed = now.duration_since(self.last_replenishment);
        let periods = elapsed.as_nanos() / THROTTLE_PERIOD.as_nanos();
        if periods == 0 {
            return;
        }
        let periods = usize::try_from(periods).unwrap_or(usize::MAX);
        self.available = self
            .available
            .saturating_add(self.tokens_per_period.saturating_mul(periods))
            .min(self.token_limit);
        let periods_u32 = u32::try_from(periods).unwrap_or(u32::MAX);
        self.last_replenishment += THROTTLE_PERIOD.saturating_mul(periods_u32);
    }

    async fn acquire(
        &mut self,
        amount: usize,
        cancellation: &CancellationToken,
    ) -> Result<(), UdpTransportError> {
        let amount = amount.clamp(1, self.token_limit);
        loop {
            self.replenish(Instant::now());
            if self.available >= amount {
                self.available -= amount;
                return Ok(());
            }
            let deficit = amount - self.available;
            let periods = deficit.div_ceil(self.tokens_per_period);
            let periods = u32::try_from(periods).unwrap_or(u32::MAX);
            let deadline = self.last_replenishment + THROTTLE_PERIOD.saturating_mul(periods.max(1));
            tokio::select! {
                () = cancellation.cancelled() => return Err(UdpTransportError::Cancelled),
                () = tokio::time::sleep_until(deadline) => {}
            }
        }
    }
}

struct UdpThrottle {
    task: TokenBucket,
    texture: TokenBucket,
    asset: TokenBucket,
}

impl UdpThrottle {
    fn new(throttle: &AgentThrottle) -> Self {
        Self {
            task: TokenBucket::new(throttle.task()),
            texture: TokenBucket::new(throttle.texture()),
            asset: TokenBucket::new(throttle.asset()),
        }
    }

    async fn acquire(
        &mut self,
        category: UdpThrottleCategory,
        amount: usize,
        cancellation: &CancellationToken,
    ) -> Result<(), UdpTransportError> {
        match category {
            UdpThrottleCategory::Unthrottled => Ok(()),
            UdpThrottleCategory::Task => self.task.acquire(amount, cancellation).await,
            UdpThrottleCategory::Texture => self.texture.acquire(amount, cancellation).await,
            UdpThrottleCategory::Asset => self.asset.acquire(amount, cancellation).await,
        }
    }
}

async fn writer_loop(
    socket: Arc<UdpSocket>,
    mut receiver: mpsc::Receiver<WriteCommand>,
    cancellation: CancellationToken,
    initial_throttle: AgentThrottle,
    stats: Arc<StatsCounters>,
    handler: Arc<dyn UdpPacketHandler>,
) {
    let mut throttle = UdpThrottle::new(&initial_throttle);
    loop {
        let command = tokio::select! {
            biased;
            () = cancellation.cancelled() => break,
            command = receiver.recv() => command,
        };
        match command {
            Some(WriteCommand::UpdateThrottle(values)) => {
                throttle = UdpThrottle::new(&values);
            }
            Some(WriteCommand::Datagram { buffer, category }) => {
                let Ok(payload) = buffer.payload() else {
                    stats.dropped_send_queue.fetch_add(1, Ordering::Relaxed);
                    continue;
                };
                if throttle
                    .acquire(category, payload.len(), &cancellation)
                    .await
                    .is_err()
                {
                    break;
                }
                match socket.send_to(payload, buffer.remote_end_point).await {
                    Ok(bytes_sent) => {
                        stats.sent_datagrams.fetch_add(1, Ordering::Relaxed);
                        stats.sent_bytes.fetch_add(
                            u64::try_from(bytes_sent).unwrap_or(u64::MAX),
                            Ordering::Relaxed,
                        );
                        invoke_handler(&stats, || handler.packet_sent(buffer, bytes_sent));
                    }
                    Err(_) => {
                        stats.socket_errors.fetch_add(1, Ordering::Relaxed);
                    }
                }
            }
            None => break,
        }
    }
}

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

    #[test]
    fn sequence_rolls_over_across_the_full_reference_header_width() {
        let config = UdpTransportConfig::default();
        let mut peer = PeerState::new(&config).unwrap();
        peer.sequence = u32::MAX - 1;
        assert_eq!(peer.next_sequence(), u32::MAX);
        assert_eq!(peer.next_sequence(), 0);
        assert_eq!(peer.next_sequence(), 1);
    }

    #[tokio::test(start_paused = true)]
    async fn fake_time_drives_token_replenishment_and_reliable_expiry() {
        let cancellation = CancellationToken::default();
        let mut bucket = TokenBucket::new(4_000.0);
        bucket.acquire(800, &cancellation).await.unwrap();
        let started = Instant::now();
        bucket.acquire(800, &cancellation).await.unwrap();
        assert_eq!(
            Instant::now().duration_since(started),
            Duration::from_millis(400)
        );

        let config = UdpTransportConfig {
            resend_timeout: Duration::from_secs(4),
            max_resend_count: 1,
            ..UdpTransportConfig::default()
        };
        let mut coordinator = CoordinatorState {
            peers: HashMap::new(),
            remote_end_point: None,
            config,
            stats: Arc::new(StatsCounters::default()),
        };
        let destination: SocketAddr = "127.0.0.1:13000".parse().unwrap();
        let packet = vec![Helpers::MSG_RELIABLE, 0, 0, 0, 0, 0, 1];
        coordinator
            .prepare_packet(packet, destination, PacketType::ObjectUpdate, false, None)
            .unwrap();
        assert!(coordinator.take_resends().is_empty());

        tokio::time::advance(Duration::from_millis(4_001)).await;
        let resend = coordinator.take_resends();
        assert_eq!(resend.len(), 1);
        let WriteCommand::Datagram { buffer, .. } = &resend[0] else {
            panic!("expected resend datagram");
        };
        assert_ne!(buffer.data[0] & Helpers::MSG_RESENT, 0);
        assert_eq!(u32::from_be_bytes(buffer.data[1..5].try_into().unwrap()), 1);

        tokio::time::advance(Duration::from_millis(4_001)).await;
        assert!(coordinator.take_resends().is_empty());
        assert_eq!(coordinator.stats.failed_resends.load(Ordering::Relaxed), 1);
        assert!(coordinator.peers[&destination].need_ack.is_empty());
    }
}