use std::sync::{
    Arc,
    atomic::{AtomicBool, Ordering},
};

use avril_http::{Address, Engine};
use bytes::Bytes;
use tokio::io::{AsyncReadExt, AsyncWriteExt};

#[tokio::test]
async fn engine_streams_without_a_js_runtime() {
    let engine = Arc::new(Engine::new(tokio::runtime::Handle::current()));
    let (id, address) = engine
        .listen(
            Address::Tcp {
                hostname: "127.0.0.1".into(),
                port: 0,
            },
            Arc::new(|_| Ok(())),
        )
        .unwrap();
    let Address::Tcp { hostname, port } = address else {
        panic!("expected TCP");
    };
    let mut client = tokio::net::TcpStream::connect((hostname.as_str(), port))
        .await
        .unwrap();
    client.write_all(b"POST /stream HTTP/1.1\r\nHost: example.test\r\nContent-Length: 5\r\nConnection: close\r\n\r\nhello").await.unwrap();
    let metadata = engine.next(id).await.unwrap().unwrap();
    assert_eq!(metadata.1, "POST");
    assert_eq!(metadata.2, "http://example.test/stream");
    let request = metadata.0;
    let completed = engine.wait(request);
    let mut body = Vec::new();
    while let Some(bytes) = engine.read(request).await.unwrap() {
        body.extend(bytes);
    }
    assert_eq!(body, b"hello");
    assert!(engine.respond((request, 200, vec![], None, true)).unwrap());
    assert!(
        engine
            .write(request, Some(Bytes::from_static(b"world")))
            .await
            .unwrap()
    );
    engine.write(request, None).await.unwrap();
    let mut response = String::new();
    client.read_to_string(&mut response).await.unwrap();
    assert!(response.starts_with("HTTP/1.1 200"));
    assert!(response.contains("world"));
    assert!(!completed.await.unwrap());
    engine.stop(id, false);
    engine.finished(id).await.unwrap();
}

#[tokio::test]
async fn native_file_ranges_preserve_keep_alive_framing() {
    use avril_http::file::{header_pairs, headers, serve_file};

    let directory = tempfile::tempdir().unwrap();
    let path = directory.path().join("data.bin");
    let bytes: Vec<u8> = (0..200_000).map(|i| (i % 251) as u8).collect();
    std::fs::write(&path, &bytes).unwrap();
    let engine = Arc::new(Engine::new(tokio::runtime::Handle::current()));
    let (id, address) = engine
        .listen(
            Address::Tcp {
                hostname: "127.0.0.1".into(),
                port: 0,
            },
            Arc::new(|_| Ok(())),
        )
        .unwrap();
    let Address::Tcp { hostname, port } = address else {
        panic!("expected TCP");
    };
    let mut client = tokio::net::TcpStream::connect((hostname.as_str(), port))
        .await
        .unwrap();
    client.write_all(b"GET /file HTTP/1.1\r\nHost: localhost\r\nRange: bytes=65530-131080\r\n\r\nGET /next HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n").await.unwrap();
    let receive = tokio::spawn(async move {
        let mut bytes = Vec::new();
        client.read_to_end(&mut bytes).await.unwrap();
        bytes
    });
    let request = engine.next(id).await.unwrap().unwrap();
    let complete = engine.wait(request.0);
    let runtime = tokio::runtime::Handle::current();
    let response = tokio::task::spawn_blocking(move || {
        serve_file(
            runtime,
            &request.1,
            headers(request.3).unwrap(),
            "data.bin",
            headers(vec![]).unwrap(),
            || Ok((std::fs::File::open(&path)?, Arc::new(|| Ok(())))),
        )
    })
    .await
    .unwrap()
    .unwrap();
    engine
        .respond_file(
            request.0,
            response.status().as_u16(),
            header_pairs(response.headers()),
            response.into_body().unwrap(),
        )
        .unwrap();
    let next = engine.next(id).await.unwrap().unwrap().0;
    let next_complete = engine.wait(next);
    engine
        .respond((next, 200, vec![], Some(Bytes::from_static(b"next")), false))
        .unwrap();
    let received = receive.await.unwrap();
    let body_start = received
        .windows(4)
        .position(|part| part == b"\r\n\r\n")
        .unwrap()
        + 4;
    assert!(received.starts_with(b"HTTP/1.1 206"));
    assert_eq!(
        &received[body_start..body_start + 65551],
        &bytes[65530..=131080]
    );
    assert!(received[body_start + 65551..].starts_with(b"HTTP/1.1 200"));
    assert!(received.ends_with(b"next"));
    assert!(!complete.await.unwrap());
    assert!(!next_complete.await.unwrap());
    engine.stop(id, false);
    engine.finished(id).await.unwrap();
}

#[tokio::test]
async fn authorization_denials_do_not_leak_pending_responses() {
    let engine = Arc::new(Engine::new(tokio::runtime::Handle::current()));
    let allowed = Arc::new(AtomicBool::new(true));
    let flag = allowed.clone();
    let authorize = Arc::new(move |_: &Address| {
        if flag.load(Ordering::Relaxed) {
            Ok(())
        } else {
            Err("denied".into())
        }
    });
    let (id, address) = engine
        .listen(
            Address::Tcp {
                hostname: "127.0.0.1".into(),
                port: 0,
            },
            authorize.clone(),
        )
        .unwrap();
    let Address::Tcp { hostname, port } = address else {
        panic!("expected TCP");
    };
    let mut client = tokio::net::TcpStream::connect((hostname.as_str(), port))
        .await
        .unwrap();
    client
        .write_all(b"GET / HTTP/1.1\r\nHost: example.test\r\nConnection: close\r\n\r\n")
        .await
        .unwrap();
    let request = engine.next(id).await.unwrap().unwrap().0;
    let completed = engine.wait(request);
    allowed.store(false, Ordering::Relaxed);
    assert_eq!(
        engine
            .respond((
                request,
                200,
                vec![],
                Some(Bytes::from_static(b"secret")),
                false
            ))
            .unwrap_err()
            .to_string(),
        "denied"
    );
    assert!(
        engine
            .listen(
                Address::Tcp {
                    hostname: "127.0.0.1".into(),
                    port: 0
                },
                authorize
            )
            .is_err()
    );
    engine.cancel(request);
    let mut response = Vec::new();
    client.read_to_end(&mut response).await.unwrap();
    assert!(response.is_empty());
    assert!(completed.await.unwrap());
    engine.stop(id, false);
    engine.finished(id).await.unwrap();
}

#[test]
#[cfg(unix)]
fn failed_listener_setup_leaves_no_socket_entry() {
    let directory = tempfile::tempdir().unwrap();
    let path = directory.path().join("failed.sock");
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    let handle = runtime.handle().clone();
    drop(runtime);
    let engine = Arc::new(Engine::new(handle));
    assert!(
        engine
            .listen(
                Address::Unix {
                    path: Some(path.to_str().unwrap().into())
                },
                Arc::new(|_| Ok(()))
            )
            .is_err()
    );
    assert!(!path.exists());
}

#[tokio::test]
#[cfg(target_os = "linux")]
async fn unix_cleanup_follows_the_pinned_directory_after_rename() {
    let directory = tempfile::tempdir().unwrap();
    let original = directory.path().join("original");
    let moved = directory.path().join("moved");
    std::fs::create_dir(&original).unwrap();
    let path = original.join("server.sock");
    let engine = Arc::new(Engine::new(tokio::runtime::Handle::current()));
    let (id, _) = engine
        .listen(
            Address::Unix {
                path: Some(path.to_str().unwrap().into()),
            },
            Arc::new(|_| Ok(())),
        )
        .unwrap();
    std::fs::rename(&original, &moved).unwrap();
    std::fs::create_dir(&original).unwrap();
    std::fs::write(&path, "replacement").unwrap();
    engine.stop(id, false);
    engine.finished(id).await.unwrap();
    assert!(!moved.join("server.sock").exists());
    assert_eq!(std::fs::read_to_string(path).unwrap(), "replacement");
}