use std::{
    io::{Read, Write},
    net::{TcpListener, TcpStream},
    process::{Command, Stdio},
    sync::mpsc,
    thread,
    time::Duration,
};

use serde_json::Value;
use tokio_tungstenite::tungstenite::{accept, Message as WsFrame};

use crate::support::TestWorkspace;

pub(crate) fn spawn_http_server(status: u16, reason: &str, content_type: &str, body: &str) -> String {
    let listener = TcpListener::bind("127.0.0.1:0").expect("listener should bind");
    let address = listener.local_addr().expect("address should be available");
    let reason = reason.to_string();
    let content_type = content_type.to_string();
    let body = body.to_string();

    thread::spawn(move || {
        let (mut stream, _) = listener.accept().expect("connection should be accepted");
        let mut buffer = [0_u8; 1024];
        let _ = stream.read(&mut buffer);
        let response = format!(
            "HTTP/1.1 {status} {reason}\r\nContent-Type: {content_type}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
            body.len()
        );
        stream
            .write_all(response.as_bytes())
            .expect("response should be written");
    });

    format!("http://{}", address)
}

pub(crate) fn spawn_mcp_http_server() -> String {
    let listener = TcpListener::bind("127.0.0.1:0").expect("listener should bind");
    let address = listener.local_addr().expect("address should be available");

    thread::spawn(move || {
        let (mut initialize_stream, _) = listener.accept().expect("initialize should connect");
        let initialize_request = read_http_request(&mut initialize_stream);
        let initialize_request_lower = initialize_request.to_ascii_lowercase();
        assert!(initialize_request_lower.contains("authorization: bearer test-token"));
        assert!(initialize_request.contains("\"method\":\"initialize\""));
        assert!(initialize_request.contains("\"protocolVersion\":\"2025-11-25\""));
        assert!(initialize_request.contains("\"name\":\"hen\""));
        assert!(initialize_request.contains(env!("CARGO_PKG_VERSION")));

        write_http_response(
            &mut initialize_stream,
            200,
            "OK",
            Some("application/json"),
            r#"{"jsonrpc":"2.0","id":1,"result":{"protocolVersion":"2025-11-25","capabilities":{},"serverInfo":{"name":"fixture-mcp","version":"1.0.0"}}}"#,
        );

        let (mut initialized_stream, _) = listener.accept().expect("initialized notification should connect");
        let initialized_request = read_http_request(&mut initialized_stream);
        let initialized_request_lower = initialized_request.to_ascii_lowercase();
        assert!(initialized_request_lower.contains("authorization: bearer test-token"));
        assert!(initialized_request.contains("\"method\":\"notifications/initialized\""));

        write_http_response(&mut initialized_stream, 202, "Accepted", None, "");

        let (mut list_tools_stream, _) = listener.accept().expect("tools/list should connect");
        let list_tools_request = read_http_request(&mut list_tools_stream);
        let list_tools_request_lower = list_tools_request.to_ascii_lowercase();
        assert!(list_tools_request_lower.contains("authorization: bearer test-token"));
        assert!(list_tools_request.contains("\"method\":\"tools/list\""));

        write_http_response(
            &mut list_tools_stream,
            200,
            "OK",
            Some("application/json"),
            r#"{"jsonrpc":"2.0","id":2,"result":{"tools":[{"name":"search"}]}}"#,
        );

        let (mut list_resources_stream, _) = listener.accept().expect("resources/list should connect");
        let list_resources_request = read_http_request(&mut list_resources_stream);
        let list_resources_request_lower = list_resources_request.to_ascii_lowercase();
        assert!(list_resources_request_lower.contains("authorization: bearer test-token"));
        assert!(list_resources_request.contains("\"method\":\"resources/list\""));

        write_http_response(
            &mut list_resources_stream,
            200,
            "OK",
            Some("application/json"),
            r#"{"jsonrpc":"2.0","id":3,"result":{"resources":[{"uri":"file:///fixture.txt","name":"Fixture Resource"}]}}"#,
        );

        let (mut tool_call_stream, _) = listener.accept().expect("tools/call should connect");
        let tool_call_request = read_http_request(&mut tool_call_stream);
        let tool_call_request_lower = tool_call_request.to_ascii_lowercase();
        assert!(tool_call_request_lower.contains("authorization: bearer test-token"));
        assert!(tool_call_request.contains("\"method\":\"tools/call\""));
        assert!(tool_call_request.contains("\"name\":\"search\""));
        assert!(tool_call_request.contains("\"query\":\"hedgehog\""));

        write_http_response(
            &mut tool_call_stream,
            200,
            "OK",
            Some("application/json"),
            r#"{"jsonrpc":"2.0","id":4,"result":{"content":[{"type":"text","text":"search result"}]}}"#,
        );
    });

    format!("http://{}", address)
}

pub(crate) fn spawn_mcp_protocol_error_server() -> String {
    let listener = TcpListener::bind("127.0.0.1:0").expect("listener should bind");
    let address = listener.local_addr().expect("address should be available");

    thread::spawn(move || {
        let (mut initialize_stream, _) = listener.accept().expect("initialize should connect");
        let initialize_request = read_http_request(&mut initialize_stream);
        let initialize_request_lower = initialize_request.to_ascii_lowercase();
        assert!(initialize_request_lower.contains("authorization: bearer test-token"));
        assert!(initialize_request.contains("\"method\":\"initialize\""));

        write_http_response(
            &mut initialize_stream,
            200,
            "OK",
            Some("application/json"),
            r#"{"jsonrpc":"2.0","id":1,"result":{"protocolVersion":"2025-11-25","capabilities":{},"serverInfo":{"name":"fixture-mcp","version":"1.0.0"}}}"#,
        );

        let (mut initialized_stream, _) = listener.accept().expect("initialized notification should connect");
        let initialized_request = read_http_request(&mut initialized_stream);
        let initialized_request_lower = initialized_request.to_ascii_lowercase();
        assert!(initialized_request_lower.contains("authorization: bearer test-token"));
        assert!(initialized_request.contains("\"method\":\"notifications/initialized\""));

        write_http_response(&mut initialized_stream, 202, "Accepted", None, "");

        let (mut tool_call_stream, _) = listener.accept().expect("tools/call should connect");
        let tool_call_request = read_http_request(&mut tool_call_stream);
        let tool_call_request_lower = tool_call_request.to_ascii_lowercase();
        assert!(tool_call_request_lower.contains("authorization: bearer test-token"));
        assert!(tool_call_request.contains("\"method\":\"tools/call\""));
        assert!(tool_call_request.contains("\"name\":\"missing-tool\""));

        write_http_response(
            &mut tool_call_stream,
            200,
            "OK",
            Some("application/json"),
            r#"{"jsonrpc":"2.0","id":2,"error":{"code":-32601,"message":"Tool not found"}}"#,
        );
    });

    format!("http://{}", address)
}

pub(crate) fn spawn_mcp_sse_http_server() -> String {
    let listener = TcpListener::bind("127.0.0.1:0").expect("listener should bind");
    let address = listener.local_addr().expect("address should be available");

    thread::spawn(move || {
        let (mut initialize_stream, _) = listener.accept().expect("initialize should connect");
        let initialize_request = read_http_request(&mut initialize_stream);
        assert!(initialize_request.contains("\"method\":\"initialize\""));

        write_http_response(
            &mut initialize_stream,
            200,
            "OK",
            Some("text/event-stream"),
            concat!(
                "event: message\r\n",
                "data:\r\n\r\n",
                "event: message\r\n",
                "data: {\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{\"protocolVersion\":\"2025-06-18\",\"capabilities\":{\"tools\":{\"listChanged\":true}},\"serverInfo\":{\"name\":\"fixture-mcp\",\"version\":\"1.0.0\"}}}\r\n\r\n"
            ),
        );

        let (mut initialized_stream, _) = listener.accept().expect("initialized notification should connect");
        let initialized_request = read_http_request(&mut initialized_stream);
        assert!(initialized_request.contains("\"method\":\"notifications/initialized\""));

        write_http_response(&mut initialized_stream, 202, "Accepted", None, "");
    });

    format!("http://{}", address)
}

pub(crate) fn spawn_sse_http_server() -> String {
    let listener = TcpListener::bind("127.0.0.1:0").expect("listener should bind");
    let address = listener.local_addr().expect("address should be available");

    thread::spawn(move || {
        let (mut stream, _) = listener.accept().expect("stream should connect");
        let request = read_http_request(&mut stream);
        let request_lower = request.to_ascii_lowercase();
        assert!(request.starts_with("GET / HTTP/1.1") || request.starts_with("GET / "));
        assert!(request_lower.contains("accept: text/event-stream"));

        write_http_response(
            &mut stream,
            200,
            "OK",
            Some("text/event-stream"),
            concat!(
                "event: price\r\n",
                "id: evt-1\r\n",
                "data: {\"symbol\":\"AAPL\",\"price\":182.4}\r\n\r\n"
            ),
        );
    });

    format!("http://{}", address)
}

pub(crate) fn spawn_ws_server() -> String {
    let listener = TcpListener::bind("127.0.0.1:0").expect("listener should bind");
    let address = listener.local_addr().expect("address should be available");

    thread::spawn(move || {
        let (stream, _) = listener.accept().expect("socket should connect");
        let mut socket = accept(stream).expect("handshake should succeed");

        let message = socket.read().expect("message should be readable");

        let text = match message {
            WsFrame::Text(text) => text.to_string(),
            other => panic!("expected text message, got {:?}", other),
        };

        let payload: Value = serde_json::from_str(&text).expect("payload should be valid json");
        assert_eq!(payload["type"], "hello");
        assert_eq!(payload["room"], "prices");

        socket
            .send(WsFrame::Text(
                r#"{"type":"ack","room":"prices"}"#.to_string().into(),
            ))
            .expect("ack should be sent");
    });

    format!("ws://{}", address)
}

fn read_http_request(stream: &mut TcpStream) -> String {
    let mut buffer = Vec::new();
    let mut chunk = [0_u8; 1024];
    let mut headers_end = None;
    let mut content_length = 0usize;

    loop {
        let bytes_read = stream.read(&mut chunk).expect("request should be readable");
        if bytes_read == 0 {
            break;
        }

        buffer.extend_from_slice(&chunk[..bytes_read]);

        if headers_end.is_none() {
            if let Some(end) = find_header_terminator(&buffer) {
                headers_end = Some(end);
                let headers = String::from_utf8_lossy(&buffer[..end]);
                content_length = parse_content_length(headers.as_ref());
            }
        }

        if let Some(end) = headers_end {
            if buffer.len() >= end + content_length {
                break;
            }
        }
    }

    String::from_utf8(buffer).expect("request should be utf-8")
}

fn find_header_terminator(buffer: &[u8]) -> Option<usize> {
    buffer
        .windows(4)
        .position(|window| window == b"\r\n\r\n")
        .map(|position| position + 4)
}

fn parse_content_length(headers: &str) -> usize {
    headers
        .lines()
        .find_map(|line| {
            let (name, value) = line.split_once(':')?;
            if name.eq_ignore_ascii_case("content-length") {
                value.trim().parse::<usize>().ok()
            } else {
                None
            }
        })
        .unwrap_or(0)
}

fn write_http_response(
    stream: &mut TcpStream,
    status: u16,
    reason: &str,
    content_type: Option<&str>,
    body: &str,
) {
    let content_type_header = content_type
        .map(|value| format!("Content-Type: {value}\r\n"))
        .unwrap_or_default();
    let response = format!(
        "HTTP/1.1 {status} {reason}\r\n{content_type_header}Content-Length: {}\r\nConnection: close\r\n\r\n{body}",
        body.len()
    );
    stream
        .write_all(response.as_bytes())
        .expect("response should be written");
}

#[cfg(unix)]
pub(crate) fn spawn_blocking_http_server(started_tx: mpsc::Sender<()>) -> String {
    let listener = TcpListener::bind("127.0.0.1:0").expect("listener should bind");
    let address = listener.local_addr().expect("address should be available");

    thread::spawn(move || {
        let (mut stream, _) = listener.accept().expect("connection should be accepted");
        let mut buffer = [0_u8; 1024];
        let _ = stream.read(&mut buffer);
        started_tx
            .send(())
            .expect("test should observe the blocking request");
        thread::sleep(Duration::from_secs(5));

        let body = r#"{"ok":true}"#;
        let response = format!(
            "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
            body.len()
        );
        let _ = stream.write_all(response.as_bytes());
    });

    format!("http://{}", address)
}

#[cfg(unix)]
pub(crate) fn run_until_signal(
    workspace: &TestWorkspace,
    args: &[&str],
    signal: &str,
    started_rx: mpsc::Receiver<()>,
) -> std::process::Output {
    let child = Command::new(env!("CARGO_BIN_EXE_hen"))
        .args(args)
        .current_dir(workspace.root())
        .env("NO_COLOR", "1")
        .env("TERM", "dumb")
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()
        .expect("hen command should execute");

    if started_rx.recv_timeout(Duration::from_secs(5)).is_err() {
        let output = child
            .wait_with_output()
            .expect("child output should be available");
        let stdout = String::from_utf8_lossy(&output.stdout);
        let stderr = String::from_utf8_lossy(&output.stderr);
        panic!(
            "second request should start before signalling\nstatus: {:?}\nstdout: {}\nstderr: {}",
            output.status.code(),
            stdout,
            stderr,
        );
    }

    thread::sleep(Duration::from_millis(200));

    let status = Command::new("kill")
        .args([signal, &child.id().to_string()])
        .status()
        .expect("signal command should execute");
    assert!(status.success(), "failed to send signal {signal}");

    child
        .wait_with_output()
        .expect("child output should be available")
}