#![cfg(target_os = "linux")]

use std::net::SocketAddr;
use std::os::fd::AsRawFd;
use std::ptr::NonNull;
use std::time::{Duration, Instant};

use dambi::Bytes;
use dope_core::driver::{Backend, DriverConfig};
use dope_futures::conn::{Conn, ConnState};
use dope_session::Codec;
use dope_session::Framer;
use dope_session::Plain;
use dope_session::pool::server::{Server as TcpServer, TcpServerConfig};
use dope_session::profile::{Profile, Tail};
use dope_session::protocol::server::{HandlerAction, ServerProtocol};
use dope_transport::Tcp;
use dope_uring::{CqeKind, CqeRouter, CqeView, CurrentSlot, Driver, UringBackend, set_current};

const TICK_BUDGET: Duration = Duration::from_secs(5);
const MAX_CONN: usize = 8;
const REACTOR_BASE: usize = 0;
const CONN_BASE: usize = MAX_CONN + 1;

struct EchoFramer;

impl Framer for EchoFramer {
    type Head<'buf> = ();

    fn parse(&mut self, buf: &[u8]) -> Option<((), usize)> {
        if buf.is_empty() {
            None
        } else {
            Some(((), buf.len()))
        }
    }

    fn body_len(_: &()) -> u32 {
        0
    }
}

struct EchoHandler {
    framer: EchoFramer,
}

impl ServerProtocol for EchoHandler {
    type Framer = EchoFramer;
    type ConnState = ();
    type CodecLayer = Plain;

    fn framer(&mut self) -> &mut EchoFramer {
        &mut self.framer
    }

    fn handle<'buf>(&mut self, req: &'buf [u8], _: (), write: &mut [u8]) -> HandlerAction {
        let n = req.len().min(write.len());
        write[..n].copy_from_slice(&req[..n]);
        HandlerAction::Send {
            written: n as u16,
            close_after: false,
        }
    }
}

struct Composite<P: Profile> {
    reactor: TcpServer<EchoHandler, Plain, P>,
    conn: Conn<Tcp>,
}

impl<P: Profile> CqeRouter for Composite<P> {
    fn on_cqe(&mut self, driver: &mut Driver, cqe: CqeView) {
        let slot_idx = match cqe.kind() {
            CqeKind::Accept(t) | CqeKind::Recv(t) | CqeKind::Write(t) => t.parts().0,
        };
        if self.reactor.slot_id_range().contains(&slot_idx) {
            self.reactor.on_cqe(driver, cqe);
        } else if self.conn.slot_id_range().contains(&slot_idx) {
            self.conn.on_cqe(driver, cqe);
        }
    }
}

fn listener_addr<H, C, P>(reactor: &TcpServer<H, C, P>) -> SocketAddr
where
    H: ServerProtocol,
    C: Codec,
    P: Profile,
{
    let raw = reactor.listener_fd().as_fd().as_raw_fd();
    let mut storage: libc::sockaddr_in = unsafe { std::mem::zeroed() };
    let mut len = std::mem::size_of::<libc::sockaddr_in>() as libc::socklen_t;
    let rc =
        unsafe { libc::getsockname(raw, &mut storage as *mut _ as *mut libc::sockaddr, &mut len) };
    assert_eq!(
        rc,
        0,
        "getsockname failed: {}",
        std::io::Error::last_os_error()
    );
    let port = u16::from_be(storage.sin_port);
    let ip = std::net::Ipv4Addr::from(u32::from_be(storage.sin_addr.s_addr));
    SocketAddr::from((ip, port))
}

#[test]
fn reactor_and_conn_share_driver_echo_roundtrip() {
    let driver_cfg = <UringBackend as Backend>::Config::for_tcp_profile(256, MAX_CONN, 4096);
    let mut driver = <UringBackend as Backend>::new_driver(driver_cfg).expect("driver");

    let reactor_cfg = TcpServerConfig {
        max_conn: MAX_CONN,
        bind: "127.0.0.1:0".parse().unwrap(),
        backlog: 16,
        accepted_socket_opts: Default::default(),
        shutdown: None,
    };
    let reactor = TcpServer::<EchoHandler, Plain, Tail>::build_raw(
        EchoHandler { framer: EchoFramer },
        reactor_cfg,
        &mut driver,
        REACTOR_BASE,
    )
    .expect("reactor");

    let addr = listener_addr(&reactor);

    let driver_ptr = NonNull::from(&mut driver);
    set_current(Some(CurrentSlot::new(driver_ptr, std::ptr::null())));

    let conn = Conn::<Tcp>::connect(addr, &mut driver, CONN_BASE).expect("conn");

    let mut comp = Composite { reactor, conn };

    let deadline = Instant::now() + TICK_BUDGET;
    while !matches!(comp.conn.state(), ConnState::Active) {
        assert!(
            Instant::now() < deadline,
            "connect timeout: state={:?}",
            comp.conn.state()
        );
        driver.dispatch_cqes(&mut comp);
        comp.reactor.tick(&mut driver);
        comp.conn.drive(&mut driver);
        let _ = driver.submit_only();
        let _ = driver.park_for(Duration::from_millis(10));
    }

    comp.conn
        .try_send(Bytes::from_static(b"hello-share"))
        .expect("try_send");

    let deadline = Instant::now() + TICK_BUDGET;
    let mut received: Vec<u8> = Vec::new();
    while received.len() < b"hello-share".len() {
        assert!(
            Instant::now() < deadline,
            "echo recv timeout, got: {:?}",
            std::str::from_utf8(&received)
        );
        driver.dispatch_cqes(&mut comp);
        comp.reactor.tick(&mut driver);
        comp.conn.drive(&mut driver);
        while let Some(chunk) = comp.conn.try_recv() {
            received.extend_from_slice(&*chunk);
        }
        let _ = driver.submit_only();
        let _ = driver.park_for(Duration::from_millis(10));
    }
    assert_eq!(&received[..], b"hello-share");

    drop(comp);
    set_current(None);
}
