eggserve-core 0.1.0

use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;
use std::{fs::File, path::PathBuf};

use eggserve_core::config::{ServeConfig, ServeState};
use eggserve_core::primitives::body::BodySource;
use eggserve_core::primitives::canonical::{
    normalize_response, NormalizeRequest, Response, ResponseBody, StatusCode,
};
use eggserve_core::primitives::request::Request;
use eggserve_core::server::config::RuntimeConfig;
use eggserve_core::server::connection::serve_connection_with_runtime_state;
use eggserve_core::server::{service_fn, RuntimeState, Server};
use hyper_util::rt::TokioIo;
use tempfile::TempDir;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;
use tokio::sync::broadcast;

fn build_state(tmp: &TempDir) -> Arc<ServeState> {
    let config = Arc::new(ServeConfig {
        root: tmp.path().to_path_buf(),
        ..ServeConfig::default()
    });
    Arc::new(ServeState::new(config).unwrap())
}

#[test]
fn static_root_is_validated_during_build() {
    let missing =
        std::env::temp_dir().join(format!("eggserve-plan109-missing-{}", std::process::id()));
    let result = Server::builder()
        .runtime(RuntimeConfig::default())
        .serve_config(Arc::new(ServeConfig {
            root: missing,
            ..ServeConfig::default()
        }))
        .build();
    assert!(
        result.is_err(),
        "invalid static root must fail during build"
    );
}

#[tokio::test]
async fn panic_in_service_returns_500() {
    let tmp = TempDir::new().unwrap();
    let _state = build_state(&tmp);
    let config = RuntimeConfig::default();
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let (tx, _rx) = broadcast::channel::<()>(1);

    let server = tokio::spawn(async move {
        let (stream, _) = listener.accept().await.unwrap();
        let io = TokioIo::new(stream);
        let mut shutdown_rx = tx.subscribe();
        let svc = service_fn(|_req: Request| async {
            panic!("intentional panic");
        });
        serve_connection_with_runtime_state(
            io,
            svc,
            &config,
            Arc::new(RuntimeState::new_for_testing(config.max_file_streams)),
            &mut shutdown_rx,
            1,
            addr,
            addr,
            false,
            None,
        )
        .await;
    });

    let mut client = tokio::net::TcpStream::connect(addr).await.unwrap();
    client
        .write_all(b"GET /test HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n")
        .await
        .unwrap();
    let mut buf = Vec::new();
    let _ = client.read_to_end(&mut buf).await;
    let _ = server.await;

    // When a service panics, the connection task panics and the TCP
    // connection is dropped. The client sees a connection reset with no
    // complete HTTP response.
    let response = String::from_utf8_lossy(&buf);
    assert!(
        !response.starts_with("HTTP/1.1 200"),
        "service panic should not produce 200: {}",
        response
    );
}

#[tokio::test]
async fn slow_handler_returns_504() {
    let tmp = TempDir::new().unwrap();
    let _state = build_state(&tmp);
    let config = RuntimeConfig::builder()
        .handler_timeout(Duration::from_millis(50))
        .build()
        .unwrap();
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let (tx, _rx) = broadcast::channel::<()>(1);

    let server = tokio::spawn(async move {
        let (stream, _) = listener.accept().await.unwrap();
        let io = TokioIo::new(stream);
        let mut shutdown_rx = tx.subscribe();
        let svc = service_fn(|_req: Request| async {
            tokio::time::sleep(Duration::from_secs(60)).await;
            Ok(Response::builder()
                .status(StatusCode::OK)
                .body(ResponseBody::Empty)
                .unwrap())
        });
        serve_connection_with_runtime_state(
            io,
            svc,
            &config,
            Arc::new(RuntimeState::new_for_testing(config.max_file_streams)),
            &mut shutdown_rx,
            1,
            addr,
            addr,
            false,
            None,
        )
        .await;
    });

    let mut client = tokio::net::TcpStream::connect(addr).await.unwrap();
    client
        .write_all(b"GET /test HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n")
        .await
        .unwrap();
    let mut buf = Vec::new();
    client.read_to_end(&mut buf).await.unwrap();
    let _ = server.await;

    let response = String::from_utf8_lossy(&buf);
    assert!(
        response.starts_with("HTTP/1.1 504"),
        "expected 504, got: {}",
        response
    );
}

#[tokio::test]
async fn malformed_request_rejected_before_service() {
    let tmp = TempDir::new().unwrap();
    let _state = build_state(&tmp);
    let config = RuntimeConfig::default();
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let (tx, _rx) = broadcast::channel::<()>(1);

    let called = Arc::new(AtomicBool::new(false));
    let called_clone = called.clone();

    let server = tokio::spawn(async move {
        let (stream, _) = listener.accept().await.unwrap();
        let io = TokioIo::new(stream);
        let mut shutdown_rx = tx.subscribe();
        let svc = service_fn(move |_req: Request| {
            let called = called_clone.clone();
            async move {
                called.store(true, Ordering::SeqCst);
                Ok(Response::builder()
                    .status(StatusCode::OK)
                    .body(ResponseBody::Empty)
                    .unwrap())
            }
        });
        serve_connection_with_runtime_state(
            io,
            svc,
            &config,
            Arc::new(RuntimeState::new_for_testing(config.max_file_streams)),
            &mut shutdown_rx,
            1,
            addr,
            addr,
            false,
            None,
        )
        .await;
    });

    let mut client = tokio::net::TcpStream::connect(addr).await.unwrap();
    client
        .write_all(b"NOT A VALID HTTP REQUEST\r\n\r\n")
        .await
        .unwrap();
    let mut buf = Vec::new();
    let _ = client.read_to_end(&mut buf).await;
    let _ = server.await;

    assert!(
        !called.load(Ordering::SeqCst),
        "service should not be called for malformed requests"
    );
}

#[tokio::test]
async fn custom_service_bytes_through_pipeline() {
    let tmp = TempDir::new().unwrap();
    let _state = build_state(&tmp);
    let config = RuntimeConfig::default();
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let (tx, _rx) = broadcast::channel::<()>(1);

    let server = tokio::spawn(async move {
        let (stream, _) = listener.accept().await.unwrap();
        let io = TokioIo::new(stream);
        let mut shutdown_rx = tx.subscribe();
        let svc = service_fn(|_req: Request| async {
            Ok(Response::builder()
                .status(StatusCode::OK)
                .body(ResponseBody::Bytes(b"hello".to_vec()))
                .unwrap())
        });
        serve_connection_with_runtime_state(
            io,
            svc,
            &config,
            Arc::new(RuntimeState::new_for_testing(config.max_file_streams)),
            &mut shutdown_rx,
            1,
            addr,
            addr,
            false,
            None,
        )
        .await;
    });

    let mut client = tokio::net::TcpStream::connect(addr).await.unwrap();
    client
        .write_all(b"GET /test HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n")
        .await
        .unwrap();
    let mut buf = Vec::new();
    client.read_to_end(&mut buf).await.unwrap();
    let _ = server.await;

    let response = String::from_utf8_lossy(&buf);
    assert!(
        response.starts_with("HTTP/1.1 200"),
        "expected 200, got: {}",
        response
    );
    assert!(
        response.contains("hello"),
        "response body should contain 'hello': {}",
        response
    );
}

async fn request_custom_file(runtime_state: Arc<RuntimeState>, path: PathBuf) -> Vec<u8> {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let (tx, _rx) = broadcast::channel::<()>(1);
    let config = RuntimeConfig::default();

    let server = tokio::spawn(async move {
        let (stream, peer_addr) = listener.accept().await.unwrap();
        let io = TokioIo::new(stream);
        let mut shutdown_rx = tx.subscribe();
        let path = path.clone();
        let svc = service_fn(move |_req: Request| {
            let path = path.clone();
            async move {
                let file = File::open(&path).unwrap();
                let len = file.metadata().unwrap().len();
                let body = BodySource::FileFull {
                    file,
                    len,
                    mime: "application/octet-stream",
                };
                Ok(Response::builder()
                    .status(StatusCode::OK)
                    .body(ResponseBody::File(body))
                    .unwrap())
            }
        });
        serve_connection_with_runtime_state(
            io,
            svc,
            &config,
            runtime_state,
            &mut shutdown_rx,
            1,
            addr,
            peer_addr,
            false,
            None,
        )
        .await;
    });

    let mut client = tokio::net::TcpStream::connect(addr).await.unwrap();
    client
        .write_all(b"GET /file HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n")
        .await
        .unwrap();
    let mut response = Vec::new();
    client.read_to_end(&mut response).await.unwrap();
    server.await.unwrap();
    response
}

#[tokio::test]
async fn custom_service_file_stream_saturation_maps_503_and_recovers() {
    let tmp = TempDir::new().unwrap();
    let path = tmp.path().join("custom.bin");
    std::fs::write(&path, b"custom file").unwrap();
    let runtime_state = Arc::new(RuntimeState::new_for_testing(1));
    let held = runtime_state
        .file_stream_semaphore()
        .clone()
        .try_acquire_owned()
        .unwrap();

    let saturated = request_custom_file(runtime_state.clone(), path.clone()).await;
    assert!(saturated.starts_with(b"HTTP/1.1 503"), "{saturated:?}");

    drop(held);
    let recovered = request_custom_file(runtime_state, path).await;
    assert!(recovered.starts_with(b"HTTP/1.1 200"), "{recovered:?}");
    assert!(
        recovered
            .windows(b"custom file".len())
            .any(|window| window == b"custom file"),
        "{recovered:?}"
    );
}

#[tokio::test]
async fn connection_permits_released() {
    let tmp = TempDir::new().unwrap();
    let config = RuntimeConfig::builder()
        .max_connections(1)
        .handler_timeout(Duration::from_secs(10))
        .build()
        .unwrap();

    let server = Server::builder()
        .runtime(config)
        .serve_config(Arc::new(ServeConfig {
            root: tmp.path().to_path_buf(),
            ..ServeConfig::default()
        }))
        .build()
        .unwrap();

    let handle = server
        .start_with_service(service_fn(|_req: Request| async {
            tokio::time::sleep(Duration::from_millis(200)).await;
            Ok(Response::builder()
                .status(StatusCode::OK)
                .body(ResponseBody::Bytes(b"done".to_vec()))
                .unwrap())
        }))
        .await
        .unwrap();

    let addr = handle.local_addr();

    // First connection holds the single permit.
    let mut conn1 = tokio::net::TcpStream::connect(addr).await.unwrap();
    conn1
        .write_all(b"GET /test HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n")
        .await
        .unwrap();

    // Give the server time to accept the first connection and acquire the permit.
    tokio::time::sleep(Duration::from_millis(50)).await;

    // Second connection should be dropped by the accept loop (permit exhausted).
    let _result = tokio::time::timeout(Duration::from_millis(500), async {
        let mut conn = tokio::net::TcpStream::connect(addr).await.unwrap();
        let mut buf = Vec::new();
        conn.read_to_end(&mut buf).await.unwrap();
        buf
    })
    .await;

    // Read the first connection's response.
    let mut buf1 = Vec::new();
    let _ = conn1.read_to_end(&mut buf1).await;

    // Wait for the first connection to complete and release the permit.
    tokio::time::sleep(Duration::from_millis(100)).await;

    // Now a new connection should succeed.
    let mut conn2 = tokio::net::TcpStream::connect(addr).await.unwrap();
    conn2
        .write_all(b"GET /test HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n")
        .await
        .unwrap();
    let mut buf2 = Vec::new();
    conn2.read_to_end(&mut buf2).await.unwrap();
    let response2 = String::from_utf8_lossy(&buf2);
    assert!(
        response2.starts_with("HTTP/1.1 200"),
        "second connection after permit release should succeed: {}",
        response2
    );

    handle.shutdown();
}

#[tokio::test]
async fn hop_by_hop_headers_stripped() {
    let response = Response::builder()
        .status(StatusCode::OK)
        .header("connection", "close")
        .unwrap()
        .header("upgrade", "websocket")
        .unwrap()
        .header("x-custom", "preserved")
        .unwrap()
        .body(ResponseBody::Bytes(b"test".to_vec()))
        .unwrap();

    let req = NormalizeRequest::new(false);
    let normalized = normalize_response(response, &req).unwrap();

    let headers = normalized.headers();
    let header_names: Vec<&str> = headers.iter().map(|f| f.name.as_str()).collect();

    assert!(
        !header_names
            .iter()
            .any(|n| n.eq_ignore_ascii_case("connection")),
        "hop-by-hop header 'Connection' should be stripped"
    );
    assert!(
        !header_names
            .iter()
            .any(|n| n.eq_ignore_ascii_case("upgrade")),
        "hop-by-hop header 'Upgrade' should be stripped"
    );
    assert!(
        header_names
            .iter()
            .any(|n| n.eq_ignore_ascii_case("x-custom")),
        "non-hop-by-hop header 'X-Custom' should be preserved"
    );
}

/// Connection metadata: verify real socket addresses are propagated to the
/// custom service through the Request envelope.
#[tokio::test]
async fn connection_metadata_propagated_to_service() {
    use eggserve_core::primitives::connection_info::Scheme;
    use std::sync::atomic::{AtomicBool, Ordering};

    static METADATA_SEEN: AtomicBool = AtomicBool::new(false);

    let tmp = TempDir::new().unwrap();
    let _state = build_state(&tmp);
    let config = RuntimeConfig::default();
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let (tx, _rx) = broadcast::channel::<()>(1);

    let server = tokio::spawn(async move {
        let (stream, peer_addr) = listener.accept().await.unwrap();
        let io = TokioIo::new(stream);
        let mut shutdown_rx = tx.subscribe();
        let svc = service_fn(move |req: Request| {
            let peer_addr = peer_addr;
            async move {
                let conn = req.connection();
                // Verify real addresses are present (not placeholder 127.0.0.1:0).
                assert_ne!(
                    conn.remote_addr.port(),
                    0,
                    "remote port should be a real ephemeral port"
                );
                assert_eq!(conn.remote_addr, peer_addr);
                assert_eq!(conn.local_addr, addr);
                assert_eq!(conn.scheme, Scheme::Http);
                assert!(conn.tls.is_none());
                METADATA_SEEN.store(true, Ordering::SeqCst);
                Ok(Response::builder()
                    .status(StatusCode::OK)
                    .body(ResponseBody::Bytes(b"ok".to_vec()))
                    .unwrap())
            }
        });
        serve_connection_with_runtime_state(
            io,
            svc,
            &config,
            Arc::new(RuntimeState::new_for_testing(config.max_file_streams)),
            &mut shutdown_rx,
            1,
            addr,
            peer_addr,
            false,
            None,
        )
        .await;
    });

    let mut client = tokio::net::TcpStream::connect(addr).await.unwrap();
    client
        .write_all(b"GET /test HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n")
        .await
        .unwrap();
    let mut buf = Vec::new();
    let _ = client.read_to_end(&mut buf).await;
    let _ = server.await;

    assert!(
        METADATA_SEEN.load(Ordering::SeqCst),
        "connection metadata should be visible to the service"
    );
}

#[tokio::test]
async fn client_disconnect_releases_file_stream_permits() {
    let tmp = TempDir::new().unwrap();
    let path = tmp.path().join("large.bin");
    std::fs::write(&path, vec![b'x'; 16 * 1024 * 1024]).unwrap();
    let config = RuntimeConfig::builder()
        .bind("127.0.0.1:0".parse().unwrap())
        .max_file_streams(1)
        .build()
        .unwrap();
    let server = Server::builder().runtime(config).build().unwrap();
    let service = service_fn(move |_request: Request| {
        let path = path.clone();
        async move {
            let file = File::open(&path).unwrap();
            let len = file.metadata().unwrap().len();
            Ok(Response::builder()
                .status(StatusCode::OK)
                .body(ResponseBody::File(BodySource::FileFull {
                    file,
                    len,
                    mime: "application/octet-stream",
                }))
                .unwrap())
        }
    });
    let handle = server.start_with_service(service).await.unwrap();
    let addr = handle.local_addr();

    // Client A connects and starts streaming a large file response.
    let mut first = tokio::net::TcpStream::connect(addr).await.unwrap();
    first
        .write_all(b"GET /first HTTP/1.1\r\nHost: localhost\r\n\r\n")
        .await
        .unwrap();
    // Read only the response headers to confirm streaming started.
    let mut first_headers = Vec::new();
    loop {
        let mut byte = [0u8; 1];
        first.read_exact(&mut byte).await.unwrap();
        first_headers.push(byte[0]);
        if first_headers.ends_with(b"\r\n\r\n") {
            break;
        }
    }
    assert!(String::from_utf8_lossy(&first_headers).starts_with("HTTP/1.1 200"));

    // Client C should get503 while client A holds the only permit.
    let mut second = tokio::net::TcpStream::connect(addr).await.unwrap();
    second
        .write_all(b"GET /second HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n")
        .await
        .unwrap();
    let mut second_response = Vec::new();
    second.read_to_end(&mut second_response).await.unwrap();
    assert!(
        String::from_utf8_lossy(&second_response).starts_with("HTTP/1.1 503"),
        "expected shared file admission to reject second stream: {:?}",
        String::from_utf8_lossy(&second_response)
    );

    // Drop client A — the server's body stream and permit are released.
    drop(first);
    // Allow the server to process the disconnect and release the permit.
    tokio::time::sleep(Duration::from_millis(100)).await;

    // Client C retries and should succeed with200 now that the permit is free.
    let mut third = tokio::net::TcpStream::connect(addr).await.unwrap();
    third
        .write_all(b"GET /third HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n")
        .await
        .unwrap();
    let mut third_response = Vec::new();
    third.read_to_end(&mut third_response).await.unwrap();
    let third_str = String::from_utf8_lossy(&third_response);
    assert!(
        third_str.starts_with("HTTP/1.1 200"),
        "expected200 after disconnect release: {}",
        third_str
    );

    handle.shutdown();
}

#[tokio::test]
async fn runtime_file_admission_is_shared_across_connections() {
    let tmp = TempDir::new().unwrap();
    let path = tmp.path().join("shared.bin");
    std::fs::write(&path, vec![b'x'; 16 * 1024 * 1024]).unwrap();
    let config = RuntimeConfig::builder()
        .bind("127.0.0.1:0".parse().unwrap())
        .max_file_streams(1)
        .build()
        .unwrap();
    let server = Server::builder().runtime(config).build().unwrap();
    let service = service_fn(move |_request: Request| {
        let path = path.clone();
        async move {
            let file = File::open(&path).unwrap();
            let len = file.metadata().unwrap().len();
            Ok(Response::builder()
                .status(StatusCode::OK)
                .body(ResponseBody::File(BodySource::FileFull {
                    file,
                    len,
                    mime: "application/octet-stream",
                }))
                .unwrap())
        }
    });
    let handle = server.start_with_service(service).await.unwrap();
    let addr = handle.local_addr();

    let mut first = tokio::net::TcpStream::connect(addr).await.unwrap();
    first
        .write_all(b"GET /first HTTP/1.1\r\nHost: localhost\r\n\r\n")
        .await
        .unwrap();
    let mut first_headers = Vec::new();
    loop {
        let mut byte = [0u8; 1];
        first.read_exact(&mut byte).await.unwrap();
        first_headers.push(byte[0]);
        if first_headers.ends_with(b"\r\n\r\n") {
            break;
        }
    }
    assert!(String::from_utf8_lossy(&first_headers).starts_with("HTTP/1.1 200"));

    let mut second = tokio::net::TcpStream::connect(addr).await.unwrap();
    second
        .write_all(b"GET /second HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n")
        .await
        .unwrap();
    let mut second_response = Vec::new();
    second.read_to_end(&mut second_response).await.unwrap();
    assert!(
        String::from_utf8_lossy(&second_response).starts_with("HTTP/1.1 503"),
        "expected shared file admission to reject second stream: {:?}",
        String::from_utf8_lossy(&second_response)
    );

    drop(first);
    handle.shutdown();
}