use std::convert::Infallible;
use std::io;
use std::time::Duration;

use hyper::server::conn::http1;
use hyper::service::service_fn;
use hyper_util::rt::{TokioIo, TokioTimer};
use tokio::net::TcpListener;
use tokio::sync::watch;
use tokio::task::JoinSet;

use super::SocketHub;
use super::handle::handle;

impl SocketHub {
    pub async fn serve(
        &self,
        listener: TcpListener,
        mut shutdown: watch::Receiver<bool>,
    ) -> io::Result<()> {
        let mut connections = JoinSet::new();
        let mut stopped = self.shutdown_receiver();
        let result = loop {
            if *shutdown.borrow() || *stopped.borrow() {
                break Ok(());
            }
            tokio::select! {
                _ = shutdown.changed() => break Ok(()),
                _ = stopped.changed() => break Ok(()),
                accepted = listener.accept() => {
                    let (stream, _) = match accepted {
                        Ok(connection) => connection,
                        Err(error) => break Err(error),
                    };
                    let hub = self.clone();
                    let mut stop = self.shutdown_receiver();
                    connections.spawn(async move {
                        let service = service_fn(move |request| {
                            let hub = hub.clone();
                            async move { Ok::<_, Infallible>(handle(request, hub).await) }
                        });
                        let connection = http1::Builder::new()
                            .timer(TokioTimer::new())
                            .header_read_timeout(Duration::from_secs(10))
                            .max_buf_size(32 * 1024)
                            .serve_connection(TokioIo::new(stream), service)
                            .with_upgrades();
                        tokio::pin!(connection);
                        if *stop.borrow() {
                            return;
                        }
                        tokio::select! {
                            result = &mut connection => {
                                if let Err(error) = result {
                                    eprintln!("kenevent: HTTP connection failed: {error}");
                                }
                            }
                            _ = stop.changed() => {
                                connection.as_mut().graceful_shutdown();
                                let _ = tokio::time::timeout(Duration::from_secs(5), &mut connection).await;
                            }
                        }
                    });
                }
                Some(result) = connections.join_next(), if !connections.is_empty() => {
                    if let Err(error) = result {
                        eprintln!("kenevent: HTTP task failed: {error}");
                    }
                }
            }
        };
        drop(listener);
        self.shutdown();
        if tokio::time::timeout(Duration::from_secs(5), async {
            while connections.join_next().await.is_some() {}
        })
        .await
        .is_err()
        {
            connections.abort_all();
            while connections.join_next().await.is_some() {}
        }
        self.finish_transports().await;
        result
    }
}