use tokio::sync::mpsc;

use crate::backend::SocketError;

use super::super::EmitPacket;
use super::super::session::{self, Input, Output};
use super::core::{SocketHub, namespace_key};

const INPUT_CAPACITY: usize = 256;

impl SocketHub {
    pub(crate) async fn spawn_transport(
        &self,
        future: impl std::future::Future<Output = ()> + Send + 'static,
    ) {
        let mut transports = self.inner.transport_shard().lock().await;
        while let Some(result) = transports.try_join_next() {
            if let Err(error) = result {
                eprintln!("kenevent: WebSocket task failed: {error}");
            }
        }
        transports.spawn(future);
    }

    pub(crate) async fn finish_transports(&self) {
        let mut transports = Vec::with_capacity(self.inner.transports.len());
        for shard in &self.inner.transports {
            transports.push(shard.lock().await);
        }
        if tokio::time::timeout(std::time::Duration::from_secs(5), async {
            for set in &mut transports {
                while set.join_next().await.is_some() {}
            }
        })
        .await
        .is_err()
        {
            for set in &mut transports {
                set.abort_all();
            }
            for set in &mut transports {
                while set.join_next().await.is_some() {}
            }
        }
    }

    pub(crate) async fn authenticate_token(&self, token: &str) -> Result<i64, SocketError> {
        tokio::time::timeout(
            self.inner.options.auth_timeout,
            self.inner.backend.authenticate(token),
        )
        .await
        .map_err(|_| SocketError::Unavailable("authentication timed out".to_string()))?
    }

    pub(crate) async fn create_session(
        &self,
        sid: String,
        user_id: i64,
        token: String,
        output: Output,
        send_open: bool,
    ) -> mpsc::Sender<Input> {
        let (input, receiver) = mpsc::channel(INPUT_CAPACITY);
        self.inner
            .sessions_shard(&sid)
            .lock()
            .await
            .insert(sid.clone(), input.clone());

        let hub = self.clone();
        let session_input = input.clone();
        tokio::spawn(async move {
            session::run(
                hub,
                sid,
                user_id,
                token,
                session_input,
                receiver,
                output,
                send_open,
            )
            .await;
        });

        input
    }

    pub(crate) async fn remove_session(&self, sid: &str) {
        self.inner.sessions_shard(sid).lock().await.remove(sid);
    }

    pub(crate) async fn input(&self, sid: &str) -> Option<mpsc::Sender<Input>> {
        self.inner
            .sessions_shard(sid)
            .lock()
            .await
            .get(sid)
            .cloned()
    }

    pub(crate) async fn register(
        &self,
        user_id: i64,
        namespace: &str,
        session_id: String,
        input: mpsc::Sender<Input>,
    ) {
        self.inner
            .sockets_shard(user_id)
            .lock()
            .await
            .entry(user_id)
            .or_default()
            .namespaces
            .entry(namespace_key(namespace))
            .or_default()
            .insert(session_id, input);
    }

    pub(crate) async fn unregister(&self, user_id: i64, namespace: &str, session_id: &str) {
        let mut sockets = self.inner.sockets_shard(user_id).lock().await;
        if let Some(user) = sockets.get_mut(&user_id) {
            let key = namespace_key(namespace);
            if let Some(entries) = user.namespaces.get_mut(&key) {
                entries.remove(session_id);
                if entries.is_empty() {
                    user.namespaces.remove(&key);
                }
            }
            if user.namespaces.is_empty() {
                sockets.remove(&user_id);
            }
        }
    }

    pub(super) async fn inputs(&self, user_id: i64, namespace: &str) -> Vec<mpsc::Sender<Input>> {
        self.inner
            .sockets_shard(user_id)
            .lock()
            .await
            .get(&user_id)
            .and_then(|user| user.namespaces.get(namespace))
            .map(|entries| entries.values().cloned().collect())
            .unwrap_or_default()
    }

    pub(super) async fn deliver(&self, user_id: i64, namespace: &str, packet: EmitPacket) -> usize {
        let mut delivered = 0;
        for input in self.inputs(user_id, namespace).await {
            if input
                .try_send(Input::Emit {
                    packet: packet.clone(),
                    ack: None,
                    accepted: None,
                })
                .is_ok()
            {
                delivered += 1;
            }
        }
        delivered
    }
}