use std::convert::Infallible;
use std::io;
use std::sync::Arc;

use avril_core::Op;
use avril_core::serde_v8::{JsBuffer, ToJsBuffer};
use serde::{Deserialize, Serialize};
use tokio::sync::Mutex;
use tokio_util::sync::CancellationToken;

use super::{Authority, Resource, Sockets, closed, hostname, reply, response};

const MAX_DATAGRAM: usize = 65_507;
#[derive(Clone, Deserialize, Serialize)]
#[serde(tag = "transport", rename_all = "lowercase")]
enum Address {
    Udp { hostname: String, port: u16 },
}
impl Address {
    fn from(value: std::net::SocketAddr) -> Self {
        Self::Udp {
            hostname: value.ip().to_string(),
            port: value.port(),
        }
    }
}
pub(super) struct Datagram {
    socket: tokio::net::UdpSocket,
    authority: Authority,
    receiving: Mutex<()>,
    pub(super) canceled: CancellationToken,
}
impl Sockets {
    fn datagram(&self, id: u64) -> io::Result<Arc<Datagram>> {
        match self.resources.lock().unwrap().get(&id) {
            Some(Resource::Datagram(value)) => Ok(value.clone()),
            _ => Err(closed()),
        }
    }
    fn listen_datagram(&self, target: Address) -> io::Result<(u64, Address)> {
        let Address::Udp { hostname: host, port } = target;
        let mut authority = Authority::Net {
            caps: self.caps.clone(),
            hostname: host.clone(),
            port,
        };
        authority.check()?;
        let socket = std::net::UdpSocket::bind((hostname(&host), port))?;
        let local = socket.local_addr()?;
        if let Authority::Net { port, .. } = &mut authority {
            *port = local.port();
        }
        authority.check()?;
        socket.set_nonblocking(true)?;
        let _enter = self.runtime.enter();
        let socket = tokio::net::UdpSocket::from_std(socket)?;
        let id = self.id()?;
        self.resources.lock().unwrap().insert(
            id,
            Resource::Datagram(Arc::new(Datagram {
                socket,
                authority,
                receiving: Mutex::new(()),
                canceled: self.canceled.child_token(),
            })),
        );
        Ok((id, Address::from(local)))
    }
}
pub(super) fn ops(sockets: Arc<Sockets>) -> Vec<Op> {
    vec![
        Op::new("op_net_datagram_listen", {
            let sockets = sockets.clone();
            move |target: Address| Ok::<_, Infallible>(response(sockets.listen_datagram(target)))
        }),
        Op::new("op_net_datagram_broadcast", {
            let sockets = sockets.clone();
            move |(id, enabled): (u64, bool)| {
                let value = (|| {
                    let socket = sockets.datagram(id)?;
                    socket.authority.check()?;
                    socket.socket.set_broadcast(enabled)
                })();
                Ok::<_, Infallible>(response(value))
            }
        }),
        Op::asynchronous("op_net_datagram_receive", {
            let sockets = sockets.clone();
            let runtime = sockets.runtime.clone();
            move |(id, count): (u64, usize)| {
                let socket = sockets.datagram(id);
                reply(runtime.spawn(async move {
                    let socket = socket?;
                    let work = async {
                        let _receiving = socket.receiving.lock().await;
                        socket.authority.check()?;
                        let mut bytes = vec![0; count.min(65535)];
                        let (count, remote) = socket.socket.recv_from(&mut bytes).await?;
                        socket.authority.check()?;
                        bytes.truncate(count);
                        Ok((ToJsBuffer::from(bytes), Address::from(remote)))
                    };
                    tokio::select! {
                        biased;
                        _ = socket.canceled.cancelled() => Err(closed()),
                        value = work => value,
                    }
                }))
            }
        }),
        Op::asynchronous("op_net_datagram_send", {
            let sockets = sockets.clone();
            let runtime = sockets.runtime.clone();
            move |(id, bytes, target): (u64, JsBuffer, Address)| {
                let socket = sockets.datagram(id);
                let bytes = if bytes.len() > MAX_DATAGRAM {
                    Err(io::Error::new(
                        io::ErrorKind::InvalidInput,
                        "Datagram exceeds 65507 bytes",
                    ))
                } else {
                    Ok(bytes.to_vec())
                };
                let Address::Udp { hostname: host, port } = target;
                let authority = Authority::Net {
                    caps: sockets.caps.clone(),
                    hostname: host.clone(),
                    port,
                };
                reply(runtime.spawn(async move {
                    let socket = socket?;
                    let bytes = bytes?;
                    let work = async {
                        socket.authority.check()?;
                        authority.check()?;
                        let ipv4 = socket.socket.local_addr()?.is_ipv4();
                        let target = tokio::net::lookup_host((hostname(&host), port))
                            .await?
                            .find(|address| address.is_ipv4() == ipv4)
                            .ok_or_else(|| {
                                io::Error::new(
                                    io::ErrorKind::AddrNotAvailable,
                                    "No compatible destination address",
                                )
                            })?;
                        socket.authority.check()?;
                        authority.check()?;
                        socket.socket.send_to(&bytes, target).await
                    };
                    tokio::select! {
                        biased;
                        _ = socket.canceled.cancelled() => Err(closed()),
                        value = work => value,
                    }
                }))
            }
        }),
    ]
}