use std::collections::BTreeMap;
use std::io::{Read, Write};
use std::net::{SocketAddr, TcpStream};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::thread;
use std::time::{Duration, Instant};

use tempfile::tempdir;
use wip_fs_server::{AccessMode, ServerConfig, TransportSecurity, WipFsServer};
use wip_https::{
    BatchRequest, BatchRequestItem, BatchRequestKind, BatchResponse, BatchSuccess, DecodedResponse,
    Endpoint, JSON_CONTENT_TYPE, Limits, Route, decode_batch_request, decode_batch_response,
    decode_call_operation_response, decode_fetch_interface_response, decode_fetch_response,
    decode_fetch_tree_response, encode_batch_request, encode_call_operation_request,
    encode_fetch_interface_request, encode_fetch_request, encode_fetch_tree_request,
};
use wip_protocol::{
    CallOperationRequest, FetchInterfaceRequest, FetchRequest, FetchTreeRequest, InterfaceTarget,
    ProtocolErrorCode, Target, Value,
};

#[test]
fn real_listener_serves_four_core_interactions_batch_and_write() {
    let temporary = tempdir().expect("temporary root");
    std::fs::write(temporary.path().join("hello.txt"), b"hello world").expect("write file");
    std::fs::create_dir(temporary.path().join("nested")).expect("create directory");

    let mut config = ServerConfig::for_root(temporary.path());
    config.access = AccessMode::ReadWrite;
    config.transport_security = TransportSecurity::DevelopmentHttp;
    let limits = config.limits.wire;
    let bound = Arc::new(
        WipFsServer::new(config)
            .expect("build server")
            .bind()
            .expect("bind"),
    );
    let address = bound.local_addr();
    let shutdown = Arc::new(AtomicBool::new(false));
    let worker = {
        let bound = Arc::clone(&bound);
        let shutdown = Arc::clone(&shutdown);
        thread::spawn(move || bound.serve_until(shutdown))
    };
    let endpoint = Endpoint::parse(&format!("https://localhost:{}/wip", address.port()))
        .expect("client endpoint");

    let file_request = FetchRequest {
        path: "/hello.txt".into(),
    };
    let response = exchange(
        address,
        encode_fetch_request(&endpoint, &file_request, limits).expect("encode fetch"),
    );
    let file =
        success(decode_fetch_response(&file_request, &response, limits).expect("decode fetch"));
    assert_eq!(file.name, "hello.txt");
    assert_eq!(file.interfaces.len(), 2);
    assert!(file.validator.is_some());

    let tree_request = FetchTreeRequest {
        path: "/".into(),
        depth: 1,
        limit: 10,
    };
    let response = exchange(
        address,
        encode_fetch_tree_request(&endpoint, &tree_request, limits).expect("encode tree"),
    );
    let tree =
        success(decode_fetch_tree_response(&tree_request, &response, limits).expect("decode tree"));
    assert!(!tree.truncated);
    assert_eq!(tree.root.children.len(), 2);

    let read_interface_request = FetchInterfaceRequest {
        interface: file.interfaces[0].clone(),
    };
    let response = exchange(
        address,
        encode_fetch_interface_request(&endpoint, &read_interface_request, limits)
            .expect("encode interface fetch"),
    );
    let read_interface = success(
        decode_fetch_interface_response(&read_interface_request, &response, limits)
            .expect("decode interface"),
    );
    assert_eq!(read_interface.descriptor.operations[0].name, "read");

    let read_request = CallOperationRequest {
        target: Target {
            path: "/hello.txt".into(),
            validator: file.validator.clone(),
        },
        interface: InterfaceTarget {
            reference: file.interfaces[0].clone(),
            validator: None,
        },
        operation: "read".into(),
        arguments: BTreeMap::from([
            ("offset".into(), Value::Integer(6)),
            ("length".into(), Value::Integer(5)),
        ]),
    };
    let response = exchange(
        address,
        encode_call_operation_request(&endpoint, &read_request, &read_interface.descriptor, limits)
            .expect("encode read call"),
    );
    let read = success(
        decode_call_operation_response(
            &read_request,
            &read_interface.descriptor,
            &response,
            limits,
        )
        .expect("decode read call"),
    );
    assert_eq!(read.result, Value::Bytes(b"world".to_vec()));

    let batch_request = BatchRequest {
        requests: vec![
            BatchRequestItem {
                id: "root".into(),
                request: BatchRequestKind::Fetch(FetchRequest { path: "/".into() }),
            },
            BatchRequestItem {
                id: "tree".into(),
                request: BatchRequestKind::FetchTree(FetchTreeRequest {
                    path: "/nested".into(),
                    depth: 1,
                    limit: 4,
                }),
            },
            BatchRequestItem {
                id: "interface".into(),
                request: BatchRequestKind::FetchInterface(read_interface_request.clone()),
            },
        ],
    };
    let response = exchange(
        address,
        encode_batch_request(&endpoint, &batch_request, limits).expect("encode batch"),
    );
    let batch =
        success(decode_batch_response(&batch_request, &response, limits).expect("decode batch"));
    assert_batch_successes(&batch);

    let write_interface_request = FetchInterfaceRequest {
        interface: file.interfaces[1].clone(),
    };
    let response = exchange(
        address,
        encode_fetch_interface_request(&endpoint, &write_interface_request, limits)
            .expect("encode write interface fetch"),
    );
    let write_interface = success(
        decode_fetch_interface_response(&write_interface_request, &response, limits)
            .expect("decode write interface"),
    );
    let write_request = CallOperationRequest {
        target: Target {
            path: "/hello.txt".into(),
            validator: file.validator,
        },
        interface: InterfaceTarget {
            reference: file.interfaces[1].clone(),
            validator: None,
        },
        operation: "replace".into(),
        arguments: BTreeMap::from([("contents".into(), Value::Bytes(b"updated".to_vec()))]),
    };
    let response = exchange(
        address,
        encode_call_operation_request(
            &endpoint,
            &write_request,
            &write_interface.descriptor,
            limits,
        )
        .expect("encode write call"),
    );
    let write = success(
        decode_call_operation_response(
            &write_request,
            &write_interface.descriptor,
            &response,
            limits,
        )
        .expect("decode write call"),
    );
    assert!(write.validator.is_some());
    assert_eq!(
        std::fs::read(temporary.path().join("hello.txt")).expect("read updated file"),
        b"updated"
    );

    shutdown.store(true, Ordering::Release);
    worker
        .join()
        .expect("join listener")
        .expect("listener exits cleanly");
}

#[test]
fn stalled_request_body_times_out_and_releases_the_only_worker() {
    let temporary = tempdir().expect("temporary root");
    let mut config = ServerConfig::for_root(temporary.path());
    config.transport_security = TransportSecurity::DevelopmentHttp;
    config.limits.max_concurrent_requests = 1;
    config.limits.request_timeout = Duration::from_millis(200);
    let limits = config.limits.wire;
    let bound = Arc::new(
        WipFsServer::new(config)
            .expect("build server")
            .bind()
            .expect("bind"),
    );
    let address = bound.local_addr();
    let shutdown = Arc::new(AtomicBool::new(false));
    let worker = {
        let bound = Arc::clone(&bound);
        let shutdown = Arc::clone(&shutdown);
        thread::spawn(move || bound.serve_until(shutdown))
    };

    let mut stalled = TcpStream::connect(address).expect("connect stalled client");
    stalled
        .set_read_timeout(Some(Duration::from_secs(1)))
        .expect("set stalled read timeout");
    write!(
        stalled,
        "POST /wip/v1/fetch HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\nAccept: application/json\r\nContent-Length: 100\r\n\r\n{{"
    )
    .expect("write partial request");
    stalled.flush().expect("flush partial request");
    thread::sleep(Duration::from_millis(25));

    let endpoint = Endpoint::parse(&format!("https://localhost:{}/wip", address.port()))
        .expect("client endpoint");
    let fetch = FetchRequest { path: "/".into() };
    let started = Instant::now();
    let response = exchange(
        address,
        encode_fetch_request(&endpoint, &fetch, limits).expect("encode fetch"),
    );
    let _root = success(decode_fetch_response(&fetch, &response, limits).expect("decode fetch"));
    assert!(
        started.elapsed() < Duration::from_secs(1),
        "the valid request remained starved after the intake deadline"
    );

    let mut byte = [0_u8; 1];
    match stalled.read(&mut byte) {
        Ok(0) => {}
        Err(error)
            if matches!(
                error.kind(),
                std::io::ErrorKind::ConnectionReset
                    | std::io::ErrorKind::ConnectionAborted
                    | std::io::ErrorKind::BrokenPipe
            ) => {}
        outcome => panic!("stalled connection was not closed after its deadline: {outcome:?}"),
    }

    shutdown.store(true, Ordering::Release);
    worker
        .join()
        .expect("join listener")
        .expect("listener exits cleanly");
}

#[test]
fn batch_keeps_known_malformed_arguments_item_local() {
    let temporary = tempdir().expect("temporary root");
    let config = ServerConfig::for_root(temporary.path());
    let limits = config.limits.wire;
    let server = WipFsServer::new(config).expect("build server");
    let endpoint = Endpoint::parse("https://localhost/wip").expect("endpoint");
    let body = br#"{"requests":[{"id":"ok","method":"fetch","arguments":{"path":"/"}},{"id":"bad","method":"fetch_tree","arguments":{"path":"/","depth":"bad","limit":1}}]}"#;
    let logical = decode_batch_request(body, limits).expect("decode item-local invalid request");
    assert!(matches!(
        logical.requests[1].request,
        BatchRequestKind::Invalid { .. }
    ));
    let request = http::Request::builder()
        .method(http::Method::POST)
        .uri(endpoint.route(Route::Batch).as_str())
        .header(http::header::CONTENT_TYPE, JSON_CONTENT_TYPE)
        .header(http::header::ACCEPT, "application/json")
        .body(body.to_vec())
        .expect("build request");
    let response = server.handle_http(request);
    let batch = success(
        decode_batch_response(&logical, &response, limits).expect("decode mixed batch response"),
    );
    assert!(matches!(
        batch.responses[0].outcome,
        Ok(BatchSuccess::Fetch(_))
    ));
    assert!(batch.responses[1].outcome.is_err());
}

#[test]
fn call_argument_decoding_does_not_mask_missing_target() {
    let temporary = tempdir().expect("temporary root");
    let config = ServerConfig::for_root(temporary.path());
    let server = WipFsServer::new(config).expect("build server");
    let endpoint = Endpoint::parse("https://localhost/wip").expect("endpoint");
    let body = br#"{"target":{"path":"/missing"},"interface":{"reference":"unknown"},"operation":"missing","arguments":{"bytes":"not-base64"}}"#;
    let request = http::Request::builder()
        .method(http::Method::POST)
        .uri(endpoint.route(Route::CallOperation).as_str())
        .header(http::header::CONTENT_TYPE, JSON_CONTENT_TYPE)
        .header(http::header::ACCEPT, "application/json")
        .body(body.to_vec())
        .expect("build request");
    let response = server.handle_http(request);
    assert_eq!(response.status(), http::StatusCode::NOT_FOUND);
    assert!(
        std::str::from_utf8(response.body())
            .expect("UTF-8 error")
            .contains("\"not_found\"")
    );
}

#[test]
fn backend_invalid_unicode_path_is_not_found_not_internal() {
    let temporary = tempdir().expect("temporary root");
    let config = ServerConfig::for_root(temporary.path());
    let limits = config.limits.wire;
    let server = WipFsServer::new(config).expect("build server");
    let endpoint = Endpoint::parse("https://localhost/wip").expect("endpoint");
    let request = FetchRequest {
        path: "/bad\0name".into(),
    };
    let response = server
        .handle_http(encode_fetch_request(&endpoint, &request, limits).expect("encode fetch"));
    assert_eq!(response.status(), http::StatusCode::NOT_FOUND);
    match decode_fetch_response(&request, &response, limits).expect("decode missing response") {
        DecodedResponse::ProtocolFailure { error, .. } => {
            assert_eq!(error.code, ProtocolErrorCode::NotFound);
        }
        DecodedResponse::Success(_) => panic!("backend-invalid path must not resolve"),
    }
}

#[test]
fn read_response_size_failure_preserves_non_commit_guarantee() {
    let temporary = tempdir().expect("temporary root");
    std::fs::write(temporary.path().join("large"), vec![b'x'; 2_000]).expect("write file");
    let mut config = ServerConfig::for_root(temporary.path());
    config.limits.wire = Limits::new(4_096, 1_024, 64, 64).expect("wire limits");
    let limits = config.limits.wire;
    let server = WipFsServer::new(config).expect("build server");
    let endpoint = Endpoint::parse("https://localhost/wip").expect("endpoint");

    let fetch_request = FetchRequest {
        path: "/large".into(),
    };
    let response = server.handle_http(
        encode_fetch_request(&endpoint, &fetch_request, limits).expect("encode fetch"),
    );
    let file =
        success(decode_fetch_response(&fetch_request, &response, limits).expect("decode fetch"));
    let interface_request = FetchInterfaceRequest {
        interface: file.interfaces[0].clone(),
    };
    let response = server.handle_http(
        encode_fetch_interface_request(&endpoint, &interface_request, limits)
            .expect("encode interface fetch"),
    );
    let interface = success(
        decode_fetch_interface_response(&interface_request, &response, limits)
            .expect("decode interface"),
    );
    let call = CallOperationRequest {
        target: Target {
            path: "/large".into(),
            validator: file.validator,
        },
        interface: InterfaceTarget {
            reference: file.interfaces[0].clone(),
            validator: None,
        },
        operation: "read".into(),
        arguments: BTreeMap::from([("length".into(), Value::Integer(1_500))]),
    };
    let response = server.handle_http(
        encode_call_operation_request(&endpoint, &call, &interface.descriptor, limits)
            .expect("encode read call"),
    );
    assert_eq!(response.status(), http::StatusCode::PAYLOAD_TOO_LARGE);
    match decode_call_operation_response(&call, &interface.descriptor, &response, limits)
        .expect("decode resource failure")
    {
        DecodedResponse::ProtocolFailure { error, .. } => {
            assert_eq!(error.code, ProtocolErrorCode::ResourceLimitExceeded);
        }
        DecodedResponse::Success(_) => panic!("oversized read response must fail"),
    }
}

fn assert_batch_successes(batch: &BatchResponse) {
    assert_eq!(batch.responses.len(), 3);
    assert!(matches!(
        batch.responses[0].outcome,
        Ok(BatchSuccess::Fetch(_))
    ));
    assert!(matches!(
        batch.responses[1].outcome,
        Ok(BatchSuccess::FetchTree(_))
    ));
    assert!(matches!(
        batch.responses[2].outcome,
        Ok(BatchSuccess::FetchInterface(_))
    ));
}

fn success<T>(response: DecodedResponse<T>) -> T {
    match response {
        DecodedResponse::Success(value) => value,
        DecodedResponse::ProtocolFailure { error, .. } => panic!("protocol failure: {error}"),
    }
}

fn exchange(address: SocketAddr, request: http::Request<Vec<u8>>) -> http::Response<Vec<u8>> {
    let mut stream = TcpStream::connect(address).expect("connect listener");
    stream
        .set_read_timeout(Some(Duration::from_secs(5)))
        .expect("set read timeout");
    let path = request
        .uri()
        .path_and_query()
        .map_or("/", |value| value.as_str());
    write!(
        stream,
        "{} {} HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n",
        request.method(),
        path
    )
    .expect("write request line");
    for (name, value) in request.headers() {
        write!(
            stream,
            "{}: {}\r\n",
            name,
            value.to_str().expect("ASCII request header")
        )
        .expect("write header");
    }
    write!(stream, "Content-Length: {}\r\n\r\n", request.body().len())
        .expect("write content length");
    stream.write_all(request.body()).expect("write body");
    stream.flush().expect("flush request");

    let mut raw = Vec::new();
    stream.read_to_end(&mut raw).expect("read response");
    let split = raw
        .windows(4)
        .position(|window| window == b"\r\n\r\n")
        .expect("HTTP response header");
    let head = std::str::from_utf8(&raw[..split]).expect("UTF-8 response head");
    let mut lines = head.split("\r\n");
    let status = lines
        .next()
        .and_then(|line| line.split_whitespace().nth(1))
        .and_then(|status| status.parse::<u16>().ok())
        .expect("response status");
    let mut builder = http::Response::builder().status(status);
    for line in lines {
        let (name, value) = line.split_once(':').expect("response header");
        builder = builder.header(name, value.trim());
    }
    builder
        .body(raw[split + 4..].to_vec())
        .expect("build response")
}