#[cfg(any(test, feature = "rpcchainvm"))]
mod common;

use std::{io::ErrorKind, time::Duration};

use avalanche_types::rpcchainvm::{
    database::{
        corruptabledb::Database as CorruptableDb,
        manager::{versioned_database::VersionedDatabase, DatabaseManager},
        memdb::Database as MemDb,
        rpcdb::{client::DatabaseClient, server::Server as RpcDb},
    },
    http::{client::Client as HttpClient, server::Server as HttpServer},
};
use jsonrpc_core::Response as JsonResp;
use semver::Version;
use tokio::net::TcpListener;
use tonic::transport::Endpoint;

use crate::common::*;

#[tokio::test]
async fn rpcdb_mutation_test() {
    let bar_value = "bar".as_bytes().to_vec();
    let baz_value = "baz".as_bytes().to_vec();

    let db = MemDb::new();
    let server = RpcDb::new(db);

    let listener = TcpListener::bind("127.0.0.1:1339").await.unwrap();
    tokio::spawn(async move {
        serve_test_database(server, listener).await.unwrap();
    });
    tokio::time::sleep(Duration::from_millis(100)).await;

    let client_conn = Endpoint::from_static("http://127.0.0.1:1339")
        .connect()
        .await
        .unwrap();

    let mut client = DatabaseClient::new(client_conn);

    let resp = client.put("foo".as_bytes(), "bar".as_bytes()).await;
    assert!(!resp.is_err());

    let resp = client.get("foo".as_bytes()).await;
    let value = resp.unwrap();
    assert_eq!(value, bar_value.clone());

    let resp = client.put("foo".as_bytes(), "baz".as_bytes()).await;
    assert!(!resp.is_err());

    let resp = client.get("foo".as_bytes()).await;
    let value = resp.unwrap();
    assert_eq!(value, baz_value.clone());

    let resp = client.get("fool".as_bytes()).await;
    assert!(resp.is_err());
    assert_eq!(resp.unwrap_err().kind(), ErrorKind::NotFound);

    let resp = client.close().await;
    assert!(!resp.is_err());

    let resp = client.get("foo".as_bytes()).await;
    assert!(resp.is_err());
    assert!(resp.unwrap_err().to_string().contains("database closed"));
}

#[tokio::test]
async fn corruptibledb_mutation_test() {
    let bar_value = "bar".as_bytes().to_vec();

    let memdb = MemDb::new();
    let mut corruptible = CorruptableDb::new(memdb);

    let resp = corruptible.put("foo".as_bytes(), "bar".as_bytes()).await;
    assert!(!resp.is_err());

    let resp = corruptible.get("foo".as_bytes()).await;
    let value = resp.unwrap();
    assert_eq!(value, bar_value.clone());

    let resp = corruptible.put("foo".as_bytes(), "baz".as_bytes()).await;
    assert!(!resp.is_err());

    let resp = corruptible.has("foo".as_bytes()).await;
    assert!(!resp.is_err());
    assert_eq!(resp.unwrap(), true);

    let resp = corruptible.get("fool".as_bytes()).await;
    assert!(resp.is_err());
    assert_eq!(resp.unwrap_err().kind(), ErrorKind::NotFound);

    let resp = corruptible.close().await;
    assert!(!resp.is_err());

    let resp = corruptible.put("foo".as_bytes(), "baz".as_bytes()).await;
    assert!(resp.is_err());
    assert!(resp.unwrap_err().to_string().contains("database closed"));
}

#[tokio::test]
async fn test_rpcdb_corruptible() {
    let bar_value = "bar".as_bytes().to_vec();

    let memdb = MemDb::new();
    let rpc_server = RpcDb::new(memdb);

    let listener = TcpListener::bind("127.0.0.1:1340").await.unwrap();
    tokio::spawn(async move {
        serve_test_database(rpc_server, listener).await.unwrap();
    });
    tokio::time::sleep(Duration::from_millis(100)).await;

    let client_conn = Endpoint::from_static("http://127.0.0.1:1340")
        .connect()
        .await
        .unwrap();

    let db = DatabaseClient::new(client_conn);
    let mut client = CorruptableDb::new(db);

    let resp = client.put("foo".as_bytes(), "bar".as_bytes()).await;
    assert!(resp.is_ok());

    let resp = client.get("fool".as_bytes()).await;
    assert!(resp.is_err());
    assert_eq!(resp.unwrap_err().kind(), ErrorKind::NotFound);

    let resp = client.get("foo".as_bytes()).await;
    assert!(!resp.is_err());
    assert_eq!(resp.unwrap(), bar_value);
}

#[tokio::test]
async fn test_db_manager() {
    let memdb = MemDb::new();
    let rpc_server = RpcDb::new(memdb);

    let listener = TcpListener::bind("127.0.0.1:1341").await.unwrap();
    tokio::spawn(async move {
        serve_test_database(rpc_server, listener).await.unwrap();
    });
    tokio::time::sleep(Duration::from_millis(100)).await;

    let client_conn = Endpoint::from_static("http://127.0.0.1:1341")
        .connect()
        .await
        .unwrap();

    let vdb = VersionedDatabase::new(DatabaseClient::new(client_conn), Version::new(0, 0, 1));

    let mut databases: Vec<VersionedDatabase> = Vec::new();
    databases.push(vdb);

    let manager = DatabaseManager::new_from_databases(databases);
    let db = manager.current().await.unwrap();

    let mut client = db.inner.write().await;
    let resp = client.put("foo".as_bytes(), "bar".as_bytes()).await;
    assert!(!resp.is_err());

    let resp = client.get("foo".as_bytes()).await;
    assert!(!resp.is_err());

    let resp = client.close().await;
    assert!(!resp.is_err());

    // known bug close hangs
    // let resp = manager.close().await;
    // assert!(!resp.is_err());

    let resp = client.get("foo".as_bytes()).await;
    assert!(resp.is_err());
    assert!(resp.unwrap_err().to_string().contains("database closed"));
}

#[tokio::test]
async fn test_http_service() {
    let mut handler = jsonrpc_core::IoHandler::new();
    handler.add_method("foo", |_params: jsonrpc_core::Params| async move {
        Ok(jsonrpc_core::Value::String(format!("Hello, from foo")))
    });

    handler.add_method("bar", |params: jsonrpc_core::Params| async move {
        let params: HttpBarParams = params.parse().unwrap();

        Ok(jsonrpc_core::Value::String(format!(
            "Hello, {}, from bar",
            params.name
        )))
    });

    let http_server = HttpServer::new(handler);
    let listener = TcpListener::bind("127.0.0.1:1234").await.unwrap();

    tokio::spawn(async move {
        serve_test_http_server(http_server, listener).await.unwrap();
    });

    tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;

    let client_conn = Endpoint::from_static("http://127.0.0.1:1234")
        .connect()
        .await
        .unwrap();

    let mut client = HttpClient::new(client_conn);

    let foo_request = generate_http_request("foo", "http://127.0.0.1:1234", &[]);
    let foo_resp = client.serve_http_simple(foo_request).await;
    assert!(!foo_resp.is_err());
    let foo_resp = foo_resp.unwrap();

    assert!(foo_resp.status().is_success());
    let foo_json_resp = JsonResp::from_json(foo_resp.body().as_str()).unwrap();

    let foo_output: jsonrpc_core::Output = match foo_json_resp {
        JsonResp::Single(val) => val,
        JsonResp::Batch(_) => panic!("Test should return single output"),
    };

    match foo_output {
        jsonrpc_core::Output::Success(_) => {}
        jsonrpc_core::Output::Failure(f) => panic!("inner resp invalid: {}", f.error),
    }

    let bar_request = generate_http_request("bar", "http://127.0.0.1:1234", &["John"]);
    let bar_resp = client.serve_http_simple(bar_request).await;
    assert!(!bar_resp.is_err());
    let bar_resp = bar_resp.unwrap();

    assert!(bar_resp.status().is_success());
    let bar_json_resp = JsonResp::from_json(bar_resp.body().as_str()).unwrap();

    let bar_output: jsonrpc_core::Output = match bar_json_resp {
        JsonResp::Single(val) => val,
        JsonResp::Batch(_) => panic!("Test should return single output"),
    };

    match bar_output {
        jsonrpc_core::Output::Success(_) => {}
        jsonrpc_core::Output::Failure(f) => panic!("inner resp invalid: {}", f.error),
    }
}
