use std::{net::SocketAddr, time::Duration};

use tokio::io::{AsyncReadExt, AsyncWriteExt};

use axum::{Router, routing::get};
use rs_zero::{
    core::ServiceGroup,
    rest::{ApiResponse, RestConfig, RestServer, RestService},
    rpc::{RpcClientConfig, TonicHealthService, connect_channel},
};
use tonic_health::pb::{HealthCheckRequest, health_client::HealthClient};

#[tokio::test]
async fn rest_service_runs_inside_service_group() {
    let addr = unused_addr().await;
    let router = Router::new().route("/ready", get(|| async { ApiResponse::success("ok") }));
    let service = RestService::new("rest", addr, RestServer::new(RestConfig::default(), router));
    let mut group = ServiceGroup::new();
    let handle = group.handle();
    group.add(service);

    let runner = tokio::spawn(async move {
        group
            .start_with_shutdown(async { std::future::pending::<()>().await })
            .await
    });

    let body = read_http_body(addr, "/ready").await;
    assert!(body.contains("\"success\":true"));

    handle.stop();
    runner.await.expect("join").expect("service group");
}

#[tokio::test]
async fn tonic_health_service_runs_inside_service_group() {
    let addr = unused_addr().await;
    let mut group = ServiceGroup::new();
    let handle = group.handle();
    group.add(TonicHealthService::new("health", addr));

    let runner = tokio::spawn(async move {
        group
            .start_with_shutdown(async { std::future::pending::<()>().await })
            .await
    });

    wait_until_ready(|| async {
        connect_channel(&RpcClientConfig::new(format!("http://{addr}")))
            .await
            .map(|_| ())
    })
    .await;

    let config = RpcClientConfig::new(format!("http://{addr}"));
    let channel = connect_channel(&config).await.expect("connect");
    let mut client = HealthClient::new(channel);
    let response = client
        .check(HealthCheckRequest {
            service: String::new(),
        })
        .await
        .expect("health check");

    assert_eq!(response.into_inner().status, 1);

    handle.stop();
    runner.await.expect("join").expect("service group");
}

async fn unused_addr() -> SocketAddr {
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
        .await
        .expect("bind");
    let addr = listener.local_addr().expect("local addr");
    drop(listener);
    addr
}

async fn read_http_body(addr: SocketAddr, path: &str) -> String {
    wait_until_ready(|| async move {
        let stream = tokio::net::TcpStream::connect(addr).await?;
        stream.writable().await?;
        Ok::<(), std::io::Error>(())
    })
    .await;

    let mut stream = tokio::net::TcpStream::connect(addr).await.expect("connect");
    let request = format!("GET {path} HTTP/1.1\r\nHost: {addr}\r\nConnection: close\r\n\r\n");
    stream
        .write_all(request.as_bytes())
        .await
        .expect("write request");

    let mut bytes = Vec::new();
    stream.read_to_end(&mut bytes).await.expect("read response");

    String::from_utf8(bytes).expect("utf8")
}

async fn wait_until_ready<F, Fut, E>(mut probe: F)
where
    F: FnMut() -> Fut,
    Fut: std::future::Future<Output = Result<(), E>>,
{
    let deadline = tokio::time::Instant::now() + Duration::from_secs(1);
    loop {
        if probe().await.is_ok() {
            return;
        }
        assert!(
            tokio::time::Instant::now() < deadline,
            "service did not become ready"
        );
        tokio::time::sleep(Duration::from_millis(10)).await;
    }
}
