//! A synchronous, dependency-free adapter driving a [`SoeMultiplexer`] over a
//! blocking [`std::net::UdpSocket`]. Always available; pulls in no async runtime.
//!
//! The I/O-agnostic [`SoeMultiplexer`] is runtime-agnostic; this module is a thin
//! convenience layer mirroring `TokioSoeSocket` (behind the `tokio` feature) for callers who
//! prefer a plain blocking loop. Both types implement [`SoeSocket`].

use std::io;
use std::net::{SocketAddr, UdpSocket};
use std::time::{Duration, Instant};

use bytes::Bytes;

use crate::protocol::DisconnectReason;
use crate::session::Channel;
use crate::socket::{SocketConfig, SocketEvent, SoeMultiplexer, SoeSocket};

/// Buffer size for a single received datagram. SOE UDP lengths default to 512 and
/// rarely exceed it.
const RECV_BUFFER_SIZE: usize = 2048;

/// A synchronous SOE socket: a [`SoeMultiplexer`] driven over a blocking
/// [`std::net::UdpSocket`].
///
/// Drive it by repeatedly calling [`step`](SyncSoeSocket::step), which performs a
/// single read-or-tick cycle and returns any [`SocketEvent`]s produced. The socket
/// is given a read timeout equal to the tick period, so `step` returns promptly when
/// a datagram arrives and otherwise wakes once per tick to run housekeeping.
#[derive(Debug)]
pub struct SyncSoeSocket {
    mux: SoeMultiplexer<SocketAddr>,
    socket: UdpSocket,
    buf: Box<[u8]>,
}

impl SyncSoeSocket {
    /// Binds a UDP socket to `local` and prepares to drive sessions, waking at least
    /// once every `tick_period` to run housekeeping. A period of 1–10ms is typical.
    pub fn bind(
        local: SocketAddr,
        config: SocketConfig,
        tick_period: Duration,
    ) -> io::Result<Self> {
        let socket = UdpSocket::bind(local)?;
        // A read timeout paces the loop: recv_from returns immediately on data, or
        // after the tick period so heartbeats, timeouts, and resends still run.
        socket.set_read_timeout(Some(tick_period))?;

        Ok(Self {
            mux: SoeMultiplexer::new(config),
            socket,
            buf: vec![0u8; RECV_BUFFER_SIZE].into_boxed_slice(),
        })
    }

    /// Performs a single drive cycle: waits up to the tick period for an incoming
    /// datagram, runs a session tick, flushes outgoing datagrams, and returns any
    /// events.
    pub fn step(&mut self) -> io::Result<Vec<SocketEvent<SocketAddr>>> {
        match self.socket.recv_from(&mut self.buf) {
            Ok((len, from)) => {
                let datagram = Bytes::copy_from_slice(&self.buf[..len]);
                self.mux.process_incoming(from, datagram, Instant::now());
            }
            // A read timeout surfaces as WouldBlock or TimedOut depending on the
            // platform; both simply mean "no datagram this tick".
            Err(e)
                if e.kind() == io::ErrorKind::WouldBlock || e.kind() == io::ErrorKind::TimedOut => {
            }
            Err(e) => return Err(e),
        }

        self.mux.run_tick(Instant::now());

        for (addr, datagram) in self.mux.take_outgoing() {
            self.socket.send_to(&datagram, addr)?;
        }

        Ok(self.mux.take_events())
    }
}

impl SoeSocket for SyncSoeSocket {
    fn local_addr(&self) -> io::Result<SocketAddr> {
        self.socket.local_addr()
    }

    fn session_count(&self) -> usize {
        self.mux.session_count()
    }

    fn connect(&mut self, remote: SocketAddr) {
        self.mux.connect(remote, Instant::now());
    }

    fn enqueue_data(&mut self, remote: &SocketAddr, data: &[u8]) -> bool {
        self.mux.enqueue_data(remote, data)
    }

    fn enqueue_data_on(&mut self, remote: &SocketAddr, data: &[u8], channel: Channel) -> bool {
        self.mux.enqueue_data_on(remote, data, channel)
    }

    fn terminate(&mut self, remote: &SocketAddr, reason: DisconnectReason) {
        self.mux.terminate(remote, reason, Instant::now());
    }
}

impl Drop for SyncSoeSocket {
    /// On shutdown, notify every connected peer with a session-close so they can
    /// prune their connection state immediately instead of waiting for a timeout.
    /// The disconnects are sent best-effort over the blocking socket.
    fn drop(&mut self) {
        for (addr, datagram) in self
            .mux
            .terminate_all(DisconnectReason::Application, Instant::now())
        {
            let _ = self.socket.send_to(&datagram, addr);
        }
    }
}

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

    const APP_PROTOCOL: &str = "SoeSyncDropTest";

    fn config(seed: u64) -> SocketConfig {
        let mut params = SessionParameters {
            application_protocol: APP_PROTOCOL.to_owned(),
            ..SessionParameters::default()
        };
        // Disable timeouts so the test can't accidentally pass via an inactivity
        // timeout instead of the disconnect we're actually testing.
        params.inactivity_timeout = Duration::ZERO;
        params.heartbeat_after = Duration::ZERO;
        SocketConfig {
            default_session_params: params,
            base_rng_seed: seed,
            ..SocketConfig::default()
        }
    }

    fn bind(seed: u64) -> SyncSoeSocket {
        SyncSoeSocket::bind(
            "127.0.0.1:0".parse().unwrap(),
            config(seed),
            Duration::from_millis(2),
        )
        .expect("bind loopback socket")
    }

    /// Drives both sockets until the client observes its session opening (by which
    /// point the server has responded and its session is `Running`).
    fn drive_until_open(server: &mut SyncSoeSocket, client: &mut SyncSoeSocket) {
        for _ in 0..500 {
            let _ = server.step().unwrap();
            for event in client.step().unwrap() {
                if matches!(event, SocketEvent::SessionOpened { .. }) {
                    return;
                }
            }
        }
        panic!("client session never opened");
    }

    #[test]
    fn drop_notifies_peer_with_session_close() {
        let mut server = bind(2);
        let mut client = bind(1);
        let server_addr = server.local_addr().unwrap();

        client.connect(server_addr);
        drive_until_open(&mut server, &mut client);

        // The server application terminates: its `Drop` should say goodbye to the
        // still-connected client.
        drop(server);

        // The client should observe the close promptly, without relying on a timeout
        // (which is disabled here).
        let mut closed_reason = None;
        for _ in 0..500 {
            for event in client.step().unwrap() {
                if let SocketEvent::SessionClosed { reason, .. } = event {
                    closed_reason = Some(reason);
                }
            }
            if closed_reason.is_some() {
                break;
            }
        }

        assert_eq!(closed_reason, Some(DisconnectReason::Application));
        assert_eq!(client.session_count(), 0);
    }
}
