use std::cell::RefCell;
use std::future::Future;
use std::pin::Pin;
use std::rc::Rc;
use std::task::{Context, Poll};

use futures::FutureExt;
use futures::future::LocalBoxFuture;
use futures::stream::{FuturesUnordered, StreamExt};
use serde::{Serialize, de::DeserializeOwned};
use v8::MapFnTo;

trait OpTask: Future<Output = ()> {
    fn complete(self: Pin<&mut Self>, scope: &mut v8::PinScope);
}

// Keep the concrete result in the future's allocation until V8 can serialize it.
pin_project_lite::pin_project! {
    struct OpFuture<F: Future> {
        #[pin]
        future: futures::future::MaybeDone<F>,
        resolver: v8::Global<v8::PromiseResolver>,
    }
}

impl<F: Future> Future for OpFuture<F> {
    type Output = ();

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
        self.project().future.poll(cx)
    }
}

impl<F, R, E> OpTask for OpFuture<F>
where
    F: Future<Output = Result<R, E>>,
    R: Serialize,
    E: std::error::Error,
{
    fn complete(self: Pin<&mut Self>, scope: &mut v8::PinScope) {
        let this = self.project();
        let result = this
            .future
            .take_output()
            .expect("op completed before its future was ready")
            .map_err(|error| OpError::Error(error.to_string()))
            .and_then(|value| {
                serde_v8::to_v8(scope, value).map_err(|error| OpError::Error(error.to_string()))
            });
        let resolver = v8::Local::new(scope, &*this.resolver);
        match result {
            Ok(value) => {
                resolver.resolve(scope, value);
            }
            Err(error) => {
                let error = error.exception(scope);
                resolver.reject(scope, error);
            }
        }
    }
}

struct OpCompletion {
    future: Pin<Box<dyn OpTask>>,
    referenced: bool,
}

enum Completion {
    Op(OpCompletion),
    Import(crate::modules::ImportPreparation),
}

enum Pending {
    Op(Option<OpCompletion>),
    Import(LocalBoxFuture<'static, crate::modules::ImportPreparation>),
}

impl Future for Pending {
    type Output = Completion;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Completion> {
        match self.get_mut() {
            Self::Op(completion) => {
                std::task::ready!(completion.as_mut().unwrap().future.as_mut().poll(cx));
                Poll::Ready(Completion::Op(completion.take().unwrap()))
            }
            Self::Import(future) => future.as_mut().poll(cx).map(Completion::Import),
        }
    }
}

struct PendingOps {
    futures: FuturesUnordered<Pending>,
    referenced: usize,
}
enum RejectionPromise {
    Pending(v8::Global<v8::Promise>),
    Reported(v8::Weak<v8::Promise>),
}
struct Rejection {
    promise: RejectionPromise,
    reason: Option<v8::Global<v8::Value>>,
}
type Rejections = Rc<RefCell<Vec<Rejection>>>;
type HandledRejections = Rc<RefCell<Vec<(v8::Global<v8::Promise>, v8::Global<v8::Value>)>>>;

extern "C" fn promise_rejected(message: v8::PromiseRejectMessage) {
    // SAFETY: V8 invokes this callback with a promise belonging to the active isolate.
    v8::callback_scope!(unsafe scope, &message);
    let rejections = scope.get_slot::<Rejections>().unwrap().clone();
    match message.get_event() {
        v8::PromiseRejectEvent::PromiseRejectWithNoHandler => {
            let promise = message.get_promise();
            let mut rejections = rejections.borrow_mut();
            if !rejections.iter().any(|rejection| match &rejection.promise {
                RejectionPromise::Pending(promise_global) => {
                    v8::Local::new(scope, promise_global) == promise
                }
                RejectionPromise::Reported(weak) => {
                    weak.to_local(scope).is_some_and(|value| value == promise)
                }
            }) {
                rejections.push(Rejection {
                    promise: RejectionPromise::Pending(v8::Global::new(scope, promise)),
                    reason: Some(v8::Global::new(scope, message.get_value().unwrap())),
                });
            }
        }
        v8::PromiseRejectEvent::PromiseHandlerAddedAfterReject => {
            let promise = message.get_promise();
            let handled = {
                let mut rejections = rejections.borrow_mut();
                rejections
                    .iter()
                    .position(|rejection| match &rejection.promise {
                        RejectionPromise::Pending(promise_global) => {
                            v8::Local::new(scope, promise_global) == promise
                        }
                        RejectionPromise::Reported(weak) => {
                            weak.to_local(scope).is_some_and(|value| value == promise)
                        }
                    })
                    .map(|index| rejections.remove(index))
            };
            if let Some(rejection) = handled
                && matches!(rejection.promise, RejectionPromise::Reported(_))
            {
                let reason = promise.result(scope);
                let handled = scope.get_slot::<HandledRejections>().unwrap().clone();
                handled.borrow_mut().push((
                    v8::Global::new(scope, promise),
                    v8::Global::new(scope, reason),
                ));
            }
        }
        _ => {}
    }
}

pub(crate) fn mark_promise_handled<'s>(
    scope: &mut v8::PinScope<'s, '_>,
    args: v8::FunctionCallbackArguments<'s>,
    _result: v8::ReturnValue,
) {
    let Ok(promise) = v8::Local::<v8::Promise>::try_from(args.get(0)) else {
        let message = v8::String::new(scope, "Expected a promise").unwrap();
        let error = v8::Exception::type_error(scope, message);
        scope.throw_exception(error);
        return;
    };
    // MarkAsHandled also notifies our rejection tracker for already-rejected promises.
    promise.mark_as_handled();
}

/// Reads a promise's state for inspection: undefined for non-promises,
/// otherwise `[state, result]`, where state 0 is pending, 1 fulfilled and
/// 2 rejected. JavaScript cannot observe this synchronously.
pub(crate) fn promise_state<'s>(
    scope: &mut v8::PinScope<'s, '_>,
    args: v8::FunctionCallbackArguments<'s>,
    mut result: v8::ReturnValue,
) {
    let Ok(promise) = v8::Local::<v8::Promise>::try_from(args.get(0)) else {
        return;
    };
    let (state, value) = match promise.state() {
        v8::PromiseState::Pending => (0, v8::undefined(scope).into()),
        v8::PromiseState::Fulfilled => (1, promise.result(scope)),
        v8::PromiseState::Rejected => (2, promise.result(scope)),
    };
    let state = v8::Integer::new(scope, state).into();
    result.set(v8::Array::new_with_elements(scope, &[state, value]).into());
}

pub(crate) fn perform_promise_then<'s>(
    scope: &mut v8::PinScope<'s, '_>,
    args: v8::FunctionCallbackArguments<'s>,
    mut result: v8::ReturnValue,
) {
    let (Ok(promise), Ok(fulfilled), Ok(rejected)) = (
        v8::Local::<v8::Promise>::try_from(args.get(0)),
        v8::Local::<v8::Function>::try_from(args.get(1)),
        v8::Local::<v8::Function>::try_from(args.get(2)),
    ) else {
        let message = v8::String::new(scope, "Expected a promise and two handlers").unwrap();
        let error = v8::Exception::type_error(scope, message);
        scope.throw_exception(error);
        return;
    };
    // V8's API uses PerformPromiseThen, not the species-sensitive JS builtin.
    if let Some(promise) = promise.then2(scope, fulfilled, rejected) {
        result.set(promise.into());
    }
}

type Callback = dyn for<'s, 'i> Fn(
    &mut v8::PinScope<'s, 'i>,
    v8::FunctionCallbackArguments<'s>,
) -> Option<v8::Local<'s, v8::Value>>;

pub struct Op {
    pub name: &'static str,
    callback: Box<Callback>,
}

#[derive(Clone)]
pub(crate) struct RegisteredOps {
    ops: Rc<[Op]>,
}

impl RegisteredOps {
    pub(crate) fn new(mut ops: Vec<Op>) -> Result<Self, crate::Error> {
        // Snapshot indices depend on names, not extension order or native addresses.
        // Restoring requires the same op names; adding or removing an op invalidates the snapshot.
        ops.sort_unstable_by_key(|op| op.name);
        for pair in ops.windows(2) {
            if pair[0].name == pair[1].name {
                return Err(crate::Error::DuplicateOp(pair[0].name));
            }
        }
        Ok(Self { ops: ops.into() })
    }

    pub(crate) fn external_references() -> Vec<v8::ExternalReference> {
        vec![v8::ExternalReference { function: dispatch.map_fn_to() }]
    }

    pub(crate) fn object<'s>(
        &self,
        scope: &mut v8::PinScope<'s, '_>,
    ) -> Option<v8::Local<'s, v8::Object>> {
        let registry = v8::Object::new(scope);
        registry.set_prototype(scope, v8::null(scope).into())?;
        for (index, op) in self.ops.iter().enumerate() {
            let key = v8::String::new(scope, op.name)?;
            let data =
                v8::Integer::new_from_unsigned(scope, index.try_into().expect("Too many ops"));
            let function = v8::Function::builder(dispatch)
                .constructor_behavior(v8::ConstructorBehavior::Throw)
                .data(data.into())
                .build(scope)?;
            registry.set(scope, key.into(), function.into())?;
        }
        registry.set_integrity_level(scope, v8::IntegrityLevel::Frozen)?;
        Some(registry)
    }
}

enum OpError {
    Type(String),
    Error(String),
}

impl OpError {
    fn exception<'s>(&self, scope: &mut v8::PinScope<'s, '_>) -> v8::Local<'s, v8::Value> {
        let (is_type_error, message) = match self {
            Self::Type(message) => (true, message),
            Self::Error(message) => (false, message),
        };
        let message = v8::String::new(scope, message).unwrap();
        if is_type_error {
            v8::Exception::type_error(scope, message)
        } else {
            v8::Exception::error(scope, message)
        }
    }
}

impl Op {
    /// Receives all positional JavaScript arguments without deserialization.
    pub fn raw<F>(name: &'static str, function: F) -> Self
    where
        F: for<'s, 'i> Fn(
                &mut v8::PinScope<'s, 'i>,
                v8::FunctionCallbackArguments<'s>,
            ) -> Option<v8::Local<'s, v8::Value>>
            + 'static,
    {
        Self {
            name,
            callback: Box::new(function),
        }
    }

    /// A positional op that returns V8 values directly and throws host errors.
    pub fn try_raw<E>(
        name: &'static str,
        function: impl for<'s, 'i> Fn(
            &mut v8::PinScope<'s, 'i>,
            v8::FunctionCallbackArguments<'s>,
        ) -> Result<v8::Local<'s, v8::Value>, E>
        + 'static,
    ) -> Self
    where
        E: std::error::Error,
    {
        Self::with_callback(name, move |scope, args| {
            function(scope, args).map_err(|error| OpError::Error(error.to_string()))
        })
    }

    fn with_callback(
        name: &'static str,
        function: impl for<'s, 'i> Fn(
            &mut v8::PinScope<'s, 'i>,
            v8::FunctionCallbackArguments<'s>,
        ) -> Result<v8::Local<'s, v8::Value>, OpError>
        + 'static,
    ) -> Self {
        Self::raw(name, move |scope, args| match function(scope, args) {
            Ok(value) => Some(value),
            Err(error) => {
                let exception = error.exception(scope);
                scope.throw_exception(exception);
                None
            }
        })
    }

    pub fn new<A, R, E>(name: &'static str, function: impl Fn(A) -> Result<R, E> + 'static) -> Self
    where
        A: DeserializeOwned,
        R: Serialize,
        E: std::error::Error,
    {
        Self::with_scope(name, move |_, args| function(args))
    }

    /// A typed op with access to the caller's V8 scope. Pending JavaScript exceptions
    /// propagate to the caller; returning `Err` replaces them with a host error.
    pub fn with_scope<A, R, E>(
        name: &'static str,
        function: impl for<'s, 'i> Fn(&mut v8::PinScope<'s, 'i>, A) -> Result<R, E> + 'static,
    ) -> Self
    where
        A: DeserializeOwned,
        R: Serialize,
        E: std::error::Error,
    {
        Self::with_callback(name, move |scope, args| {
            let args = serde_v8::from_v8(scope, args.get(0))
                .map_err(|error| OpError::Type(error.to_string()))?;
            let result =
                function(scope, args).map_err(|error| OpError::Error(error.to_string()))?;
            serde_v8::to_v8(scope, result).map_err(|error| OpError::Error(error.to_string()))
        })
    }

    pub fn asynchronous<A, R, E, F>(name: &'static str, function: impl Fn(A) -> F + 'static) -> Self
    where
        A: DeserializeOwned,
        R: Serialize + 'static,
        E: std::error::Error + 'static,
        F: Future<Output = Result<R, E>> + 'static,
    {
        Self::asynchronous_with_liveness(name, function, true)
    }

    pub fn asynchronous_unreferenced<A, R, E, F>(
        name: &'static str,
        function: impl Fn(A) -> F + 'static,
    ) -> Self
    where
        A: DeserializeOwned,
        R: Serialize + 'static,
        E: std::error::Error + 'static,
        F: Future<Output = Result<R, E>> + 'static,
    {
        Self::asynchronous_with_liveness(name, function, false)
    }

    fn asynchronous_with_liveness<A, R, E, F>(
        name: &'static str,
        function: impl Fn(A) -> F + 'static,
        referenced: bool,
    ) -> Self
    where
        A: DeserializeOwned,
        R: Serialize + 'static,
        E: std::error::Error + 'static,
        F: Future<Output = Result<R, E>> + 'static,
    {
        Self::with_callback(name, move |scope, args| {
            let args = serde_v8::from_v8(scope, args.get(0))
                .map_err(|error| OpError::Type(error.to_string()))?;
            let future = function(args);
            let resolver = v8::PromiseResolver::new(scope).unwrap();
            let promise = resolver.get_promise(scope);
            let resolver = v8::Global::new(scope, resolver);
            let pending = scope.get_slot::<Rc<RefCell<PendingOps>>>().unwrap().clone();
            let mut pending = pending.borrow_mut();
            if referenced {
                pending.referenced += 1;
            }
            pending.futures.push(Pending::Op(Some(OpCompletion {
                future: Box::pin(OpFuture {
                    future: futures::future::maybe_done(future),
                    resolver,
                }),
                referenced,
            })));
            drop(pending);
            if let Some(waker) = scope.get_slot::<std::task::Waker>() {
                waker.wake_by_ref();
            }
            Ok(promise.into())
        })
    }
}

pub(crate) fn clear(isolate: &mut v8::Isolate) {
    isolate.remove_slot::<Rc<RefCell<PendingOps>>>();
    isolate.remove_slot::<Rejections>();
    isolate.remove_slot::<HandledRejections>();
}

pub(crate) fn init(isolate: &mut v8::Isolate) {
    isolate.set_slot::<Rejections>(Rc::new(RefCell::new(Vec::new())));
    isolate.set_slot::<HandledRejections>(Rc::new(RefCell::new(Vec::new())));
    isolate.set_promise_reject_callback(promise_rejected);
    isolate.set_slot::<Rc<RefCell<PendingOps>>>(Rc::new(RefCell::new(PendingOps {
        futures: FuturesUnordered::new(),
        referenced: 0,
    })));
}

pub(crate) fn enqueue_import(
    scope: &mut v8::PinScope,
    future: LocalBoxFuture<'static, crate::modules::ImportPreparation>,
) {
    let pending = scope.get_slot::<Rc<RefCell<PendingOps>>>().unwrap().clone();
    let mut pending = pending.borrow_mut();
    pending.referenced += 1;
    pending.futures.push(Pending::Import(future));
    drop(pending);
    if let Some(waker) = scope.get_slot::<std::task::Waker>() {
        waker.wake_by_ref();
    }
}

pub(crate) fn process_rejections(scope: &mut v8::PinScope) -> Option<v8::Global<v8::Value>> {
    let handled = scope.get_slot::<HandledRejections>().unwrap().clone();
    let reports = handled.borrow_mut().drain(..).collect::<Vec<_>>();
    for (promise, reason) in reports {
        crate::reporting::report_rejection_handled(
            scope,
            v8::Local::new(scope, &promise),
            v8::Local::new(scope, &reason),
        );
    }
    let rejections = scope.get_slot::<Rejections>().unwrap().clone();
    rejections
        .borrow_mut()
        .retain(|rejection| match &rejection.promise {
            RejectionPromise::Pending(_) => true,
            RejectionPromise::Reported(weak) => !weak.is_empty(),
        });
    let mut fatal = None;
    loop {
        let (promise, reason) = {
            let rejections = rejections.borrow();
            let Some(rejection) = rejections
                .iter()
                .find(|rejection| matches!(rejection.promise, RejectionPromise::Pending(_)))
            else {
                break;
            };
            let RejectionPromise::Pending(promise) = &rejection.promise else {
                unreachable!();
            };
            (promise.clone(), rejection.reason.as_ref().unwrap().clone())
        };
        let promise_local = v8::Local::new(scope, &promise);
        let reason_local = v8::Local::new(scope, &reason);
        let canceled =
            crate::reporting::report_unhandled_rejection(scope, promise_local, reason_local);
        let weak = v8::Weak::new(scope, promise_local);
        let mut rejections = rejections.borrow_mut();
        let Some(rejection) = rejections.iter_mut().find(|rejection| {
            matches!(&rejection.promise, RejectionPromise::Pending(promise) if v8::Local::new(scope, promise) == promise_local)
        }) else {
            continue;
        };
        rejection.promise = RejectionPromise::Reported(weak);
        rejection.reason.take();
        drop(rejections);
        if !canceled && fatal.is_none() {
            fatal = Some(reason);
        }
    }
    if let Some(error) = &fatal {
        crate::reporting::suppress_exception(scope, v8::Local::new(scope, error));
    }
    fatal
}

pub(crate) fn checkpoint(scope: &mut v8::PinScope) {
    loop {
        scope.perform_microtask_checkpoint();
        if !v8::Platform::pump_message_loop(&v8::V8::get_current_platform(), scope, false) {
            break;
        }
    }
}

pub(crate) fn has_pending_work(isolate: &v8::Isolate) -> bool {
    !isolate
        .get_slot::<Rc<RefCell<PendingOps>>>()
        .unwrap()
        .borrow()
        .futures
        .is_empty()
}

pub(crate) fn poll_event_loop(scope: &mut v8::PinScope, context: &mut Context<'_>) -> Option<bool> {
    if crate::tasks::run_next(scope)? {
        return Some(true);
    }
    let pending = scope.get_slot::<Rc<RefCell<PendingOps>>>().unwrap().clone();
    let completion = pending.borrow_mut().futures.poll_next_unpin(context);
    let Poll::Ready(Some(completion)) = completion else {
        return Some(false);
    };

    match completion {
        Completion::Import(preparation) => {
            pending.borrow_mut().referenced -= 1;
            crate::modules::import_prepared(scope, preparation);
        }
        Completion::Op(mut completion) => {
            if completion.referenced {
                pending.borrow_mut().referenced -= 1;
            }
            completion.future.as_mut().complete(scope);
        }
    }
    Some(true)
}

pub(crate) fn run_event_loop(
    scope: &mut v8::PinScope,
    promise: Option<v8::Local<v8::Promise>>,
    wait_for_idle: bool,
) -> Option<()> {
    let pending = scope.get_slot::<Rc<RefCell<PendingOps>>>().unwrap().clone();
    loop {
        checkpoint(scope);
        if scope.is_execution_terminating() {
            return None;
        }
        if let Some(error) = crate::microtasks::process_exceptions(scope) {
            let error = v8::Local::new(scope, error);
            crate::reporting::mark_processed_exception(scope, error);
            scope.throw_exception(error);
            return None;
        }
        if let Some(promise) = promise
            && promise.state() == v8::PromiseState::Rejected
        {
            let error = promise.result(scope);
            crate::reporting::report_exception_for_value(scope, error);
            crate::reporting::mark_processed_exception(scope, error);
            scope.throw_exception(error);
            return None;
        }
        if let Some(error) = process_rejections(scope) {
            let error = v8::Local::new(scope, error);
            scope.throw_exception(error);
            return None;
        }
        if !wait_for_idle
            && promise.is_some_and(|promise| promise.state() == v8::PromiseState::Fulfilled)
        {
            return Some(());
        }
        if crate::tasks::run_next(scope)? {
            continue;
        }
        let completion = if pending.borrow().referenced == 0 {
            let mut pending = pending.borrow_mut();
            let Some(completion) = pending.futures.next().now_or_never().flatten() else {
                return match promise.map(|promise| promise.state()) {
                    None | Some(v8::PromiseState::Fulfilled) => Some(()),
                    _ => {
                        let message = v8::String::new(
                            scope,
                            "Module evaluation is still pending with no runnable host work",
                        )
                        .unwrap();
                        let error = v8::Exception::error(scope, message);
                        scope.throw_exception(error);
                        None
                    }
                };
            };
            completion
        } else {
            futures::executor::block_on(pending.borrow_mut().futures.next()).unwrap()
        };
        let mut completion = match completion {
            Completion::Op(completion) => completion,
            Completion::Import(preparation) => {
                pending.borrow_mut().referenced -= 1;
                crate::modules::import_prepared(scope, preparation);
                continue;
            }
        };
        if completion.referenced {
            pending.borrow_mut().referenced -= 1;
        }
        completion.future.as_mut().complete(scope);
    }
}

fn dispatch<'s>(
    scope: &mut v8::PinScope<'s, '_>,
    args: v8::FunctionCallbackArguments<'s>,
    mut result: v8::ReturnValue,
) {
    // Retain the closures across guest reentrancy without borrowing the isolate slot.
    let ops = scope.get_slot::<RegisteredOps>().cloned();
    let op = v8::Local::<v8::Uint32>::try_from(args.data())
        .ok()
        .and_then(|index| ops.as_ref()?.ops.get(index.value() as usize));
    let Some(op) = op else {
        let error = OpError::Error("Op is not registered in this isolate".into()).exception(scope);
        scope.throw_exception(error);
        return;
    };
    if let Some(value) = (op.callback)(scope, args) {
        result.set(value);
    }
}

#[cfg(test)]
mod tests;