use hyper::{header, Body, Request, Response, Server, StatusCode};
use hyper::service::{make_service_fn, service_fn};
use std::convert::Infallible;
use std::net::SocketAddr;
use std::sync::Arc;

use bytes::Bytes;
use futures::StreamExt;
use tokio_stream::wrappers::BroadcastStream;

use crate::systems::sys_plugin::{BroadcastEvent, BroadcastTx, PluginManager};

/// Internal handler function that logs the request and defers to the PluginManager.
async fn handle_request(
    req: Request<Body>,
    plugin_manager: Arc<PluginManager>,
) -> Result<Response<Body>, Infallible> {
    let path = req.uri().path();

    // SSE event stream for clients to subscribe to updates.
    if path == "/events" {
        return Ok(sse_response(plugin_manager.broadcast().clone()).await);
    }

    // Ask the plugin manager to handle the request.
    if let Some(response) = plugin_manager.handle_request(req).await {
        return Ok(response);
    }

    // Fallback response.
    Ok(Response::new(Body::from("No Plugin Found")))
}

/// Build an SSE response that streams all broadcast events to the client.
/// This implementation targets hyper 0.14.x.
async fn sse_response(broadcast: BroadcastTx) -> Response<Body> {
    // Subscribe to the bus
    let rx = broadcast.subscribe();

    // Convert BroadcastStream<BroadcastEvent> -> Stream<Item = Result<Bytes, Infallible>>
    let stream = BroadcastStream::new(rx).filter_map(|msg| async move {
        match msg {
            Ok(BroadcastEvent::UpdateId { id }) => {
                // SSE format: an 'update' event whose data is the ID
                let payload = format!("event:update\ndata:{}\n\n", id);
                Some(Ok::<Bytes, Infallible>(Bytes::from(payload)))
            }
            Err(_lagged) => {
                // Client fell behind; drop missed messages and continue.
                None
            }
        }
    });

    // hyper 0.14: wrap a futures::Stream as a Body
    let body = Body::wrap_stream(stream);

    let mut res = Response::new(body);
    let headers = res.headers_mut();
    headers.insert(header::CONTENT_TYPE, header::HeaderValue::from_static("text/event-stream"));
    headers.insert(header::CACHE_CONTROL, header::HeaderValue::from_static("no-cache"));
    headers.insert(header::CONNECTION, header::HeaderValue::from_static("keep-alive"));
    headers.insert(header::ACCESS_CONTROL_ALLOW_ORIGIN, header::HeaderValue::from_static("*"));
    *res.status_mut() = StatusCode::OK;
    res
}

/// Run the server on the given port using the provided PluginManager.
pub async fn run_server(port: u16, public: bool, plugin_manager: Arc<PluginManager>) {
    let addr = if public {
        SocketAddr::from(([0, 0, 0, 0], port))
    } else {
        SocketAddr::from(([127, 0, 0, 1], port))
    };

    let make_svc = make_service_fn(move |_conn| {
        let manager = plugin_manager.clone();
        async move {
            Ok::<_, Infallible>(service_fn(move |req| {
                let manager = manager.clone();
                async move { handle_request(req, manager).await }
            }))
        }
    });

    let server = Server::bind(&addr).serve(make_svc);
    println!("Server running on http://{}", addr);
    println!("SSE stream available at http://{}/events", addr);
    if let Err(e) = server.await {
        eprintln!("Server error: {}", e);
    }
}
