use std::{sync::Arc, time::Duration};

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

mod common;

// ---------------------------------------------------------------------------
// Shared helpers to reduce boilerplate across discovery tests
// ---------------------------------------------------------------------------

const POLL_ATTEMPTS: usize = 10;
const POLL_INTERVAL: Duration = Duration::from_secs(1);
const POLL_TIMEOUT: Duration = Duration::from_secs(11);

/// Retrieve the discovery hint for a peer from `system:transports_info`.
async fn get_discovery_hint(ditto: &Ditto) -> String {
    let row = ditto
        .store()
        .execute("SELECT * from system:transports_info where _id = 'discovery_hint'")
        .await
        .unwrap()
        .iter()
        .next()
        .unwrap()
        .value();
    let CborValue::Text(hint) = row.get("value").unwrap() else {
        panic!("Failed to get discovery hint");
    };
    hint.clone()
}

/// Execute `ALTER SYSTEM SET transports_discovered_peers = <peers>`.
#[throws]
async fn set_discovered_peers(ditto: &Ditto, peers: &str) {
    let _ = ditto
        .store()
        .execute(format!(
            "ALTER SYSTEM SET transports_discovered_peers = {peers}"
        ))
        .await?;
}

/// Poll presence until at least one remote peer is found, then assert success.
async fn assert_has_remote_peer(ditto: Arc<Ditto>) -> Result<(), Error> {
    let found = tokio::time::timeout(POLL_TIMEOUT, async {
        for _ in 0..POLL_ATTEMPTS {
            if !ditto.presence().graph().remote_peers.is_empty() {
                return true;
            }
            tokio::time::sleep(POLL_INTERVAL).await;
        }
        false
    })
    .await?;
    assert!(found);
    Ok(())
}

/// Poll presence until the local peer has no connections, then assert success.
async fn assert_disconnected(ditto: Arc<Ditto>) -> Result<(), Error> {
    let disconnected = tokio::time::timeout(POLL_TIMEOUT, async {
        for _ in 0..POLL_ATTEMPTS {
            if ditto.presence().graph().local_peer.connections.is_empty() {
                return true;
            }
            tokio::time::sleep(POLL_INTERVAL).await;
        }
        false
    })
    .await?;
    assert!(disconnected);
    Ok(())
}

/// Build a TCP listen config bound to the given port on localhost.
fn tcp_listen_config(port: u16) -> TransportConfig {
    let mut config = TransportConfig::new();
    config.listen.tcp = TcpListenConfig {
        enabled: true,
        interface_ip: "127.0.0.1".into(),
        port,
    };
    config
}

/// Build an HTTP/WebSocket listen config bound to the given port on localhost.
fn websocket_listen_config(port: u16) -> TransportConfig {
    let mut config = TransportConfig::new();
    config.listen.http = HttpListenConfig {
        enabled: true,
        interface_ip: "127.0.0.1".into(),
        port,
        websocket_sync: true,
        ..Default::default()
    };
    config
}

/// Create a Ditto instance with UDP transport enabled via system parameters.
/// NGN must be enabled at init time, so this cannot use the builder API.
fn create_udp_ditto_sync(database_id: DatabaseId, udp_port: u16) -> Result<Ditto, DittoError> {
    let tempdir = tempfile::tempdir().expect("should create tempdir");
    let persistent_dir = tempdir.keep();
    let connect = DittoConfigConnect::SmallPeersOnly { private_key: None };
    let mut config = DittoConfig::new(database_id.to_string(), connect)
        .with_persistence_directory(persistent_dir);
    config
        .system_parameters
        .insert("network_enable_ngn".into(), json!(true));
    config
        .system_parameters
        .insert("replication_over_ngn".into(), json!(true));
    config
        .system_parameters
        .insert("udp_server_enabled".into(), json!(true));
    config
        .system_parameters
        .insert("transports_udp_port".into(), json!(udp_port));
    let ditto = Ditto::open_sync(config)?;
    ditto.set_license_from_env("DITTO_LICENSE")?;
    Ok(ditto)
}

/// Async variant of [`create_udp_ditto_sync`].
async fn create_udp_ditto_async(
    database_id: DatabaseId,
    udp_port: u16,
) -> Result<Ditto, DittoError> {
    let tempdir = tempfile::tempdir().expect("should create tempdir");
    let persistent_dir = tempdir.keep();
    let connect = DittoConfigConnect::SmallPeersOnly { private_key: None };
    let mut config = DittoConfig::new(database_id.to_string(), connect)
        .with_persistence_directory(persistent_dir);
    config
        .system_parameters
        .insert("network_enable_ngn".into(), json!(true));
    config
        .system_parameters
        .insert("replication_over_ngn".into(), json!(true));
    config
        .system_parameters
        .insert("udp_server_enabled".into(), json!(true));
    config
        .system_parameters
        .insert("transports_udp_port".into(), json!(udp_port));
    let ditto = Ditto::open(config).await?;
    ditto.set_license_from_env("DITTO_LICENSE")?;
    Ok(ditto)
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[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("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!([]));

    set_discovered_peers(
        &ditto,
        "[{'address': 'tcp://localhost:1234'}, {'address': 'tcp://127.0.0.1:2345'}]",
    )
    .await?;

    set_discovered_peers(
        &ditto,
        "[{'address': 'tcp://localhost:1234', 'type': 'force', 'discovery_hint': 'Q2RICGABLYKw'}]",
    )
    .await?;

    set_discovered_peers(
        &ditto,
        "[{'address': 'tcp://localhost:1234', 'type': 'candidate'}]",
    )
    .await?;

    set_discovered_peers(
        &ditto,
        "[{'address': 'tcp://localhost:1234', 'discovery_hint': 'Q2RICGABLYKw'}]",
    )
    .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) {
    let query_result = ditto
        .store()
        .execute("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 result = set_discovered_peers(&ditto, "12345").await;
    assert!(result.is_err());

    // Contains a bad key: 'wat'
    let result = set_discovered_peers(
        &ditto,
        "[{'wat': 'tcp://localhost:1234'}, {'address': 'tcp://127.0.0.1:2345'}]",
    )
    .await;
    assert!(result.is_err());

    // Missing the 'address' key
    let result =
        set_discovered_peers(&ditto, "[{'type': 'force', 'discovery_hint': 'abc123'}]").await;
    assert!(result.is_err());

    // 'address' is not a valid URL
    let result = set_discovered_peers(&ditto, "[{'address': '127.0.0.1:12345}]").await;
    assert!(result.is_err());

    // Contains a bad value for the 'type' key
    let result = set_discovered_peers(
        &ditto,
        "[{'address': 'tcp://localhost:1234', 'type': 'wat', 'discovery_hint': 'abc123'}]",
    )
    .await;
    assert!(result.is_err());
}

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

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

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

    let ditto1 = common::get_revamp_ditto_async(Some(database_id.clone()))
        .await
        .unwrap();
    let ditto2 = common::get_revamp_ditto_async(Some(database_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;

    ditto1.set_transport_config(tcp_listen_config(PORT));

    let discovery_hint = get_discovery_hint(&ditto1).await;

    // ditto2 listens on a random port
    ditto2.set_transport_config(tcp_listen_config(0));

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

    let peers =
        format!("[{{'address': 'tcp://localhost:{PORT}', 'discovery_hint': '{discovery_hint}'}}]");
    set_discovered_peers(&ditto2, &peers).await?;

    assert_has_remote_peer(Arc::new(ditto2)).await?;
}

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

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

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

    let ditto1 = common::get_revamp_ditto_async(Some(database_id.clone()))
        .await
        .unwrap();
    let ditto2 = common::get_revamp_ditto_async(Some(database_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;

    ditto1.set_transport_config(tcp_listen_config(PORT));
    ditto2.set_transport_config(tcp_listen_config(0));

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

    let peers = format!("[{{'address': 'tcp://localhost:{PORT}', 'type': 'force'}}]");
    let ditto2 = Arc::new(ditto2);
    set_discovered_peers(&ditto2, &peers).await?;

    assert_has_remote_peer(Arc::clone(&ditto2)).await?;

    set_discovered_peers(&ditto2, "[]").await?;

    assert_disconnected(ditto2).await?;
}

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

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

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

    let ditto1 = common::get_revamp_ditto_async(Some(database_id.clone()))
        .await
        .unwrap();
    let ditto2 = common::get_revamp_ditto_async(Some(database_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;

    ditto1.set_transport_config(tcp_listen_config(PORT));
    ditto2.set_transport_config(tcp_listen_config(0));

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

    let peers = format!("[{{'address': 'tcp://localhost:{PORT}'}}]");
    let ditto2 = Arc::new(ditto2);
    set_discovered_peers(&ditto2, &peers).await?;

    assert_has_remote_peer(Arc::clone(&ditto2)).await?;

    // Stop sync and wait for disconnect
    ditto2.sync().stop();
    tokio::time::sleep(Duration::from_secs(1)).await;

    assert_disconnected(Arc::clone(&ditto2)).await?;

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

    assert_has_remote_peer(Arc::clone(&ditto2)).await?;
}

#[throws]
#[tokio::test]
async fn test_connect_to_discovered_peer_websocket() {
    let database_id = DatabaseId::generate();

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

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

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

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

    ditto1.set_transport_config(websocket_listen_config(PORT));

    let discovery_hint = get_discovery_hint(&ditto1).await;

    // ditto2 doesn't need a listener, just a default config
    ditto2.set_transport_config(TransportConfig::new());

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

    let peers =
        format!("[{{'address': 'ws://localhost:{PORT}', 'discovery_hint': '{discovery_hint}'}}]");
    set_discovered_peers(&ditto2, &peers).await?;

    assert_has_remote_peer(Arc::new(ditto2)).await?;
}

#[throws]
#[tokio::test]
async fn test_connect_discovered_peer_websocket_forced_connection_and_disconnection() {
    let database_id = DatabaseId::generate();

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

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

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

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

    ditto1.set_transport_config(websocket_listen_config(PORT));

    // ditto2 doesn't need a listener, just a default config
    ditto2.set_transport_config(TransportConfig::new());

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

    let peers = format!("[{{'address': 'ws://localhost:{PORT}', 'type': 'force'}}]");
    let ditto2 = Arc::new(ditto2);
    set_discovered_peers(&ditto2, &peers).await?;

    assert_has_remote_peer(Arc::clone(&ditto2)).await?;

    set_discovered_peers(&ditto2, "[]").await?;

    assert_disconnected(ditto2).await?;
}

#[throws]
#[tokio::test]
async fn test_connect_to_discovered_peer_udp() {
    const PORT: u16 = 34569;
    let database_id = DatabaseId::generate();

    // UDP requires NGN enabled at init time, so we use DittoConfig (no builder path).
    let ditto1 = create_udp_ditto_sync(database_id.clone(), PORT)?;
    let ditto2 = create_udp_ditto_sync(database_id.clone(), 0)?;
    test_connect_to_discovered_peer_udp_impl(ditto1, ditto2, PORT).await?;

    let ditto1 = create_udp_ditto_async(database_id.clone(), PORT).await?;
    let ditto2 = create_udp_ditto_async(database_id, 0).await?;
    test_connect_to_discovered_peer_udp_impl(ditto1, ditto2, PORT).await?;
}

#[throws]
async fn test_connect_to_discovered_peer_udp_impl(ditto1: Ditto, ditto2: Ditto, port: u16) {
    // Disable all P2P transports so only the explicit UDP discovered peer is used
    let tc = TransportConfig::new();
    ditto1.set_transport_config(tc.clone());
    ditto2.set_transport_config(tc);

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

    let discovery_hint = get_discovery_hint(&ditto1).await;

    let peers =
        format!("[{{'address': 'udp://127.0.0.1:{port}', 'discovery_hint': '{discovery_hint}'}}]");
    set_discovered_peers(&ditto2, &peers).await?;

    assert_has_remote_peer(Arc::new(ditto2)).await?;
}

#[throws]
#[tokio::test]
async fn test_connect_discovered_peer_udp_forced_connection() {
    const PORT: u16 = 34570;
    let database_id = DatabaseId::generate();

    // UDP requires NGN enabled at init time, so we use DittoConfig (no builder path).
    let ditto1 = create_udp_ditto_sync(database_id.clone(), PORT)?;
    let ditto2 = create_udp_ditto_sync(database_id.clone(), 0)?;
    test_connect_discovered_peer_udp_forced_connection_impl(ditto1, ditto2, PORT).await?;

    let ditto1 = create_udp_ditto_async(database_id.clone(), PORT).await?;
    let ditto2 = create_udp_ditto_async(database_id, 0).await?;
    test_connect_discovered_peer_udp_forced_connection_impl(ditto1, ditto2, PORT).await?;
}

#[throws]
async fn test_connect_discovered_peer_udp_forced_connection_impl(
    ditto1: Ditto,
    ditto2: Ditto,
    port: u16,
) {
    // Disable all P2P transports so only the explicit UDP discovered peer is used
    let tc = TransportConfig::new();
    ditto1.set_transport_config(tc.clone());
    ditto2.set_transport_config(tc);

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

    // Force connection via UDP — unlike TCP, UDP/NGN teardown relies on inactivity
    // detection (~30s) rather than immediate socket closure, so we only verify the
    // connection is established without testing disconnection.
    let peers = format!("[{{'address': 'udp://127.0.0.1:{port}', 'type': 'force'}}]");
    set_discovered_peers(&ditto2, &peers).await?;

    assert_has_remote_peer(Arc::new(ditto2)).await?;
}
