use crate::{pb::*, test_assert, TestAssertion};
use futures_util::{future, stream, SinkExt, StreamExt};
use hyper::client::conn::{Builder, SendRequest};
use hyper::client::connect::HttpConnector;
use hyper::client::service::{Connect, MakeService};
use std::net::SocketAddr;
use tokio::sync::mpsc;
use tonic::service::add_origin::AddOrigin;
use tonic::{metadata::MetadataValue, Code, Request, Response, Status};

pub type Client = TestServiceClient<AddOrigin<SendRequest<tonic::BoxBody>>>;
pub type UnimplementedClient = UnimplementedServiceClient<AddOrigin<SendRequest<tonic::BoxBody>>>;

tonic::client!(service = "grpc.testing.TestService", proto = "crate::pb");
tonic::client!(
    service = "grpc.testing.UnimplementedService",
    proto = "crate::pb"
);

const LARGE_REQ_SIZE: usize = 271828;
const LARGE_RSP_SIZE: i32 = 314159;
const REQUEST_LENGTHS: &'static [i32] = &[27182, 8, 1828, 45904];
const RESPONSE_LENGTHS: &'static [i32] = &[31415, 9, 2653, 58979];
const TEST_STATUS_MESSAGE: &'static str = "test status message";
const SPECIAL_TEST_STATUS_MESSAGE: &'static str =
    "\t\ntest with whitespace\r\nand Unicode BMP ☺ and non-BMP 😈\t\n";

pub async fn create(addr: SocketAddr) -> Result<Client, Box<dyn std::error::Error>> {
    let origin = http::Uri::from_shared(format!("http://{}", addr).into()).unwrap();

    let settings = Builder::new().http2_only(true).clone();
    let mut maker = Connect::new(HttpConnector::new(1), settings);

    let svc = maker.make_service(origin.clone()).await?;
    let svc = AddOrigin::new(svc, origin);

    Ok(TestServiceClient::new(svc))
}

pub async fn create_unimplemented(
    addr: SocketAddr,
) -> Result<UnimplementedClient, Box<dyn std::error::Error>> {
    let origin = http::Uri::from_shared(format!("http://{}", addr).into()).unwrap();

    let settings = Builder::new().http2_only(true).clone();
    let mut maker = Connect::new(HttpConnector::new(1), settings);

    let svc = maker.make_service(origin.clone()).await?;
    let svc = AddOrigin::new(svc, origin);

    Ok(UnimplementedServiceClient::new(svc))
}

pub async fn empty_unary(client: &mut Client, assertions: &mut Vec<TestAssertion>) {
    let result = client.empty_call(Request::new(Empty {})).await;

    assertions.push(test_assert!(
        "call must be successful",
        result.is_ok(),
        format!("result={:?}", result)
    ));

    if let Ok(response) = result {
        let body = response.into_inner();
        assertions.push(test_assert!(
            "body must not be null",
            body == Empty {},
            format!("body={:?}", body)
        ));
    }
}

pub async fn large_unary(client: &mut Client, assertions: &mut Vec<TestAssertion>) {
    use std::mem;
    let payload = crate::client_payload(LARGE_REQ_SIZE);
    let req = SimpleRequest {
        response_type: PayloadType::Compressable as i32,
        response_size: LARGE_RSP_SIZE,
        payload: Some(payload),
        ..Default::default()
    };

    let result = client.unary_call(Request::new(req)).await;

    assertions.push(test_assert!(
        "call must be successful",
        result.is_ok(),
        format!("result={:?}", result)
    ));

    if let Ok(response) = result {
        let body = response.into_inner();
        let payload_len = body.payload.as_ref().map(|p| p.body.len()).unwrap_or(0);

        assertions.push(test_assert!(
            "body must be 314159 bytes",
            payload_len == LARGE_RSP_SIZE as usize,
            format!("mem::size_of_val(&body)={:?}", mem::size_of_val(&body))
        ));
    }
}

// pub async fn cachable_unary(client: &mut Client, assertions: &mut Vec<TestAssertion>) {
//     let payload = Payload {
//         r#type: PayloadType::Compressable as i32,
//         body: format!("{:?}", std::time::Instant::now()).into_bytes(),
//     };
//     let req = SimpleRequest {
//         response_type: PayloadType::Compressable as i32,
//         payload: Some(payload),
//         ..Default::default()
//     };

//     client.
// }

pub async fn client_streaming(client: &mut Client, assertions: &mut Vec<TestAssertion>) {
    let requests = REQUEST_LENGTHS
        .iter()
        .map(|len| StreamingInputCallRequest {
            payload: Some(crate::client_payload(*len as usize)),
            ..Default::default()
        })
        .map(|v| Ok(v));

    let stream = stream::iter(requests);

    let result = client.streaming_input_call(Request::new(stream)).await;

    assertions.push(test_assert!(
        "call must be successful",
        result.is_ok(),
        format!("result={:?}", result)
    ));

    if let Ok(response) = result {
        let body = response.into_inner();

        assertions.push(test_assert!(
            "aggregated payload size must be 74922 bytes",
            body.aggregated_payload_size == 74922,
            format!("aggregated_payload_size={:?}", body.aggregated_payload_size)
        ));
    }
}

pub async fn server_streaming(client: &mut Client, assertions: &mut Vec<TestAssertion>) {
    let req = StreamingOutputCallRequest {
        response_parameters: RESPONSE_LENGTHS
            .iter()
            .map(|len| ResponseParameters::with_size(*len))
            .collect(),
        ..Default::default()
    };
    let req = Request::new(req);

    let result = client.streaming_output_call(req).await;

    assertions.push(test_assert!(
        "call must be successful",
        result.is_ok(),
        format!("result={:?}", result)
    ));

    if let Ok(response) = result {
        let responses = response
            .into_inner()
            .filter_map(|m| future::ready(m.ok()))
            .collect::<Vec<_>>()
            .await;
        let actual_response_lengths = crate::response_lengths(&responses);
        let asserts = vec![
            test_assert!(
                "there should be four responses",
                responses.len() == 4,
                format!("responses.len()={:?}", responses.len())
            ),
            test_assert!(
                "the response payload sizes should match input",
                RESPONSE_LENGTHS == actual_response_lengths.as_slice(),
                format!("{:?}={:?}", RESPONSE_LENGTHS, actual_response_lengths)
            ),
        ];

        assertions.extend(asserts);
    }
}

pub async fn ping_pong(client: &mut Client, assertions: &mut Vec<TestAssertion>) {
    let (mut tx, rx) = mpsc::unbounded_channel();
    tx.try_send(make_ping_pong_request(0)).unwrap();

    let result = client
        .full_duplex_call(Request::new(rx.map(|s| Ok(s))))
        .await;

    assertions.push(test_assert!(
        "call must be successful",
        result.is_ok(),
        format!("result={:?}", result)
    ));

    if let Ok(mut response) = result.map(Response::into_inner) {
        let mut responses = Vec::new();

        loop {
            match response.next().await {
                Some(result) => {
                    // TODO: what to do with this result?
                    responses.push(result.unwrap());
                    if responses.len() == REQUEST_LENGTHS.len() {
                        drop(tx);
                        break;
                    } else {
                        tx.send(make_ping_pong_request(responses.len()))
                            .await
                            .unwrap();
                    }
                }
                None => {
                    assertions.push(TestAssertion::Failed {
                        description:
                            "server should keep the stream open until the client closes it",
                        expression: "Stream terminated unexpectedly early",
                        why: None,
                    });
                    break;
                }
            }
        }

        let actual_response_lengths = crate::response_lengths(&responses);
        assertions.push(test_assert!(
            "there should be four responses",
            responses.len() == RESPONSE_LENGTHS.len(),
            format!("{:?}={:?}", responses.len(), RESPONSE_LENGTHS.len())
        ));
        assertions.push(test_assert!(
            "the response payload sizes should match input",
            RESPONSE_LENGTHS == actual_response_lengths.as_slice(),
            format!("{:?}={:?}", RESPONSE_LENGTHS, actual_response_lengths)
        ));
    }
}

pub async fn empty_stream(client: &mut Client, assertions: &mut Vec<TestAssertion>) {
    let stream = stream::iter(Vec::new());
    let result = client.full_duplex_call(Request::new(stream)).await;

    assertions.push(test_assert!(
        "call must be successful",
        result.is_ok(),
        format!("result={:?}", result)
    ));

    if let Ok(response) = result.map(Response::into_inner) {
        let responses = response.collect::<Vec<_>>().await;

        assertions.push(test_assert!(
            "there should be no responses",
            responses.len() == 0,
            format!("responses.len()={:?}", responses.len())
        ));
    }
}

pub async fn status_code_and_message(client: &mut Client, assertions: &mut Vec<TestAssertion>) {
    fn validate_response<T>(result: Result<T, Status>, assertions: &mut Vec<TestAssertion>)
    where
        T: std::fmt::Debug,
    {
        assertions.push(test_assert!(
            "call must fail with unknown status code",
            match &result {
                Err(status) => status.code() == Code::Unknown,
                _ => false,
            },
            format!("result={:?}", result)
        ));

        assertions.push(test_assert!(
            "call must respsond with expected status message",
            match &result {
                Err(status) => status.message() == TEST_STATUS_MESSAGE,
                _ => false,
            },
            format!("result={:?}", result)
        ));
    }

    let simple_req = SimpleRequest {
        response_status: Some(EchoStatus {
            code: 2,
            message: TEST_STATUS_MESSAGE.to_string(),
            ..Default::default()
        }),
        ..Default::default()
    };

    let duplex_req = StreamingOutputCallRequest {
        response_status: Some(EchoStatus {
            code: 2,
            message: TEST_STATUS_MESSAGE.to_string(),
            ..Default::default()
        }),
        ..Default::default()
    };

    let result = client.unary_call(Request::new(simple_req)).await;
    validate_response(result, assertions);

    let stream = stream::iter(vec![Ok(duplex_req)]);
    let result = match client.full_duplex_call(Request::new(stream)).await {
        Ok(response) => {
            let stream = response.into_inner();
            let responses = stream.collect::<Vec<_>>().await;
            Ok(responses)
        }
        Err(e) => Err(e),
    };

    validate_response(result, assertions);
}

pub async fn special_status_message(client: &mut Client, assertions: &mut Vec<TestAssertion>) {
    let req = SimpleRequest {
        response_status: Some(EchoStatus {
            code: 2,
            message: SPECIAL_TEST_STATUS_MESSAGE.to_string(),
            ..Default::default()
        }),
        ..Default::default()
    };

    let result = client.unary_call(Request::new(req)).await;

    assertions.push(test_assert!(
        "call must fail with unknown status code",
        match &result {
            Err(status) => status.code() == Code::Unknown,
            _ => false,
        },
        format!("result={:?}", result)
    ));

    assertions.push(test_assert!(
        "call must respsond with expected status message",
        match &result {
            Err(status) => status.message() == SPECIAL_TEST_STATUS_MESSAGE,
            _ => false,
        },
        format!("result={:?}", result)
    ));
}

pub async fn unimplemented_method(client: &mut Client, assertions: &mut Vec<TestAssertion>) {
    let result = client.unimplemented_call(Request::new(Empty {})).await;
    assertions.push(test_assert!(
        "call must fail with unimplemented status code",
        match &result {
            Err(status) => status.code() == Code::Unimplemented,
            _ => false,
        },
        format!("result={:?}", result)
    ));
}

pub async fn unimplemented_service(
    client: &mut UnimplementedClient,
    assertions: &mut Vec<TestAssertion>,
) {
    let result = client.unimplemented_call(Request::new(Empty {})).await;
    assertions.push(test_assert!(
        "call must fail with unimplemented status code",
        match &result {
            Err(status) => status.code() == Code::Unimplemented,
            _ => false,
        },
        format!("result={:?}", result)
    ));
}

pub async fn custom_metadata(client: &mut Client, assertions: &mut Vec<TestAssertion>) {
    let key1 = "x-grpc-test-echo-initial";
    let value1 = MetadataValue::from_str("test_initial_metadata_value").unwrap();
    let key2 = "x-grpc-test-echo-trailing-bin";
    let value2 = MetadataValue::from_bytes(&[0xab, 0xab, 0xab]);

    let req = SimpleRequest {
        response_type: PayloadType::Compressable as i32,
        response_size: LARGE_RSP_SIZE,
        payload: Some(crate::client_payload(LARGE_REQ_SIZE)),
        ..Default::default()
    };
    let mut req_unary = Request::new(req);
    req_unary.metadata_mut().insert(key1, value1.clone());
    req_unary.metadata_mut().insert_bin(key2, value2.clone());

    // TODO: custom metadata for fullduplex
    let stream = stream::iter(vec![Ok(make_ping_pong_request(0))]);
    let mut req_stream = Request::new(stream);
    req_stream.metadata_mut().insert(key1, value1.clone());
    req_stream.metadata_mut().insert_bin(key2, value2.clone());

    // let response = client
    //     .unary_call(req_unary)
    //     .await
    //     .expect("call should pass.");

    // assertions.push(test_assert!(
    //     "metadata string must match in unary",
    //     response.metadata().get(key1) == Some(&value1),
    //     format!("result={:?}", response.metadata().get(key1))
    // ));
    // assertions.push(test_assert!(
    //     "metadata bin must match in unary",
    //     response.metadata().get_bin(key2) == Some(&value2),
    //     format!("result={:?}", response.metadata().get_bin(key1))
    // ));


    let response = client
        .full_duplex_call(req_stream)
        .await
        .expect("call should pass.");

    assertions.push(test_assert!(
        "metadata string must match in unary",
        response.metadata().get(key1) == Some(&value1),
        format!("result={:?}", response.metadata().get(key1))
    ));

    let mut stream = response.into_inner();

    // while let Some(_) = stream.next().await {}

    let trailers = stream.trailers().await.unwrap().unwrap();

    assertions.push(test_assert!(
        "metadata bin must match in unary",
        trailers.get_bin(key2) == Some(&value2),
        format!("result={:?}", trailers.get_bin(key1))
    ));
}

fn make_ping_pong_request(idx: usize) -> StreamingOutputCallRequest {
    let req_len = REQUEST_LENGTHS[idx];
    let resp_len = RESPONSE_LENGTHS[idx];
    StreamingOutputCallRequest {
        response_parameters: vec![ResponseParameters::with_size(resp_len)],
        payload: Some(crate::client_payload(req_len as usize)),
        ..Default::default()
    }
}