use std::cell::RefCell;
use std::future::Future;
use std::os::unix::io::{AsRawFd, FromRawFd};
use std::pin::Pin;
use std::rc::Rc;
use std::{io, mem, net, ptr, task, collections};

use io_uring::{opcode, squeue, IoUring};

use slotmap::Key;

pub type Flags = squeue::Flags;

struct UringInner {
    ring: IoUring,
    tasks: slotmap::SlotMap<slotmap::DefaultKey, Task>,
}

#[derive(Clone)]
pub struct Uring {
    // TODO this is pretty gross; make a better concurrent interface
    inner: Rc<RefCell<UringInner>>,
}

impl Uring {
    pub fn new(entries: u32) -> io::Result<Uring> {
        let ring = IoUring::new(entries)?;
        let tasks = slotmap::SlotMap::new();
        let inner = UringInner{ring, tasks};
        let inner = Rc::new(RefCell::new(inner));

        Ok(Self{inner})
    }

    pub async fn run(&mut self, mut submission: Submission<'_>, flags: Flags) -> io::Result<Completion> {
        let entry = match submission {
            Submission::Accept { ref socket } => {
                opcode::Accept::new(
                    opcode::types::Fd(socket.as_raw_fd()),
                    ptr::null_mut(),
                    ptr::null_mut(),
                )
                .build()
            },
            Submission::Connect {
                ref socket,
                addr,
            } => {
                let (addr, size) = match addr {
                    net::SocketAddr::V4(ref a) => (
                        a as *const _ as *const _,
                        mem::size_of_val(a) as libc::socklen_t,
                    ),
                    net::SocketAddr::V6(ref a) => (
                        a as *const _ as *const _,
                        mem::size_of_val(a) as libc::socklen_t,
                    ),
                };

                opcode::Connect::new(opcode::types::Fd(socket.as_raw_fd()), addr, size).build()
            }
            Submission::Read {
                ref stream,
                ref mut buffer,
            } => {
                opcode::Read::new(
                    opcode::types::Fd(stream.as_raw_fd()),
                    buffer.as_mut_ptr(),
                    buffer.len() as _,
                )
                .build()
            }
            Submission::Write {
                ref stream,
                ref buffer,
            } => {
                opcode::Write::new(
                    opcode::types::Fd(stream.as_raw_fd()),
                    buffer.as_ptr(),
                    buffer.len() as _,
                )
                .build()
            }
        };

        let task = Task::default();

        {
            let mut inner = self.inner.borrow_mut(); // grab the lock
            let key = inner.tasks.insert(task.clone());
            let entry = entry.user_data(key.data().as_ffi()).flags(flags);

            let mut submissions = inner.ring.submission().available(); // TODO avoid refreshing every time
            unsafe {
                // TODO implement a backlog
                if let Err(_) = submissions.push(entry) {
                    return Err(io::Error::new(io::ErrorKind::WouldBlock, "uring is full"));
                }
            }
        }

        let ret = task.await;
        if ret < 0 {
            return Err(io::Error::from_raw_os_error(-ret));
        }

        Ok(match submission {
            Submission::Accept { .. } => {
                let stream = unsafe { net::TcpStream::from_raw_fd(ret) };
                Completion::Accept { stream }
            }
            Submission::Connect { socket, .. } => Completion::Connect{stream: socket},
            Submission::Read { .. } => Completion::Read{ size: ret as usize },
            Submission::Write { .. } => Completion::Write { size: ret as usize },
        })
    }

    /// Wait for progress to be made
    pub fn submit_and_wait(&mut self) -> anyhow::Result<usize> {
        let entries;

        {
            let mut inner = self.inner.borrow_mut(); // grab the lock
            inner.ring.submit_and_wait(1)?;

            let completions = inner.ring.completion().available();
            entries = completions.collect::<collections::VecDeque<_>>();
        }

        let size = entries.len();

        for entry in entries {
            let key = slotmap::KeyData::from_ffi(entry.user_data()).into();

            let mut inner = self.inner.borrow_mut(); // grab the lock again
            let mut task = match inner.tasks.remove(key) {
                Some(task) => task,
                None => anyhow::bail!("no submission available"),
            };

            task.done(entry.result());
        }

        Ok(size)
    }
}

#[derive(Default)]
struct TaskResult {
    ret: Option<i32>,
    waker: Option<task::Waker>,
}

#[derive(Default, Clone)]
struct Task {
    result: Rc<RefCell<TaskResult>>,
}

impl Task {
    fn done(&mut self, ret: i32) {
        let mut result = self.result.borrow_mut();

        result.ret = Some(ret);

        if let Some(waker) = result.waker.take() {
            waker.wake();
        }
    }
}

impl Future for Task {
    type Output = i32;

    fn poll(self: Pin<&mut Self>, cx: &mut task::Context) -> task::Poll<Self::Output> {
        let mut result = self.result.borrow_mut();

        match result.ret.take() {
            Some(ret) => task::Poll::Ready(ret),
            None => {
                result.waker = Some(cx.waker().clone());
                task::Poll::Pending
            }
        }
    }
}

#[derive(Debug)]
pub enum Submission<'a> {
    Accept {
        socket: &'a mut net::TcpListener,
    },
    Connect {
        socket: net::TcpStream,
        addr: net::SocketAddr,
    },
    Read {
        stream: &'a mut net::TcpStream,
        buffer: &'a mut [u8],
    },
    Write {
        stream: &'a mut net::TcpStream,
        buffer: &'a [u8],
    },
}

#[derive(Debug)]
pub enum Completion {
    Accept {stream: net::TcpStream},
    Connect {stream: net::TcpStream},
    Read{size: usize},
    Write{size: usize},
}
