use futures_util::{SinkExt, StreamExt};
use tokio_tungstenite::tungstenite::Message;

use super::WebSocket;
use tokio::sync::mpsc;

use super::frames::write_frames;
use crate::Frame;
use crate::engine::{self, Data, Packet as EnginePacket};
use crate::session::Input;

pub(super) async fn run_direct(
    input: mpsc::Sender<Input>,
    mut socket: WebSocket,
    mut outbound: mpsc::Receiver<Vec<Frame>>,
) {
    loop {
        tokio::select! {
            _ = input.closed() => break,
            incoming = socket.next() => {
                let Some(Ok(message)) = incoming else {
                    break;
                };
                match message {
                    Message::Text(text) => {
                        let Some(packet) = engine::decode_packet(&text) else {
                            break;
                        };
                        if input.try_send(Input::Packets(vec![packet])).is_err() {
                            break;
                        }
                    }
                    Message::Binary(data) => {
                        if input
                            .try_send(Input::Packets(vec![EnginePacket::Message(Data::Binary(
                                data.to_vec(),
                            ))]))
                            .is_err()
                        {
                            break;
                        }
                    }
                    Message::Ping(_) => {
                        if !matches!(tokio::time::timeout(std::time::Duration::from_secs(5), socket.flush()).await, Ok(Ok(()))) {
                            break;
                        }
                    }
                    Message::Pong(_) | Message::Frame(_) => {}
                    Message::Close(_) => break,
                }
            }
            frames = outbound.recv() => {
                match frames {
                    Some(frames) => {
                        if !write_frames(&mut socket, frames).await {
                            break;
                        }
                    }
                    None => break,
                }
            }
        }
    }

    let _ = tokio::time::timeout(std::time::Duration::from_secs(5), input.send(Input::Close)).await;
    let _ = tokio::time::timeout(
        std::time::Duration::from_secs(5),
        SinkExt::close(&mut socket),
    )
    .await;
}