mod common;

#[cfg(test)]
mod transport_config {
    use std::sync::Arc;

    use dittolive_ditto::transport::TransportConfig;

    use super::common::*;

    fn fully_customized_config() -> TransportConfig {
        let mut config = TransportConfig::new();

        config.peer_to_peer.bluetooth_le.enabled = true;
        config.peer_to_peer.lan.enabled = true;
        config.peer_to_peer.lan.mdns_enabled = false;
        config.peer_to_peer.lan.multicast_enabled = true;

        config.global.routing_hint = 123;
        config.global.sync_group = 99;

        config
            .connect
            .tcp_servers
            .insert("1.2.3.4:5678".to_string());
        config
            .connect
            .tcp_servers
            .insert("2.3.4.5:6789".to_string());
        config
            .connect
            .websocket_urls
            .insert("wss://example.com:123".to_string());
        config
            .connect
            .websocket_urls
            .insert("wss://example.com:456".to_string());
        config.connect.retry_interval = std::time::Duration::from_millis(12345);

        config.listen.tcp.enabled = true;
        config.listen.tcp.interface_ip = "3.4.5.6".to_string();
        config.listen.tcp.port = 789;

        config.listen.http.enabled = true;
        config.listen.http.interface_ip = "4.5.6.7".to_string();
        config.listen.http.port = 123;
        config.listen.http.static_content_path = Some("blah".into());
        config.listen.http.websocket_sync = true;
        config.listen.http.tls_key_path = Some("some-tls-key-path".into());
        config.listen.http.tls_certificate_path = Some("some-tls-certificate-path".into());

        config
    }

    // If `TransportConfig` has been updated and you need to update these literals below then
    // placing the following snippet in the middle of the test body will dump cleanly formatted
    // payloads which you can copy/paste to update the definitions. Note that there's a discrepancy
    // between the `tcp_servers` and `websocket_urls` in the fully customized config above and the
    // expected payloads below. This is because the `HashSet` serialization is non-deterministic and
    // so we modify the config in the test below before serializing it to ensure determinism.
    //
    // ```
    // println!("{}", json_encoded);
    // println!("{:#04X?}", cbor_encoded);
    // ```
    pub(crate) const CONFIG_JSON: &str = r#"{"peer_to_peer":{"bluetooth_le":{"enabled":true},"lan":{"enabled":true,"mdns_enabled":false,"multicast_enabled":true}},"connect":{"tcp_servers":["1.2.3.4:5678"],"websocket_urls":["wss://example.com:123"],"retry_interval":12345},"listen":{"tcp":{"enabled":true,"interface_ip":"3.4.5.6","port":789},"http":{"enabled":true,"interface_ip":"4.5.6.7","port":123,"static_content_path":"blah","websocket_sync":true,"tls_key_path":"some-tls-key-path","tls_certificate_path":"some-tls-certificate-path"}},"global":{"sync_group":99,"routing_hint":123}}"#;
    pub(crate) const CONFIG_CBOR: &[u8] = &[
        0xA4, 0x6C, 0x70, 0x65, 0x65, 0x72, 0x5F, 0x74, 0x6F, 0x5F, 0x70, 0x65, 0x65, 0x72, 0xA2,
        0x6C, 0x62, 0x6C, 0x75, 0x65, 0x74, 0x6F, 0x6F, 0x74, 0x68, 0x5F, 0x6C, 0x65, 0xA1, 0x67,
        0x65, 0x6E, 0x61, 0x62, 0x6C, 0x65, 0x64, 0xF5, 0x63, 0x6C, 0x61, 0x6E, 0xA3, 0x67, 0x65,
        0x6E, 0x61, 0x62, 0x6C, 0x65, 0x64, 0xF5, 0x6C, 0x6D, 0x64, 0x6E, 0x73, 0x5F, 0x65, 0x6E,
        0x61, 0x62, 0x6C, 0x65, 0x64, 0xF4, 0x71, 0x6D, 0x75, 0x6C, 0x74, 0x69, 0x63, 0x61, 0x73,
        0x74, 0x5F, 0x65, 0x6E, 0x61, 0x62, 0x6C, 0x65, 0x64, 0xF5, 0x67, 0x63, 0x6F, 0x6E, 0x6E,
        0x65, 0x63, 0x74, 0xA3, 0x6B, 0x74, 0x63, 0x70, 0x5F, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72,
        0x73, 0x81, 0x6C, 0x31, 0x2E, 0x32, 0x2E, 0x33, 0x2E, 0x34, 0x3A, 0x35, 0x36, 0x37, 0x38,
        0x6E, 0x77, 0x65, 0x62, 0x73, 0x6F, 0x63, 0x6B, 0x65, 0x74, 0x5F, 0x75, 0x72, 0x6C, 0x73,
        0x81, 0x75, 0x77, 0x73, 0x73, 0x3A, 0x2F, 0x2F, 0x65, 0x78, 0x61, 0x6D, 0x70, 0x6C, 0x65,
        0x2E, 0x63, 0x6F, 0x6D, 0x3A, 0x31, 0x32, 0x33, 0x6E, 0x72, 0x65, 0x74, 0x72, 0x79, 0x5F,
        0x69, 0x6E, 0x74, 0x65, 0x72, 0x76, 0x61, 0x6C, 0x19, 0x30, 0x39, 0x66, 0x6C, 0x69, 0x73,
        0x74, 0x65, 0x6E, 0xA2, 0x63, 0x74, 0x63, 0x70, 0xA3, 0x67, 0x65, 0x6E, 0x61, 0x62, 0x6C,
        0x65, 0x64, 0xF5, 0x6C, 0x69, 0x6E, 0x74, 0x65, 0x72, 0x66, 0x61, 0x63, 0x65, 0x5F, 0x69,
        0x70, 0x67, 0x33, 0x2E, 0x34, 0x2E, 0x35, 0x2E, 0x36, 0x64, 0x70, 0x6F, 0x72, 0x74, 0x19,
        0x03, 0x15, 0x64, 0x68, 0x74, 0x74, 0x70, 0xA7, 0x67, 0x65, 0x6E, 0x61, 0x62, 0x6C, 0x65,
        0x64, 0xF5, 0x6C, 0x69, 0x6E, 0x74, 0x65, 0x72, 0x66, 0x61, 0x63, 0x65, 0x5F, 0x69, 0x70,
        0x67, 0x34, 0x2E, 0x35, 0x2E, 0x36, 0x2E, 0x37, 0x64, 0x70, 0x6F, 0x72, 0x74, 0x18, 0x7B,
        0x73, 0x73, 0x74, 0x61, 0x74, 0x69, 0x63, 0x5F, 0x63, 0x6F, 0x6E, 0x74, 0x65, 0x6E, 0x74,
        0x5F, 0x70, 0x61, 0x74, 0x68, 0x64, 0x62, 0x6C, 0x61, 0x68, 0x6E, 0x77, 0x65, 0x62, 0x73,
        0x6F, 0x63, 0x6B, 0x65, 0x74, 0x5F, 0x73, 0x79, 0x6E, 0x63, 0xF5, 0x6C, 0x74, 0x6C, 0x73,
        0x5F, 0x6B, 0x65, 0x79, 0x5F, 0x70, 0x61, 0x74, 0x68, 0x71, 0x73, 0x6F, 0x6D, 0x65, 0x2D,
        0x74, 0x6C, 0x73, 0x2D, 0x6B, 0x65, 0x79, 0x2D, 0x70, 0x61, 0x74, 0x68, 0x74, 0x74, 0x6C,
        0x73, 0x5F, 0x63, 0x65, 0x72, 0x74, 0x69, 0x66, 0x69, 0x63, 0x61, 0x74, 0x65, 0x5F, 0x70,
        0x61, 0x74, 0x68, 0x78, 0x19, 0x73, 0x6F, 0x6D, 0x65, 0x2D, 0x74, 0x6C, 0x73, 0x2D, 0x63,
        0x65, 0x72, 0x74, 0x69, 0x66, 0x69, 0x63, 0x61, 0x74, 0x65, 0x2D, 0x70, 0x61, 0x74, 0x68,
        0x66, 0x67, 0x6C, 0x6F, 0x62, 0x61, 0x6C, 0xA2, 0x6A, 0x73, 0x79, 0x6E, 0x63, 0x5F, 0x67,
        0x72, 0x6F, 0x75, 0x70, 0x18, 0x63, 0x6C, 0x72, 0x6F, 0x75, 0x74, 0x69, 0x6E, 0x67, 0x5F,
        0x68, 0x69, 0x6E, 0x74, 0x18, 0x7B,
    ];

    #[test]
    fn serialization_roundtrips() {
        let mut config = fully_customized_config();

        // Check JSON {de,}serialization
        let json_encoded = String::from_utf8(::serde_json::to_vec(&config).unwrap()).unwrap();
        let json_decoded: TransportConfig = serde_json::from_str(&json_encoded).unwrap();
        assert_eq!(config, json_decoded);

        // Check CBOR {de,}serialization
        let cbor_encoded = serde_cbor::to_vec(&config).unwrap();
        let cbor_decoded: TransportConfig = serde_cbor::from_slice(&cbor_encoded).unwrap();
        assert_eq!(config, cbor_decoded);

        // Check that the serialized payloads match the literals above, but first remove one of each
        // of the `tcp_servers` and `websocket_urls` so we don't have to worry about
        // non-deterministic ordering of the `HashSet` serialization.
        config.connect.tcp_servers.clear();
        config
            .connect
            .tcp_servers
            .insert("1.2.3.4:5678".to_string());
        config.connect.websocket_urls.clear();
        config
            .connect
            .websocket_urls
            .insert("wss://example.com:123".to_string());

        // Check JSON payload
        let json_encoded = String::from_utf8(::serde_json::to_vec(&config).unwrap()).unwrap();
        assert_eq!(json_encoded, CONFIG_JSON);

        // Check CBOR payload
        let cbor_encoded = serde_cbor::to_vec(&config).unwrap();
        assert_eq!(cbor_encoded, CONFIG_CBOR);
    }

    #[test]
    fn set_customized_config() {
        let ditto = get_builder_ditto(None).unwrap();
        set_customized_config_impl(ditto);

        let ditto = get_revamp_ditto(None).unwrap();
        set_customized_config_impl(ditto);
    }

    fn set_customized_config_impl(ditto: dittolive_ditto::prelude::Ditto) {
        let config = fully_customized_config();
        ditto.set_transport_config(config);
    }

    /// A test to make sure we can connect to a static TCP server using a DNS name.
    /// Trying to connect to localhost:{PORT} should be sufficient to verify this
    /// (instead of the usual IPv4/IPv6 localhost address).
    #[tokio::test]
    async fn can_use_dns_names_for_static_tcp() -> anyhow::Result<()> {
        let attempts = 3;
        for _ in 0..attempts {
            let future = try_can_use_dns_names_for_static_tcp();
            let result = tokio::time::timeout(std::time::Duration::from_secs(12), future).await?;
            if let Ok(true) = result {
                return Ok(()); // passed test
            }

            // otherwise loop for next attempt
        }

        anyhow::bail!("failed after {attempts} attempts");
    }

    async fn try_can_use_dns_names_for_static_tcp() -> anyhow::Result<bool> {
        let app_id = AppId::generate();

        let ditto1 = get_builder_ditto(Some(app_id.clone())).unwrap();
        let ditto2 = get_builder_ditto(Some(app_id)).unwrap();

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

        ditto1.set_transport_config(server);

        let mut client = TransportConfig::new();
        client
            .connect
            .tcp_servers
            .insert("localhost:53623".to_string());
        ditto2.set_transport_config(client);

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

        let ditto1 = Arc::new(ditto1);
        assert_connected_peers(Arc::clone(&ditto1), 1).await;

        Ok(true)
    }

    /// Static peers should be forcibly disconnected when the SDK user removes them from
    /// TransportConfig. It is not sufficient to test this with a single pair, because if there are
    /// no static peers in the list the TCP client will be shut down.
    #[tokio::test]
    async fn will_disconnect_from_static_peer_when_static_peer_entry_is_removed(
    ) -> anyhow::Result<()> {
        let app_id = AppId::generate();

        let ditto1 = get_builder_ditto(Some(app_id.clone())).unwrap();
        let ditto2 = get_builder_ditto(Some(app_id.clone())).unwrap();
        let ditto3 = get_builder_ditto(Some(app_id)).unwrap();

        let mut server1 = TransportConfig::new();
        server1.listen.tcp = TcpListenConfig {
            enabled: true,
            interface_ip: "127.0.0.1".into(),
            port: 53633,
        };
        ditto1.set_transport_config(server1);

        let mut server2 = TransportConfig::new();
        server2.listen.tcp = TcpListenConfig {
            enabled: true,
            interface_ip: "127.0.0.1".into(),
            port: 53634,
        };
        ditto2.set_transport_config(server2);

        let mut client = TransportConfig::new();
        client
            .connect
            .tcp_servers
            .insert("localhost:53633".to_string());
        client
            .connect
            .tcp_servers
            .insert("localhost:53634".to_string());
        ditto3.set_transport_config(client);

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

        let ditto3 = Arc::new(ditto3);

        assert_connected_peers(Arc::clone(&ditto3), 2).await;

        let mut client = TransportConfig::new();
        client
            .connect
            .tcp_servers
            .insert("localhost:53633".to_string());
        ditto3.set_transport_config(client.clone());

        assert_connected_peers(Arc::clone(&ditto3), 1).await;

        let mut client = TransportConfig::new();
        client
            .connect
            .tcp_servers
            .insert("localhost:53633".to_string());
        client
            .connect
            .tcp_servers
            .insert("localhost:53634".to_string());
        ditto3.set_transport_config(client);

        assert_connected_peers(Arc::clone(&ditto3), 2).await;

        Ok(())
    }

    /// Check whether the provided Ditto instance has `count` number of connections. Retry for 10
    /// seconds. This function will panic if the `count` requirement is not met within 10 seconds.
    async fn assert_connected_peers(ditto: Arc<Ditto>, count: usize) {
        let f = tokio::spawn(async move {
            let mut cur_peers = 0;
            for _ in 0..10 {
                cur_peers = ditto.presence().graph().remote_peers.len();
                if cur_peers == count {
                    return Ok(());
                }

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

            Err(cur_peers)
        });

        let result = tokio::time::timeout(std::time::Duration::from_secs(11), f)
            .await
            .unwrap()
            .unwrap();

        if let Err(peers_count) = result {
            panic!("Expected {count} connected peer(s), got {peers_count}");
        }
    }
}
