use serde_json::Value;
use tokio::sync::oneshot;

use super::io::Input;
use super::state::Session;
use crate::packet::{self, Kind, Packet as SocketPacket};
use crate::{EmitPacket, SocketEvent, SocketPayload};

const RESERVED_EVENTS: [&str; 6] = [
    "connect",
    "connect_error",
    "disconnect",
    "disconnecting",
    "newListener",
    "removeListener",
];

impl Session {
    pub(super) fn dispatch(&mut self, packet: SocketPacket, binaries: Vec<Vec<u8>>) -> bool {
        let mut values = match packet.data {
            Value::Array(values) => values.into_iter(),
            _ => return false,
        };
        let Some(Value::String(event)) = values.next() else {
            return false;
        };
        if RESERVED_EVENTS.contains(&event.as_str()) {
            return false;
        }

        let session_id = self
            .namespaces
            .get(&packet.namespace)
            .cloned()
            .unwrap_or_default();
        let ack = packet.id.map(|id| {
            let (sender, receiver) = oneshot::channel::<SocketPayload>();
            let input = self.input.clone();
            let namespace = packet.namespace.clone();
            tokio::spawn(async move {
                if let Ok(Ok(payload)) =
                    tokio::time::timeout(std::time::Duration::from_secs(30), receiver).await
                {
                    let _ = tokio::time::timeout(
                        std::time::Duration::from_secs(5),
                        input.send(Input::Ack {
                            namespace,
                            id,
                            payload,
                        }),
                    )
                    .await;
                }
            });
            sender
        });

        let event = SocketEvent {
            user_id: self.user_id,
            device_id: self.device_id.clone(),
            access_token: self.access_token.clone(),
            namespace: packet.namespace,
            session_id,
            event,
            data: values.collect(),
            binaries,
            ack,
        };
        self.hub.events().try_send(event).is_ok()
    }

    pub(super) fn handle_binary(&mut self, data: &[u8]) -> bool {
        let mut overflow = false;
        {
            let Some(pending) = self.pending_binary.as_mut() else {
                return true;
            };
            match pending.bytes.checked_add(data.len()) {
                Some(total) if total <= self.max_payload => {
                    pending.bytes = total;
                    pending.data.push(data.to_vec());
                }
                _ => overflow = true,
            }
        }

        if overflow {
            self.pending_binary = None;
            return false;
        }

        let complete = self
            .pending_binary
            .as_ref()
            .is_some_and(|pending| pending.data.len() == pending.packet.attachments);
        if !complete {
            return true;
        }

        let pending = self.pending_binary.take().expect("pending binary packet");
        match pending.packet.kind {
            Kind::BinaryEvent => return self.dispatch(pending.packet, pending.data),
            Kind::BinaryAck => {
                let payload = payload(pending.packet.data, pending.data);
                self.resolve_ack(&pending.packet.namespace, pending.packet.id, payload);
            }
            _ => {}
        }
        true
    }

    pub(super) fn resolve_ack(&mut self, namespace: &str, id: Option<u64>, payload: SocketPayload) {
        let Some(id) = id else {
            return;
        };
        if self
            .pending_acks
            .get(&id)
            .is_some_and(|(pending_namespace, deadline, _)| {
                pending_namespace == namespace && *deadline > tokio::time::Instant::now()
            })
        {
            if let Some((_, _, sender)) = self.pending_acks.remove(&id) {
                let _ = sender.send(payload);
            }
        }
    }

    pub(super) async fn emit(
        &mut self,
        packet: EmitPacket,
        ack: Option<(oneshot::Sender<SocketPayload>, tokio::time::Instant)>,
    ) -> bool {
        if !self.namespaces.contains_key(&packet.namespace) {
            return true;
        }
        if ack.as_ref().is_some_and(|(sender, deadline)| {
            sender.is_closed() || *deadline <= tokio::time::Instant::now()
        }) {
            return true;
        }

        let id = if ack.is_some() {
            let id = self.next_ack;
            self.next_ack = self.next_ack.wrapping_add(1);
            Some(id)
        } else {
            None
        };
        if let (Some(id), Some((ack, deadline))) = (id, ack) {
            self.pending_acks
                .insert(id, (packet.namespace.clone(), deadline, ack));
        }

        let mut data = Vec::with_capacity(packet.args.len() + 1);
        data.push(Value::String(packet.event));
        data.extend(packet.args);

        let text = if packet.binaries.is_empty() {
            packet::encode_event(&packet.namespace, id, &data)
        } else {
            packet::encode_binary_event(&packet.namespace, id, &data, packet.binaries.len())
        };
        let mut packets = Vec::with_capacity(packet.binaries.len() + 1);
        packets.push(crate::engine::text(text));
        packets.extend(packet.binaries.into_iter().map(crate::engine::binary));

        let sent = self.output.send(packets).await;
        if !sent {
            if let Some(id) = id {
                self.pending_acks.remove(&id);
            }
        }
        sent
    }

    pub(super) async fn send_ack(
        &mut self,
        namespace: String,
        id: u64,
        payload: SocketPayload,
    ) -> bool {
        if !self.namespaces.contains_key(&namespace) {
            return true;
        }

        let SocketPayload { data, binaries } = payload;
        let text = if binaries.is_empty() {
            packet::encode_ack(&namespace, id, &data)
        } else {
            packet::encode_binary_ack(&namespace, id, &data, binaries.len())
        };
        let mut packets = Vec::with_capacity(binaries.len() + 1);
        packets.push(crate::engine::text(text));
        packets.extend(binaries.into_iter().map(crate::engine::binary));
        self.output.send(packets).await
    }
}

pub(super) fn payload(data: Value, binaries: Vec<Vec<u8>>) -> SocketPayload {
    let data = match data {
        Value::Array(values) => values,
        _ => Vec::new(),
    };
    SocketPayload { data, binaries }
}