use std::cell::RefCell;
use std::collections::{BTreeMap, HashMap, VecDeque};
use std::convert::Infallible;
use std::rc::Rc;
use std::sync::{Arc, Mutex, Weak};

use avril_core::{Op, message, v8};
use tokio::sync::Notify;

type Id = u64;

const MAX_ID: u64 = 9_007_199_254_740_991;

fn u64_arg<'s, 'i>(
    scope: &mut v8::PinScope<'s, 'i>,
    value: v8::Local<'s, v8::Value>,
) -> Option<u64> {
    match avril_core::serde_v8::from_v8::<u64>(scope, value) {
        Ok(id) => Some(id),
        Err(error) => {
            let message = v8::String::new(scope, &error.to_string())?;
            scope.throw_exception(v8::Exception::type_error(scope, message));
            None
        }
    }
}

fn port_id<'s, 'i>(
    scope: &mut v8::PinScope<'s, 'i>,
    value: v8::Local<'s, v8::Value>,
) -> Option<Id> {
    u64_arg(scope, value)
}

struct Packet {
    message: message::Message,
    _transfers: Vec<Arc<Endpoint>>,
}

#[derive(Default)]
struct State {
    queue: VecDeque<Packet>,
    epoch: u64,
    closed: bool,
}

pub struct Endpoint {
    id: Id,
    broker: Weak<Broker>,
    peer: Mutex<Weak<Endpoint>>,
    state: Mutex<State>,
    notify: Notify,
}

impl Endpoint {
    pub fn close(&self) {
        let mut state = self.state.lock().unwrap();
        state.closed = true;
        state.queue.clear();
        drop(state);
        self.notify.notify_waiters();
        if let Some(peer) = self.peer.lock().unwrap().upgrade() {
            peer.notify.notify_waiters();
        }
    }

    pub fn id(&self) -> Id {
        self.id
    }

    pub async fn drained(&self) {
        loop {
            let notified = self.notify.notified();
            tokio::pin!(notified);
            notified.as_mut().enable();
            if self.state.lock().unwrap().queue.is_empty() {
                return;
            }
            notified.await;
        }
    }

    async fn wait(&self, epoch: u64) -> bool {
        loop {
            let notified = self.notify.notified();
            tokio::pin!(notified);
            notified.as_mut().enable();
            let peer_closed = self
                .peer
                .lock()
                .unwrap()
                .upgrade()
                .is_none_or(|peer| peer.state.lock().unwrap().closed);
            {
                let state = self.state.lock().unwrap();
                if state.closed || state.epoch != epoch {
                    return false;
                }
                if !state.queue.is_empty() {
                    return true;
                }
                if peer_closed {
                    return false;
                }
            }
            notified.await;
        }
    }
}

impl Drop for Endpoint {
    fn drop(&mut self) {
        if let Some(broker) = self.broker.upgrade()
            && let Ok(mut registry) = broker.0.lock()
        {
            registry.ports.remove(&self.id);
        }
        if let Some(peer) = self.peer.get_mut().unwrap().upgrade() {
            peer.notify.notify_waiters();
        }
    }
}

#[derive(Default)]
struct Registry {
    next: Id,
    ports: HashMap<Id, Weak<Endpoint>>,
    broadcasts: HashMap<Vec<u16>, BTreeMap<Id, Weak<Endpoint>>>,
}

#[derive(Default)]
pub struct Broker(Mutex<Registry>);

impl Broker {
    pub fn pair(self: &Arc<Self>) -> std::io::Result<(Arc<Endpoint>, Arc<Endpoint>)> {
        let mut registry = self.0.lock().unwrap();
        let id = registry.next;
        if id > MAX_ID - 1 {
            return Err(std::io::Error::other("MessagePort ID space exhausted"));
        }
        registry.next = id + 2;
        let first = Arc::new(Endpoint {
            id,
            broker: Arc::downgrade(self),
            peer: Mutex::default(),
            state: Mutex::default(),
            notify: Notify::new(),
        });
        let second = Arc::new(Endpoint {
            id: id + 1,
            broker: Arc::downgrade(self),
            peer: Mutex::default(),
            state: Mutex::default(),
            notify: Notify::new(),
        });
        *first.peer.lock().unwrap() = Arc::downgrade(&second);
        *second.peer.lock().unwrap() = Arc::downgrade(&first);
        registry.ports.insert(id, Arc::downgrade(&first));
        registry.ports.insert(id + 1, Arc::downgrade(&second));
        Ok((first, second))
    }

    pub fn endpoint(&self, id: Id) -> Option<Arc<Endpoint>> {
        self.0
            .lock()
            .unwrap()
            .ports
            .get(&id)
            .and_then(Weak::upgrade)
    }
}

struct Subscription {
    id: Id,
    name: Vec<u16>,
    sender: Arc<Endpoint>,
    broker: Arc<Broker>,
}

impl Drop for Subscription {
    fn drop(&mut self) {
        self.sender.close();
        let Ok(mut registry) = self.broker.0.lock() else {
            return;
        };
        if let Some(group) = registry.broadcasts.get_mut(&self.name) {
            group.remove(&self.id);
            if group.is_empty() {
                registry.broadcasts.remove(&self.name);
            }
        }
    }
}

struct Ports {
    broker: Arc<Broker>,
    owned: RefCell<HashMap<Id, Arc<Endpoint>>>,
    moving: RefCell<Vec<Arc<Endpoint>>>,
    broadcasts: RefCell<HashMap<Id, Subscription>>,
}

impl Ports {
    fn resolve(&self, id: Id) -> Option<Arc<Endpoint>> {
        if let Some(endpoint) = self.owned.borrow().get(&id).cloned() {
            return Some(endpoint);
        }
        self.broker.endpoint(id)
    }
}

impl Drop for Ports {
    fn drop(&mut self) {
        for (_, endpoint) in self.owned.get_mut().drain() {
            endpoint.close();
        }
    }
}

pub fn ops(broker: Arc<Broker>) -> Vec<Op> {
    let ports = Rc::new(Ports {
        broker: broker.clone(),
        owned: RefCell::default(),
        moving: RefCell::default(),
        broadcasts: RefCell::default(),
    });
    vec![
        Op::raw("op_broadcast_open", {
            let ports = ports.clone();
            move |scope, args| {
                let value = v8::Local::<v8::String>::try_from(args.get(0)).ok()?;
                let mut name = vec![0; value.length()];
                value.write_v2(scope, 0, &mut name, v8::WriteFlags::empty());
                let (sender, receiver) = match ports.broker.pair() {
                    Ok(pair) => pair,
                    Err(error) => {
                        let message = v8::String::new(scope, &error.to_string())?;
                        scope.throw_exception(v8::Exception::error(scope, message));
                        return None;
                    }
                };
                let id = receiver.id;
                ports.owned.borrow_mut().insert(id, receiver.clone());
                ports
                    .broker
                    .0
                    .lock()
                    .unwrap()
                    .broadcasts
                    .entry(name.clone())
                    .or_default()
                    .insert(id, Arc::downgrade(&receiver));
                ports.broadcasts.borrow_mut().insert(
                    id,
                    Subscription {
                        id,
                        name,
                        sender,
                        broker: ports.broker.clone(),
                    },
                );
                Some(v8::Number::new(scope, id as f64).into())
            }
        }),
        Op::new("op_broadcast_close", {
            let ports = ports.clone();
            move |id: Id| {
                ports.broadcasts.borrow_mut().remove(&id);
                Ok::<_, Infallible>(())
            }
        }),
        Op::raw("op_broadcast_send", {
            let ports = ports.clone();
            move |scope, args| {
                let id = port_id(scope, args.get(0))?;
                let name = ports.broadcasts.borrow().get(&id)?.name.clone();
                let transfer = v8::Array::new(scope, 0);
                let callbacks = [
                    v8::Local::<v8::Function>::try_from(args.get(2)).ok()?,
                    v8::Local::<v8::Function>::try_from(args.get(3)).ok()?,
                    v8::Local::<v8::Function>::try_from(args.get(4)).ok()?,
                    v8::Local::<v8::Function>::try_from(args.get(5)).ok()?,
                    v8::Local::<v8::Function>::try_from(args.get(6)).ok()?,
                ];
                let message = message::serialize(scope, args.get(1), transfer, callbacks)?;
                let recipients: Vec<_> = ports
                    .broker
                    .0
                    .lock()
                    .unwrap()
                    .broadcasts
                    .get(&name)
                    .into_iter()
                    .flat_map(|group| group.iter())
                    .filter(|(recipient, _)| **recipient != id)
                    .filter_map(|(_, endpoint)| endpoint.upgrade())
                    .collect();
                for endpoint in recipients {
                    let mut state = endpoint.state.lock().unwrap();
                    if !state.closed {
                        state.queue.push_back(Packet {
                            message: message
                                .try_clone()
                                .expect("Broadcast messages have no transfers"),
                            _transfers: Vec::new(),
                        });
                    }
                    drop(state);
                    endpoint.notify.notify_waiters();
                }
                Some(v8::undefined(scope).into())
            }
        }),
        Op::new("op_port_pair", {
            let ports = ports.clone();
            move |(): ()| {
                let (a, b) = ports.broker.pair()?;
                ports
                    .owned
                    .borrow_mut()
                    .extend([(a.id, a.clone()), (b.id, b.clone())]);
                Ok::<_, std::io::Error>((a.id, b.id))
            }
        }),
        Op::new("op_port_adopt", {
            let ports = ports.clone();
            move |id: Id| {
                let endpoint = ports
                    .resolve(id)
                    .ok_or_else(|| std::io::Error::other("MessagePort endpoint is closed"))?;
                ports.owned.borrow_mut().insert(id, endpoint.clone());
                ports.moving.borrow_mut().retain(|port| port.id != id);
                Ok::<_, std::io::Error>(endpoint.state.lock().unwrap().epoch)
            }
        }),
        Op::new("op_port_move", {
            let ports = ports.clone();
            move |id: Id| {
                let endpoint = ports
                    .owned
                    .borrow_mut()
                    .remove(&id)
                    .ok_or_else(|| std::io::Error::other("Transferred port is not owned"))?;
                endpoint.state.lock().unwrap().epoch += 1;
                endpoint.notify.notify_waiters();
                ports.moving.borrow_mut().push(endpoint);
                Ok::<_, std::io::Error>(())
            }
        }),
        Op::new("op_port_close", {
            let ports = ports.clone();
            move |id: Id| {
                if let Some(endpoint) = ports.owned.borrow_mut().remove(&id) {
                    endpoint.close();
                }
                Ok::<_, Infallible>(())
            }
        }),
        Op::new("op_port_peer", {
            let ports = ports.clone();
            move |id: Id| {
                Ok::<_, Infallible>(ports.resolve(id).and_then(|endpoint| {
                    endpoint.peer.lock().unwrap().upgrade().map(|peer| peer.id)
                }))
            }
        }),
        Op::asynchronous_unreferenced("op_port_wait", {
            let ports = ports.clone();
            move |(id, epoch): (Id, u64)| {
                let endpoint = ports.resolve(id);
                async move {
                    Ok::<_, Infallible>(match endpoint {
                        Some(endpoint) => endpoint.wait(epoch).await,
                        None => false,
                    })
                }
            }
        }),
        Op::asynchronous("op_port_wait_ref", {
            let ports = ports.clone();
            move |(id, epoch): (Id, u64)| {
                let endpoint = ports.resolve(id);
                async move {
                    Ok::<_, Infallible>(match endpoint {
                        Some(endpoint) => endpoint.wait(epoch).await,
                        None => false,
                    })
                }
            }
        }),
        Op::raw("op_port_send", {
            let ports = ports.clone();
            move |scope, args| {
                let id_value = args.get(0);
                let id = if id_value.is_null() {
                    None
                } else {
                    Some(port_id(scope, id_value)?)
                };
                let value = args.get(1);
                let transfer = v8::Local::<v8::Array>::try_from(args.get(2)).ok()?;
                let callbacks = [
                    v8::Local::<v8::Function>::try_from(args.get(3)).ok()?,
                    v8::Local::<v8::Function>::try_from(args.get(4)).ok()?,
                    v8::Local::<v8::Function>::try_from(args.get(5)).ok()?,
                    v8::Local::<v8::Function>::try_from(args.get(6)).ok()?,
                    v8::Local::<v8::Function>::try_from(args.get(7)).ok()?,
                ];
                let serialized = message::serialize(scope, value, transfer, callbacks);
                let transfers = std::mem::take(&mut *ports.moving.borrow_mut());
                let packet = Packet {
                    message: serialized?,
                    _transfers: transfers,
                };
                let endpoint = id.and_then(|id| ports.resolve(id));
                if let Some(endpoint) = endpoint
                    && !endpoint.state.lock().unwrap().closed
                    && let Some(peer) = endpoint.peer.lock().unwrap().upgrade()
                {
                    let mut state = peer.state.lock().unwrap();
                    if !state.closed {
                        state.queue.push_back(packet);
                    }
                    drop(state);
                    peer.notify.notify_waiters();
                }
                Some(v8::undefined(scope).into())
            }
        }),
        Op::raw("op_port_take", {
            let ports = ports.clone();
            move |scope, args| {
                let id = port_id(scope, args.get(0))?;
                let epoch = u64_arg(scope, args.get(1))?;
                let host = v8::Local::<v8::Function>::try_from(args.get(2)).ok()?;
                let error = v8::Local::<v8::Function>::try_from(args.get(3)).ok()?;
                let Some(endpoint) = ports.resolve(id) else {
                    return Some(v8::null(scope).into());
                };
                let mut state = endpoint.state.lock().unwrap();
                if state.epoch != epoch || state.closed {
                    return Some(v8::null(scope).into());
                }
                let packet = state.queue.pop_front();
                drop(state);
                endpoint.notify.notify_waiters();
                match packet {
                    Some(packet) => message::deserialize(scope, packet.message, host, error),
                    None => Some(v8::null(scope).into()),
                }
            }
        }),
    ]
}