use std::sync::Arc;

use anyhow::Error;
use culpa::throws;
use dittolive_ditto::{
    prelude::*,
    transport::{TcpListenConfig, TransportConfig},
    AppId,
};
use serde_json::json;

mod common;

#[throws]
#[tokio::test]
async fn test_execute_alter_system_supports_discovered_peers() {
    let ditto = common::get_builder_ditto(None)?;
    test_execute_alter_system_supports_discovered_peers_impl(ditto).await?;

    let ditto = common::get_revamp_ditto(None)?;
    test_execute_alter_system_supports_discovered_peers_impl(ditto).await?;

    let ditto = common::get_revamp_ditto_async(None).await?;
    test_execute_alter_system_supports_discovered_peers_impl(ditto).await?;
}

#[throws]
async fn test_execute_alter_system_supports_discovered_peers_impl(ditto: Ditto) {
    let query_result = ditto
        .store()
        .execute_v2("SHOW transports_discovered_peers")
        .await?;
    assert_eq!(query_result.item_count(), 1);
    let item = query_result
        .get_item(0)
        .unwrap()
        .deserialize_value::<serde_json::Value>()?;
    assert_eq!(item["transports_discovered_peers"], json!([]));

    let peers = "[{'address': 'tcp://localhost:1234'}, {'address': 'tcp://127.0.0.1:2345'}]";
    let _ = ditto
        .store()
        .execute_v2(format!(
            "ALTER SYSTEM SET transports_discovered_peers = {peers}"
        ))
        .await?;

    let peers =
        "[{'address': 'tcp://localhost:1234', 'type': 'force', 'discovery_hint': 'Q2RICGABLYKw'}]";
    let _ = ditto
        .store()
        .execute_v2(format!(
            "ALTER SYSTEM SET transports_discovered_peers = {peers}"
        ))
        .await?;

    let peers = "[{'address': 'tcp://localhost:1234', 'type': 'candidate'}]";
    let _ = ditto
        .store()
        .execute_v2(format!(
            "ALTER SYSTEM SET transports_discovered_peers = {peers}"
        ))
        .await?;

    let peers = "[{'address': 'tcp://localhost:1234', 'discovery_hint': 'Q2RICGABLYKw'}]";
    let _ = ditto
        .store()
        .execute_v2(format!(
            "ALTER SYSTEM SET transports_discovered_peers = {peers}"
        ))
        .await?;
}

#[throws]
#[tokio::test]
async fn test_execute_alter_system_validates_values() {
    let ditto = common::get_builder_ditto(None)?;
    test_execute_alter_system_validates_values_impl(ditto).await?;

    let ditto = common::get_revamp_ditto(None)?;
    test_execute_alter_system_validates_values_impl(ditto).await?;

    let ditto = common::get_revamp_ditto_async(None).await?;
    test_execute_alter_system_validates_values_impl(ditto).await?;
}

#[throws]
async fn test_execute_alter_system_validates_values_impl(ditto: Ditto) {
    ditto.disable_sync_with_v3()?;

    let query_result = ditto
        .store()
        .execute_v2("SHOW transports_discovered_peers")
        .await?;
    assert_eq!(query_result.item_count(), 1);
    let item = query_result
        .get_item(0)
        .unwrap()
        .deserialize_value::<serde_json::Value>()?;
    assert_eq!(item["transports_discovered_peers"], json!([]));

    // Wrong type; should be an array
    let peers = "12345";
    let result = ditto
        .store()
        .execute_v2(format!(
            "ALTER SYSTEM SET transports_discovered_peers = {peers}"
        ))
        .await;
    assert!(result.is_err());

    // Contains a bad key: 'wat'
    let peers = "[{'wat': 'tcp://localhost:1234'}, {'address': 'tcp://127.0.0.1:2345'}]";
    let result = ditto
        .store()
        .execute_v2(format!(
            "ALTER SYSTEM SET transports_discovered_peers = {peers}"
        ))
        .await;
    assert!(result.is_err());

    // Missing the 'address' key
    let peers = "[{'type': 'force', 'discovery_hint': 'abc123'}]";
    let result = ditto
        .store()
        .execute_v2(format!(
            "ALTER SYSTEM SET transports_discovered_peers = {peers}"
        ))
        .await;
    assert!(result.is_err());

    // 'address' is not a valid URL
    let peers = "[{'address': '127.0.0.1:12345}]";
    let result = ditto
        .store()
        .execute_v2(format!(
            "ALTER SYSTEM SET transports_discovered_peers = {peers}"
        ))
        .await;
    assert!(result.is_err());

    // Contains a bad value for the 'type' key
    let peers = "[{'address': 'tcp://localhost:1234', 'type': 'wat', 'discovery_hint': 'abc123'}]";
    let result = ditto
        .store()
        .execute_v2(format!(
            "ALTER SYSTEM SET transports_discovered_peers = {peers}"
        ))
        .await;
    assert!(result.is_err());
}

#[throws]
#[tokio::test]
async fn test_connect_to_discovered_peer() {
    let app_id = AppId::generate();

    let ditto1 = common::get_builder_ditto(Some(app_id.clone())).unwrap();
    let ditto2 = common::get_builder_ditto(Some(app_id.clone())).unwrap();
    test_connect_to_discovered_peer_impl(ditto1, ditto2).await?;

    let ditto1 = common::get_revamp_ditto(Some(app_id.clone())).unwrap();
    let ditto2 = common::get_revamp_ditto(Some(app_id.clone())).unwrap();
    test_connect_to_discovered_peer_impl(ditto1, ditto2).await?;

    let ditto1 = common::get_revamp_ditto_async(Some(app_id.clone()))
        .await
        .unwrap();
    let ditto2 = common::get_revamp_ditto_async(Some(app_id)).await.unwrap();
    test_connect_to_discovered_peer_impl(ditto1, ditto2).await?;
}

#[throws]
async fn test_connect_to_discovered_peer_impl(ditto1: Ditto, ditto2: Ditto) {
    const PORT: u16 = 12345;

    // Binds a TCP listener to PORT
    let mut config = TransportConfig::new();
    config.listen.tcp = TcpListenConfig {
        enabled: true,
        interface_ip: "127.0.0.1".into(),
        port: PORT,
    };
    ditto1.set_transport_config(config);

    // Get the discovery_hint for the first peer
    let discovery_hint = ditto1
        .store()
        .execute_v2("SELECT * from system:transports_info where _id = 'discovery_hint'")
        .await
        .unwrap()
        .iter()
        .next()
        .unwrap()
        .value();
    let CborValue::Text(discovery_hint) = discovery_hint.get("value").unwrap() else {
        panic!("Failed to get discovery hint");
    };

    // Binds a TCP listener to a random port
    let mut config = TransportConfig::new();
    config.listen.tcp = TcpListenConfig {
        enabled: true,
        interface_ip: "127.0.0.1".into(),
        port: 0,
    };
    ditto2.set_transport_config(config);

    ditto1.sync().start().unwrap();
    ditto2.sync().start().unwrap();

    let peers =
        format!("[{{'address': 'tcp://localhost:{PORT}', 'discovery_hint': '{discovery_hint}'}}]");
    let _ = ditto2
        .store()
        .execute_v2(format!(
            "ALTER SYSTEM SET transports_discovered_peers = {peers}"
        ))
        .await?;

    // repeatedly check presence to see if we now have a connection with the server
    let f = tokio::spawn(async move {
        for _ in 0..10 {
            let remote_peers = ditto2.presence().graph().remote_peers;
            if !remote_peers.is_empty() {
                return true; // pass if we find a remote peer
            }

            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
        }

        false // fail if we don't find a remote peer in 10 tries
    });

    assert!(tokio::time::timeout(std::time::Duration::from_secs(11), f).await??);
}

#[throws]
#[tokio::test]
async fn test_connect_discovered_peer_forced_connection_and_disconnection() {
    let app_id = AppId::generate();

    let ditto1 = common::get_builder_ditto(Some(app_id.clone())).unwrap();
    let ditto2 = common::get_builder_ditto(Some(app_id.clone())).unwrap();
    test_connect_discovered_peer_forced_connection_and_disconnection_impl(ditto1, ditto2).await?;

    let ditto1 = common::get_revamp_ditto(Some(app_id.clone())).unwrap();
    let ditto2 = common::get_revamp_ditto(Some(app_id.clone())).unwrap();
    test_connect_discovered_peer_forced_connection_and_disconnection_impl(ditto1, ditto2).await?;

    let ditto1 = common::get_revamp_ditto_async(Some(app_id.clone()))
        .await
        .unwrap();
    let ditto2 = common::get_revamp_ditto_async(Some(app_id)).await.unwrap();
    test_connect_discovered_peer_forced_connection_and_disconnection_impl(ditto1, ditto2).await?;
}

#[throws]
async fn test_connect_discovered_peer_forced_connection_and_disconnection_impl(
    ditto1: Ditto,
    ditto2: Ditto,
) {
    const PORT: u16 = 23456;

    // Binds a TCP listener to PORT
    let mut config = TransportConfig::new();
    config.listen.tcp = TcpListenConfig {
        enabled: true,
        interface_ip: "127.0.0.1".into(),
        port: PORT,
    };
    ditto1.set_transport_config(config);

    // Binds a TCP listener to a random port
    let mut config = TransportConfig::new();
    config.listen.tcp = TcpListenConfig {
        enabled: true,
        interface_ip: "127.0.0.1".into(),
        port: 0,
    };
    ditto2.set_transport_config(config);

    ditto1.sync().start().unwrap();
    ditto2.sync().start().unwrap();

    let peers = format!("[{{'address': 'tcp://localhost:{PORT}', 'type': 'force'}}]");
    let _ = ditto2
        .store()
        .execute_v2(format!(
            "ALTER SYSTEM SET transports_discovered_peers = {peers}"
        ))
        .await?;

    // repeatedly check presence to see if we now have a connection with the server
    let ditto2 = Arc::new(ditto2);
    let ditto2_copy = Arc::clone(&ditto2);
    let f = tokio::spawn(async move {
        for _ in 0..10 {
            let remote_peers = ditto2_copy.presence().graph().remote_peers;
            if !remote_peers.is_empty() {
                return true; // pass if we find a remote peer
            }

            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
        }

        false // fail if we don't find a remote peer in 10 tries
    });

    assert!(tokio::time::timeout(std::time::Duration::from_secs(11), f).await??);

    let _ = ditto2
        .store()
        .execute_v2("ALTER SYSTEM SET transports_discovered_peers = []")
        .await?;

    // repeatedly check presence to see if we have disconnected from the server
    let f = tokio::spawn(async move {
        for _ in 0..10 {
            let local_peer = ditto2.presence().graph().local_peer;
            if local_peer.connections.is_empty() {
                return true; // pass if we don't find a connection
            }

            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
        }

        false // fail if we find a connection
    });

    assert!(tokio::time::timeout(std::time::Duration::from_secs(11), f).await??);
}

#[throws]
#[tokio::test]
async fn test_discovered_peer_reconnect_after_toggling_sync() {
    let app_id = AppId::generate();

    let ditto1 = common::get_builder_ditto(Some(app_id.clone())).unwrap();
    let ditto2 = common::get_builder_ditto(Some(app_id.clone())).unwrap();
    test_discovered_peer_reconnect_after_toggling_sync_impl(ditto1, ditto2).await?;

    let ditto1 = common::get_revamp_ditto(Some(app_id.clone())).unwrap();
    let ditto2 = common::get_revamp_ditto(Some(app_id.clone())).unwrap();
    test_discovered_peer_reconnect_after_toggling_sync_impl(ditto1, ditto2).await?;

    let ditto1 = common::get_revamp_ditto_async(Some(app_id.clone()))
        .await
        .unwrap();
    let ditto2 = common::get_revamp_ditto_async(Some(app_id)).await.unwrap();
    test_discovered_peer_reconnect_after_toggling_sync_impl(ditto1, ditto2).await?;
}

#[throws]
async fn test_discovered_peer_reconnect_after_toggling_sync_impl(ditto1: Ditto, ditto2: Ditto) {
    const PORT: u16 = 23457;

    // Binds a TCP listener to PORT
    let mut config = TransportConfig::new();
    config.listen.tcp = TcpListenConfig {
        enabled: true,
        interface_ip: "127.0.0.1".into(),
        port: PORT,
    };
    ditto1.set_transport_config(config);

    // Binds a TCP listener to a random port
    let mut config = TransportConfig::new();
    config.listen.tcp = TcpListenConfig {
        enabled: true,
        interface_ip: "127.0.0.1".into(),
        port: 0,
    };
    ditto2.set_transport_config(config);

    ditto1.sync().start().unwrap();
    ditto2.sync().start().unwrap();

    let peers = format!("[{{'address': 'tcp://localhost:{PORT}'}}]");
    let _ = ditto2
        .store()
        .execute_v2(format!(
            "ALTER SYSTEM SET transports_discovered_peers = {peers}"
        ))
        .await?;

    // repeatedly check presence to see if we now have a connection with the server
    let ditto2 = Arc::new(ditto2);
    let ditto2_copy = Arc::clone(&ditto2);
    let f = tokio::spawn(async move {
        for _ in 0..10 {
            let remote_peers = ditto2_copy.presence().graph().remote_peers;
            if !remote_peers.is_empty() {
                return true; // pass if we find a remote peer
            }

            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
        }

        false // fail if we don't find a remote peer in 10 tries
    });

    assert!(tokio::time::timeout(std::time::Duration::from_secs(11), f).await??);

    // Stop sync and wait for disconnect
    ditto2.sync().stop();

    tokio::time::sleep(std::time::Duration::from_secs(1)).await;

    // repeatedly check presence to see if we have disconnected from the server
    let ditto2_copy = Arc::clone(&ditto2);
    let f = tokio::spawn(async move {
        for _ in 0..10 {
            let local_peer = ditto2_copy.presence().graph().local_peer;
            if local_peer.connections.is_empty() {
                return true; // pass if we don't find a connection
            }

            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
        }

        false // fail if we find a connection
    });

    assert!(tokio::time::timeout(std::time::Duration::from_secs(11), f).await??);

    // Start sync and wait for automatic re-connection
    ditto2.sync().start().unwrap();

    // repeatedly check presence to see if we now have a connection with the server
    let ditto2_copy = Arc::clone(&ditto2);
    let f = tokio::spawn(async move {
        for _ in 0..10 {
            let remote_peers = ditto2_copy.presence().graph().remote_peers;
            if !remote_peers.is_empty() {
                return true; // pass if we find a remote peer
            }

            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
        }

        false // fail if we don't find a remote peer in 10 tries
    });

    assert!(tokio::time::timeout(std::time::Duration::from_secs(11), f).await??);
}
