use std::any::Any;
use std::cell::RefCell;
use std::io;
use std::rc::Rc;
use std::sync::Arc;

use bytes::Bytes;
use v8::{ValueDeserializerHelper, ValueSerializerHelper};

#[derive(Default)]
pub struct Message {
    bytes: Bytes,
    has_transfers: bool,
    buffers: Vec<v8::SharedRef<v8::BackingStore>>,
    shared: Vec<v8::SharedRef<v8::BackingStore>>,
    modules: Vec<Arc<v8::CompiledWasmModule>>,
    host_payloads: Vec<Arc<dyn Any + Send + Sync>>,
}

impl Message {
    pub fn try_clone(&self) -> Option<Self> {
        if self.has_transfers {
            return None;
        }
        Some(Self {
            bytes: self.bytes.clone(),
            has_transfers: false,
            buffers: Vec::new(),
            shared: self.shared.clone(),
            modules: self.modules.clone(),
            host_payloads: self.host_payloads.clone(),
        })
    }
}

#[derive(Clone, Copy, PartialEq, Eq)]
enum Direction {
    Encode,
    Decode,
}

#[derive(Clone)]
struct HostContext {
    message: Rc<RefCell<Message>>,
    direction: Direction,
}

type HostSlot = Rc<RefCell<Option<HostContext>>>;

struct HostScope {
    slot: HostSlot,
    previous: Option<HostContext>,
}

impl Drop for HostScope {
    fn drop(&mut self) {
        self.slot.replace(self.previous.take());
    }
}

fn current_message(scope: &v8::PinScope, direction: Direction) -> io::Result<Rc<RefCell<Message>>> {
    scope
        .get_slot::<HostSlot>()
        .and_then(|slot| {
            slot.borrow()
                .as_ref()
                .filter(|context| context.direction == direction)
                .map(|context| context.message.clone())
        })
        .ok_or_else(|| io::Error::other("No matching host serialization callback is active"))
}

/// Carry an opaque Rust payload outside the byte stream during host encoding.
pub fn serialize_host<T: Any + Send + Sync>(
    scope: &v8::PinScope,
    payload: Arc<T>,
) -> io::Result<u32> {
    let message = current_message(scope, Direction::Encode)?;
    let mut message = message.borrow_mut();
    let id = u32::try_from(message.host_payloads.len())
        .map_err(|_| io::Error::other("Too many host payloads"))?;
    message.host_payloads.push(payload);
    Ok(id)
}

/// Recover an opaque Rust payload during host decoding, checking its type.
pub fn deserialize_host<T: Any + Send + Sync>(scope: &v8::PinScope, id: u32) -> io::Result<Arc<T>> {
    current_message(scope, Direction::Decode)?
        .borrow()
        .host_payloads
        .get(id as usize)
        .cloned()
        .and_then(|payload| payload.downcast().ok())
        .ok_or_else(|| io::Error::other("Invalid host payload"))
}

struct Policy {
    direction: Direction,
    classify: Option<v8::Global<v8::Function>>,
    host: v8::Global<v8::Function>,
    error: v8::Global<v8::Function>,
    message: Rc<RefCell<Message>>,
}

impl Policy {
    fn call_host<'s>(
        &self,
        scope: &mut v8::PinScope<'s, '_>,
        value: v8::Local<'s, v8::Value>,
    ) -> Option<v8::Local<'s, v8::Value>> {
        let slot = scope.get_slot::<HostSlot>().cloned().unwrap_or_else(|| {
            let slot = HostSlot::default();
            scope.set_slot(slot.clone());
            slot
        });
        let previous = slot.replace(Some(HostContext {
            message: self.message.clone(),
            direction: self.direction,
        }));
        // Nested messaging must restore the outer callback's payload list, even on failure.
        let _guard = HostScope { slot, previous };
        let callback = v8::Local::new(scope, &self.host);
        let receiver = v8::undefined(scope).into();
        callback.call(scope, receiver, &[value])
    }

    fn fail(&self, scope: &mut v8::PinScope, message: &str) {
        let message = v8::String::new(scope, message).unwrap();
        self.throw(scope, message);
    }

    fn throw(&self, scope: &mut v8::PinScope, message: v8::Local<v8::String>) {
        let callback = v8::Local::new(scope, &self.error);
        let receiver = v8::undefined(scope).into();
        if let Some(error) = callback.call(scope, receiver, &[message.into()]) {
            scope.throw_exception(error);
        }
    }
}

impl v8::ValueSerializerImpl for &Policy {
    fn throw_data_clone_error<'s>(
        &self,
        scope: &mut v8::PinScope<'s, '_>,
        message: v8::Local<'s, v8::String>,
    ) {
        self.throw(scope, message);
    }

    fn has_custom_host_object(&self, _: &v8::Isolate) -> bool {
        true
    }

    fn is_host_object<'s>(
        &self,
        scope: &mut v8::PinScope<'s, '_>,
        object: v8::Local<'s, v8::Object>,
    ) -> Option<bool> {
        let callback = v8::Local::new(scope, self.classify.as_ref()?);
        let receiver = v8::undefined(scope).into();
        Some(
            callback
                .call(scope, receiver, &[object.into()])?
                .uint32_value(scope)?
                != 0,
        )
    }

    fn write_host_object<'s>(
        &self,
        scope: &mut v8::PinScope<'s, '_>,
        object: v8::Local<'s, v8::Object>,
        serializer: &dyn ValueSerializerHelper,
    ) -> Option<bool> {
        let value = self.call_host(scope, object.into())?;
        serializer.write_value(scope.get_current_context(), value)
    }

    fn get_shared_array_buffer_id<'s>(
        &self,
        scope: &mut v8::PinScope<'s, '_>,
        buffer: v8::Local<'s, v8::SharedArrayBuffer>,
    ) -> Option<u32> {
        let mut message = self.message.borrow_mut();
        let Ok(id) = u32::try_from(message.shared.len()) else {
            drop(message);
            self.fail(scope, "Too many shared buffers");
            return None;
        };
        message.shared.push(buffer.get_backing_store());
        Some(id)
    }

    fn get_wasm_module_transfer_id(
        &self,
        scope: &mut v8::PinScope<'_, '_>,
        module: v8::Local<v8::WasmModuleObject>,
    ) -> Option<u32> {
        let mut message = self.message.borrow_mut();
        let Ok(id) = u32::try_from(message.modules.len()) else {
            drop(message);
            self.fail(scope, "Too many WebAssembly modules");
            return None;
        };
        message.modules.push(Arc::new(module.get_compiled_module()));
        Some(id)
    }
}

impl v8::ValueDeserializerImpl for &Policy {
    fn read_host_object<'s>(
        &self,
        scope: &mut v8::PinScope<'s, '_>,
        deserializer: &dyn ValueDeserializerHelper,
    ) -> Option<v8::Local<'s, v8::Object>> {
        let value = deserializer.read_value(scope.get_current_context())?;
        // Host reconstruction calls trusted bootstrap code, not guest constructors.
        let allowed = std::pin::pin!(v8::AllowJavascriptExecutionScope::new(scope));
        let scope = &mut allowed.init();
        v8::Local::try_from(self.call_host(scope, value)?).ok()
    }

    fn get_shared_array_buffer_from_id<'s>(
        &self,
        scope: &mut v8::PinScope<'s, '_>,
        id: u32,
    ) -> Option<v8::Local<'s, v8::SharedArrayBuffer>> {
        Some(v8::SharedArrayBuffer::with_backing_store(
            scope,
            self.message.borrow().shared.get(id as usize)?,
        ))
    }

    fn get_wasm_module_from_id<'s>(
        &self,
        scope: &mut v8::PinScope<'s, '_>,
        id: u32,
    ) -> Option<v8::Local<'s, v8::WasmModuleObject>> {
        v8::WasmModuleObject::from_compiled_module(
            scope,
            self.message.borrow().modules.get(id as usize)?,
        )
    }
}

/// Callbacks are isolate-local; the returned message contains no V8 handles.
pub fn serialize<'s>(
    scope: &mut v8::PinScope<'s, '_>,
    value: v8::Local<'s, v8::Value>,
    transfer: v8::Local<'s, v8::Array>,
    callbacks: [v8::Local<'s, v8::Function>; 5],
) -> Option<Message> {
    let [classify, host, error, validate, commit] = callbacks;
    let policy = Policy {
        direction: Direction::Encode,
        classify: Some(v8::Global::new(scope, classify)),
        host: v8::Global::new(scope, host),
        error: v8::Global::new(scope, error),
        message: Rc::new(RefCell::new(Message {
            has_transfers: transfer.length() != 0,
            ..Default::default()
        })),
    };
    let mut buffers = Vec::new();
    let seen = v8::Set::new(scope);
    for index in 0..transfer.length() {
        let value = transfer.get_index(scope, index)?;
        if seen.has(scope, value)? {
            policy.fail(scope, "Transfer list contains a duplicate value");
            return None;
        }
        seen.add(scope, value)?;
        if let Ok(buffer) = v8::Local::<v8::ArrayBuffer>::try_from(value) {
            buffers.push(buffer);
        } else {
            let receiver = v8::undefined(scope).into();
            if classify
                .call(scope, receiver, &[value])?
                .uint32_value(scope)?
                != 2
            {
                policy.fail(scope, "Transfer list contains a non-transferable value");
                return None;
            }
        }
    }
    let receiver = v8::undefined(scope).into();
    validate.call(scope, receiver, &[])?;
    let serializer = v8::ValueSerializer::new(scope, Box::new(&policy));
    for (id, buffer) in buffers.iter().enumerate() {
        serializer.transfer_array_buffer(id as u32, *buffer);
    }
    serializer.write_header();
    if serializer.write_value(scope.get_current_context(), value) != Some(true) {
        return None;
    }
    validate.call(scope, receiver, &[])?;
    policy.message.borrow_mut().bytes = serializer.release().into();
    for buffer in buffers {
        if buffer.was_detached() {
            policy.fail(scope, "Cannot transfer a detached ArrayBuffer");
            return None;
        }
        if !buffer.is_detachable() {
            let message = v8::String::new(scope, "Cannot detach this ArrayBuffer")?;
            scope.throw_exception(v8::Exception::type_error(scope, message));
            return None;
        }
        let store = buffer.get_backing_store();
        if buffer.detach(None) != Some(true) {
            return None;
        }
        policy.message.borrow_mut().buffers.push(store);
    }
    commit.call(scope, receiver, &[])?;
    drop(serializer);
    Some(policy.message.take())
}

pub fn deserialize<'s>(
    scope: &mut v8::PinScope<'s, '_>,
    mut message: Message,
    host: v8::Local<'s, v8::Function>,
    error: v8::Local<'s, v8::Function>,
) -> Option<v8::Local<'s, v8::Value>> {
    let bytes = std::mem::take(&mut message.bytes);
    let buffers = std::mem::take(&mut message.buffers);
    let policy = Policy {
        direction: Direction::Decode,
        classify: None,
        host: v8::Global::new(scope, host),
        error: v8::Global::new(scope, error),
        message: Rc::new(RefCell::new(message)),
    };
    let deserializer = v8::ValueDeserializer::new(scope, Box::new(&policy), &bytes);
    for (id, store) in buffers.iter().enumerate() {
        let buffer = v8::ArrayBuffer::with_backing_store(scope, store);
        deserializer.transfer_array_buffer(id as u32, buffer);
    }
    let context = scope.get_current_context();
    let value = {
        v8::tc_scope!(let caught, scope);
        let value = if deserializer.read_header(context) == Some(true) {
            deserializer.read_value(context)
        } else {
            None
        };
        if value.is_none() {
            caught.reset();
        }
        value
    };
    if value.is_none() {
        policy.fail(scope, "Cannot deserialize this message");
    }
    value
}