use std::cell::RefCell;
use std::convert::Infallible;
use std::future::Future;
use std::io::{self, Read, Write};
use std::rc::Rc;
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use std::sync::mpsc::{self, Sender, SyncSender};

use avril_core::serde_v8::{JsBuffer, ToJsBuffer};
use avril_core::{Extension, Op};
use futures::channel::oneshot;
use futures::future::{Either, select};

enum WriteReply {
    Async(oneshot::Sender<io::Result<()>>),
    Blocking(SyncSender<io::Result<()>>),
}

struct WriteRequest {
    bytes: Option<Vec<u8>>,
    reply: WriteReply,
}

struct Output(Sender<WriteRequest>);

async fn response<T>(reply: oneshot::Receiver<io::Result<T>>) -> io::Result<T> {
    reply
        .await
        .map_err(|_| io::Error::other("Stdio worker stopped"))?
}

impl Output {
    fn new(name: &str, mut writer: impl Write + Send + 'static) -> io::Result<Self> {
        let (sender, requests) = mpsc::channel::<WriteRequest>();
        std::thread::Builder::new()
            .name(name.into())
            .spawn(move || {
                let mut failure: Option<(io::ErrorKind, String)> = None;
                for request in requests.iter() {
                    let result = if let Some((kind, message)) = &failure {
                        Err(io::Error::new(*kind, message.clone()))
                    } else {
                        (|| {
                            if let Some(bytes) = request.bytes {
                                writer.write_all(&bytes)?;
                            }
                            writer.flush()
                        })()
                    };
                    if let Err(error) = &result {
                        failure = Some((error.kind(), error.to_string()));
                    }
                    match request.reply {
                        WriteReply::Async(reply) => {
                            let _ = reply.send(result);
                        }
                        WriteReply::Blocking(reply) => {
                            let _ = reply.send(result);
                        }
                    }
                }
            })?;
        Ok(Self(sender))
    }

    fn submit(&self, bytes: Option<Vec<u8>>) -> io::Result<oneshot::Receiver<io::Result<()>>> {
        let (reply, receiver) = oneshot::channel();
        // Stream writes are serialized by the sink; raw ops must not block the isolate on capacity.
        self.0
            .send(WriteRequest {
                bytes,
                reply: WriteReply::Async(reply),
            })
            .map_err(|_| io::Error::other("Stdio worker stopped"))?;
        Ok(receiver)
    }

    fn blocking_write(&self, bytes: Option<Vec<u8>>) -> io::Result<()> {
        let (reply, receiver) = mpsc::sync_channel(1);
        self.0
            .send(WriteRequest {
                bytes,
                reply: WriteReply::Blocking(reply),
            })
            .map_err(|_| io::Error::other("Stdio worker stopped"))?;
        receiver
            .recv()
            .map_err(|_| io::Error::other("Stdio worker stopped"))?
    }

    fn flush(&self) -> io::Result<()> {
        self.blocking_write(None)
    }
}

struct ReadRequest {
    limit: usize,
    reply: oneshot::Sender<io::Result<Vec<u8>>>,
}

enum InputState {
    Idle,
    Reading(oneshot::Sender<()>),
    Canceled,
}

struct PendingRead(Rc<Stdio>);

impl Drop for PendingRead {
    fn drop(&mut self) {
        if matches!(*self.0.input.borrow(), InputState::Reading(_)) {
            self.0.cancel_read();
        }
    }
}

pub struct Stdio {
    stdin: SyncSender<ReadRequest>,
    stdin_available: bool,
    stdout: Output,
    stderr: Output,
    input: RefCell<InputState>,
    interrupted: Arc<AtomicBool>,
}

impl Stdio {
    pub fn new() -> io::Result<Rc<Self>> {
        Self::new_with_stdin(true, Arc::default())
    }

    pub fn new_for_repl() -> io::Result<Rc<Self>> {
        Self::new_with_stdin(false, Arc::default())
    }

    pub fn new_for_worker(interrupted: Arc<AtomicBool>) -> io::Result<Rc<Self>> {
        Self::new_with_stdin(false, interrupted)
    }

    pub fn interrupt_flag(&self) -> Arc<AtomicBool> {
        self.interrupted.clone()
    }

    fn new_with_stdin(stdin_available: bool, interrupted: Arc<AtomicBool>) -> io::Result<Rc<Self>> {
        let (stdin, requests) = mpsc::sync_channel::<ReadRequest>(1);
        if stdin_available {
            // An outstanding OS read cannot be interrupted portably; never join this worker.
            std::thread::Builder::new()
                .name("avril:stdin".into())
                .spawn(move || {
                    for request in requests {
                        let result = (|| {
                            let mut bytes = vec![0; request.limit];
                            let mut stdin = io::stdin().lock();
                            let count = loop {
                                match stdin.read(&mut bytes) {
                                    Err(error) if error.kind() == io::ErrorKind::Interrupted => {
                                        continue;
                                    }
                                    result => break result?,
                                }
                            };
                            bytes.truncate(count);
                            Ok(bytes)
                        })();
                        let _ = request.reply.send(result);
                    }
                })?;
        } else {
            drop(requests);
        }
        Ok(Rc::new(Self {
            stdin,
            stdin_available,
            stdout: Output::new("avril:stdout", io::stdout())?,
            stderr: Output::new("avril:stderr", io::stderr())?,
            input: RefCell::new(InputState::Idle),
            interrupted,
        }))
    }

    fn read(
        self: &Rc<Self>,
        limit: usize,
    ) -> impl Future<Output = io::Result<Option<ToJsBuffer>>> + use<> {
        let started = (|| {
            if !self.stdin_available {
                return Err(io::Error::new(
                    io::ErrorKind::Unsupported,
                    "Guest stdin in the REPL is not yet implemented; the REPL owns standard input",
                ));
            }
            if limit == 0 {
                return Ok(None);
            }
            let (reply, receiver) = oneshot::channel();
            let (cancel, canceled) = oneshot::channel();
            {
                let mut input = self.input.borrow_mut();
                match &*input {
                    InputState::Reading(_) => {
                        return Err(io::Error::new(
                            io::ErrorKind::WouldBlock,
                            "A stdin read is already pending",
                        ));
                    }
                    InputState::Canceled => return Ok(None),
                    InputState::Idle => *input = InputState::Reading(cancel),
                }
            }
            let pending = PendingRead(self.clone());
            // Only one read can be submitted; cancellation permanently closes the input.
            self.stdin
                .send(ReadRequest {
                    limit: limit.min(64 * 1024),
                    reply,
                })
                .map_err(|_| io::Error::other("Stdin worker stopped"))?;
            Ok(Some((pending, receiver, canceled)))
        })();
        async move {
            let Some((pending, receiver, canceled)) = started? else {
                return Ok(if limit == 0 {
                    Some(ToJsBuffer::from(Vec::new()))
                } else {
                    None
                });
            };
            let result = match select(receiver, canceled).await {
                Either::Left((result, _)) => result
                    .map_err(|_| io::Error::other("Stdin worker stopped"))
                    .and_then(|result| result)
                    .map(|bytes| {
                        if bytes.is_empty() {
                            None
                        } else {
                            Some(ToJsBuffer::from(bytes))
                        }
                    }),
                Either::Right(_) => Ok(None),
            };
            {
                let mut input = pending.0.input.borrow_mut();
                if matches!(*input, InputState::Reading(_)) {
                    *input = InputState::Idle;
                }
            }
            result
        }
    }

    fn cancel_read(&self) {
        if let InputState::Reading(sender) = self.input.replace(InputState::Canceled) {
            let _ = sender.send(());
        }
    }

    fn output(&self, stderr: bool) -> &Output {
        if stderr { &self.stderr } else { &self.stdout }
    }

    pub fn flush(&self) -> io::Result<()> {
        let stdout = self.stdout.flush();
        let stderr = self.stderr.flush();
        stdout.and(stderr)
    }

    pub fn flush_interruptible(&self, interrupted: impl Fn() -> bool) -> io::Result<()> {
        let stdout = self.stdout.submit(None);
        let stderr = self.stderr.submit(None);
        avril_host::wait(
            async {
                let stdout = response(stdout?).await;
                let stderr = response(stderr?).await;
                stdout.and(stderr)
            },
            interrupted,
        )
        .unwrap_or(Ok(()))
    }

    pub fn print(&self, stderr: bool, text: String) -> io::Result<()> {
        self.output(stderr).blocking_write(Some(text.into_bytes()))
    }

    pub fn extension(self: &Rc<Self>) -> Extension {
        let read = self.clone();
        let read_sync = self.clone();
        let write_sync = self.clone();
        let cancel = self.clone();
        let write = self.clone();
        let flush = self.clone();
        Extension {
            name: "ext:stdio",
            dependencies: &["ext:webstreams", "ext:file"],
            ops: vec![
                Op::asynchronous("op_stdio_read", move |limit: usize| read.read(limit)),
                Op::with_scope("op_stdio_read_sync", move |scope, limit: usize| {
                    let pending = read_sync.read(limit);
                    Ok::<_, Infallible>(
                        avril_host::wait(pending, || {
                            avril_host::interrupted(scope, &read_sync.interrupted)
                        })
                        .map(avril_host::io::response),
                    )
                }),
                Op::with_scope(
                    "op_stdio_write_sync",
                    move |scope, (stderr, bytes): (bool, JsBuffer)| {
                        let count = bytes.len();
                        let pending = write_sync.output(stderr).submit(Some(bytes.to_vec()));
                        let result = avril_host::wait(
                            async { response(pending?).await.map(|()| count) },
                            || avril_host::interrupted(scope, &write_sync.interrupted),
                        );
                        Ok::<_, Infallible>(result.map(avril_host::io::response))
                    },
                ),
                Op::new("op_stdio_cancel_read", move |_: ()| {
                    cancel.cancel_read();
                    Ok::<_, io::Error>(())
                }),
                Op::asynchronous(
                    "op_stdio_write",
                    move |(stderr, bytes): (bool, JsBuffer)| {
                        let pending = write.output(stderr).submit(Some(bytes.to_vec()));
                        async move { response(pending?).await }
                    },
                ),
                Op::asynchronous("op_stdio_flush", move |stderr: bool| {
                    let pending = flush.output(stderr).submit(None);
                    async move { response(pending?).await }
                }),
            ],
            modules: &[
                (
                    "ext:stdio/main.js",
                    include_str!(concat!(env!("OUT_DIR"), "/main.js")),
                ),
                (
                    "ext:stdio/sync.js",
                    include_str!(concat!(env!("OUT_DIR"), "/sync.js")),
                ),
            ],
            entry_point: Some("ext:stdio/main.js"),
        }
    }
}

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

    fn input() -> (Rc<Stdio>, mpsc::Receiver<ReadRequest>) {
        let (stdin, requests) = mpsc::sync_channel(1);
        let stdio = Rc::new(Stdio {
            stdin,
            stdin_available: true,
            stdout: Output::new("test:stdout", io::sink()).unwrap(),
            stderr: Output::new("test:stderr", io::sink()).unwrap(),
            input: RefCell::new(InputState::Idle),
            interrupted: Arc::default(),
        });
        (stdio, requests)
    }

    #[test]
    fn reads_reserve_input_before_polling_and_empty_reads_do_not_consume_it() {
        let (stdio, requests) = input();
        let first = stdio.read(2);
        let request = requests.try_recv().unwrap();
        assert_eq!(request.limit, 2);
        let second = avril_host::wait(stdio.read(1), || false).unwrap();
        assert_eq!(second.err().unwrap().kind(), io::ErrorKind::WouldBlock);
        let empty = avril_host::wait(stdio.read(0), || false).unwrap().unwrap();
        assert!(empty.is_some());
        assert!(requests.try_recv().is_err());
        request.reply.send(Ok(vec![65, 66])).unwrap();
        assert!(
            avril_host::wait(first, || false)
                .unwrap()
                .unwrap()
                .is_some()
        );
        let next = stdio.read(1);
        let request = requests.try_recv().unwrap();
        request.reply.send(Ok(Vec::new())).unwrap();
        assert!(avril_host::wait(next, || false).unwrap().unwrap().is_none());
    }

    #[test]
    fn an_interrupted_read_does_not_leave_a_second_reader_on_the_native_queue() {
        let (stdio, requests) = input();
        let first = stdio.read(1);
        let request = requests.try_recv().unwrap();
        assert!(avril_host::wait(first, || true).is_none());
        assert!(
            avril_host::wait(stdio.read(1), || false)
                .unwrap()
                .unwrap()
                .is_none()
        );
        assert!(requests.try_recv().is_err());
        assert!(request.reply.send(Ok(vec![65])).is_err());
    }
}