use std::cell::RefCell;
use std::collections::HashMap;
use std::convert::Infallible;
use std::io;
use std::rc::Rc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::thread::JoinHandle;

use avril_caps::{Capabilities, Category, Pledge};
use avril_core::{Extension, Op, v8};
use avril_host::environment::Environment;
use avril_host::messaging::{Broker, Endpoint};
use futures::channel::oneshot;
use futures::task::AtomicWaker;
use tokio::sync::Notify;
use tokio_util::sync::CancellationToken;

const MAX_ID: u64 = 9_007_199_254_740_991;

pub struct Request {
    pub specifier: String,
    pub referrer: String,
    pub name: String,
    pub caps: Arc<Capabilities>,
    pub environment: Environment,
}

/// Runs on the child isolate thread. The host owns loading and runtime assembly.
pub trait Launch: Send + Sync {
    fn run(
        &self,
        request: Request,
        control: Arc<Control>,
        inbox: Arc<Endpoint>,
    ) -> Result<(), Box<dyn std::error::Error>>;
}

#[derive(Default)]
pub struct Control {
    stopped: AtomicBool,
    terminated: Arc<AtomicBool>,
    processes: CancellationToken,
    isolate: Mutex<Option<v8::IsolateHandle>>,
    notify: Notify,
    waker: AtomicWaker,
}

impl Control {
    pub fn is_stopped(&self) -> bool {
        self.stopped.load(Ordering::SeqCst)
    }

    pub fn termination_flag(&self) -> Arc<AtomicBool> {
        self.terminated.clone()
    }

    pub fn process_cancellation(&self) -> CancellationToken {
        self.processes.clone()
    }

    pub fn set_isolate(&self, isolate: Option<v8::IsolateHandle>) {
        let mut handle = self.isolate.lock().unwrap();
        *handle = isolate;
        if self.is_stopped()
            && let Some(handle) = &*handle
        {
            handle.terminate_execution();
        }
    }

    pub fn register_waker(&self, waker: &std::task::Waker) {
        self.waker.register(waker);
    }

    fn stop(&self, terminate: bool) {
        self.stopped.store(true, Ordering::SeqCst);
        self.processes.cancel();
        if terminate {
            self.terminated.store(true, Ordering::SeqCst);
            if let Some(isolate) = &*self.isolate.lock().unwrap() {
                isolate.terminate_execution();
            }
        }
        self.notify.notify_waiters();
        self.waker.wake();
    }

    pub async fn cancelled(&self) {
        loop {
            let notified = self.notify.notified();
            tokio::pin!(notified);
            notified.as_mut().enable();
            if self.stopped.load(Ordering::SeqCst) {
                return;
            }
            notified.await;
        }
    }
}

pub struct WorkerScope {
    pub control: Arc<Control>,
    pub inbox: Arc<Endpoint>,
    pub url: String,
    pub name: String,
}

struct Child {
    control: Arc<Control>,
    inbox: Arc<Endpoint>,
    outbox: Arc<Endpoint>,
    thread: Option<JoinHandle<()>>,
    completion: Option<oneshot::Receiver<Option<String>>>,
}

impl Child {
    fn terminate(&self) {
        self.control.stop(true);
        self.inbox.close();
    }
}

impl Drop for Child {
    fn drop(&mut self) {
        self.terminate();
        if let Some(thread) = self.thread.take() {
            let _ = thread.join();
        }
    }
}

/// One serialized capability pledge row inherited from the parent scope.
type PledgeRow = (String, Option<Vec<String>>, Vec<String>);

/// `op_worker_create` arguments: specifier, referrer, worker name, pledges.
type CreateArgs = (String, String, String, Vec<PledgeRow>);

pub fn init(
    entry_url: impl Fn() -> String + 'static,
    caps: Arc<Capabilities>,
    broker: Arc<Broker>,
    launch: Arc<dyn Launch>,
    environment: Rc<RefCell<Environment>>,
    worker: Option<WorkerScope>,
) -> Extension {
    let children = Rc::new(RefCell::new(HashMap::<u64, Child>::new()));
    let next = Rc::new(std::cell::Cell::new(0u64));
    let metadata = worker
        .as_ref()
        .map(|scope| (scope.inbox.id(), scope.url.clone(), scope.name.clone()));
    Extension {
        name: "ext:workers",
        dependencies: &["ext:base", "ext:url", "ext:web"],
        ops: vec![
            Op::new("op_worker_metadata", move |(): ()| {
                Ok::<_, Infallible>(metadata.clone())
            }),
            Op::new("op_worker_create", {
                let children = children.clone();
                move |(specifier, referrer, name, restrictions): CreateArgs| {
                    let environment = environment.borrow().clone();
                    let caps = Capabilities::inherit(caps.clone());
                    let launch = launch.clone();
                    let restrictions = restrictions
                        .into_iter()
                        .map(|(category, allow, deny)| {
                            category
                                .parse::<Category>()
                                .map(|category| (category, allow, deny))
                        })
                        .collect::<Result<Pledge, _>>()
                        .map_err(io::Error::other)?;
                    caps.pledge(restrictions).map_err(io::Error::other)?;
                    let (parent, inbox) = broker.pair()?;
                    let control = Arc::new(Control::default());
                    let (done, completion) = oneshot::channel();
                    let child_control = control.clone();
                    let child_inbox = inbox.clone();
                    let referrer = if referrer.is_empty() {
                        caps.cwd()
                            .join("__worker__.js")
                            .to_string_lossy()
                            .into_owned()
                    } else {
                        referrer
                    };
                    let id = next.get();
                    if id > MAX_ID {
                        return Err(io::Error::other("Worker ID space exhausted"));
                    }
                    next.set(id + 1);
                    let thread = std::thread::Builder::new()
                        .name(format!("avril:worker:{id}"))
                        .spawn(move || {
                            let result = launch.run(
                                Request {
                                    specifier,
                                    referrer,
                                    name,
                                    caps,
                                    environment,
                                },
                                child_control.clone(),
                                child_inbox.clone(),
                            );
                            child_inbox.close();
                            let error = if child_control.stopped.load(Ordering::SeqCst) {
                                None
                            } else {
                                result.err().map(|error| error.to_string())
                            };
                            let _ = done.send(error);
                        })?;
                    children.borrow_mut().insert(
                        id,
                        Child {
                            control,
                            inbox,
                            outbox: parent.clone(),
                            thread: Some(thread),
                            completion: Some(completion),
                        },
                    );
                    Ok::<_, io::Error>((id, parent.id()))
                }
            }),
            Op::asynchronous("op_worker_wait", {
                let children = children.clone();
                move |id: u64| {
                    let completion = children
                        .borrow_mut()
                        .get_mut(&id)
                        .and_then(|child| child.completion.take());
                    let outbox = children.borrow().get(&id).map(|child| child.outbox.clone());
                    let children = children.clone();
                    async move {
                        let result = match completion {
                            Some(completion) => completion
                                .await
                                .unwrap_or(Some("Worker thread stopped unexpectedly".into())),
                            None => None,
                        };
                        if let Some(outbox) = outbox {
                            outbox.drained().await;
                        }
                        children.borrow_mut().remove(&id);
                        Ok::<_, Infallible>(result)
                    }
                }
            }),
            Op::new("op_worker_terminate", move |id: u64| {
                if let Some(child) = children.borrow().get(&id) {
                    child.terminate();
                }
                Ok::<_, Infallible>(())
            }),
            Op::new("op_worker_close", move |(): ()| {
                if let Some(scope) = &worker {
                    scope.control.stop(false);
                    scope.inbox.close();
                }
                Ok::<_, Infallible>(())
            }),
            Op::new("op_worker_base", move |(): ()| {
                // The entry URL is the worker API's base, not the constructor's call site.
                Ok::<_, Infallible>(entry_url())
            }),
        ],
        modules: &[(
            "ext:workers/main.js",
            include_str!(concat!(env!("OUT_DIR"), "/main.js")),
        )],
        entry_point: Some("ext:workers/main.js"),
    }
}