//! Integration tests for UDP transport with mDNS discovery.

use std::time::Duration;

use anyhow::Error;
use culpa::throws;
use dittolive_ditto::{fs::TempRoot, prelude::*, transport::TransportConfig, DatabaseId};
use serde_json::json;
use tokio::sync::mpsc;

mod common;

/// Test that two Ditto instances can discover each other via mDNS
/// and replicate documents over UDP transport.
#[throws]
#[tokio::test]
async fn test_udp_mdns_replication() {
    // Use same app_id so peers can discover each other
    let app_id = DatabaseId::generate();

    // Create two Ditto instances with UDP+mDNS configuration
    let (ditto1, _root1) = create_udp_mdns_ditto(app_id.clone()).await?;
    let (ditto2, _root2) = create_udp_mdns_ditto(app_id).await?;

    // Start sync on both
    ditto1.sync().start()?;
    ditto2.sync().start()?;

    println!("Ditto instances created and sync started");

    // Register sync subscription on ditto2 to receive documents
    let _subscription = ditto2
        .sync()
        .register_subscription("SELECT * FROM test_docs")?;

    // Set up observer on ditto2 to detect replicated documents
    let (tx, mut rx) = mpsc::unbounded_channel::<usize>();
    let _observer = ditto2
        .store()
        .register_observer("SELECT * FROM test_docs", move |result| {
            let _ = tx.send(result.item_count());
        })?;

    // Wait for initial empty result
    let initial = tokio::time::timeout(Duration::from_secs(5), rx.recv())
        .await?
        .expect("Should receive initial observer callback");
    assert_eq!(initial, 0, "Should start with no documents");

    // Insert a document on ditto1
    ditto1
        .store()
        .execute((
            "INSERT INTO test_docs DOCUMENTS (:doc)",
            json!({"doc": {"message": "hello via udp"}}),
        ))
        .await?;

    // Wait for document to replicate to ditto2
    let mut received = false;
    for _ in 0..30 {
        if let Ok(Some(count)) = tokio::time::timeout(Duration::from_secs(1), rx.recv()).await {
            if count > 0 {
                received = true;
                break;
            }
        }
    }

    assert!(
        received,
        "Document should replicate via UDP+mDNS within 30 seconds"
    );
    println!("Document replicated successfully");

    // Verify the document content on ditto2
    let result = ditto2.store().execute("SELECT * FROM test_docs").await?;
    assert_eq!(result.item_count(), 1, "Should have exactly one document");

    let item = result.get_item(0).unwrap();
    let value: serde_json::Value = item.deserialize_value()?;
    assert_eq!(value["message"], "hello via udp");
}

async fn create_udp_mdns_ditto(app_id: DatabaseId) -> Result<(Ditto, TempRoot), DittoError> {
    // Create DittoConfig with system parameters
    let root = TempRoot::new();

    let connect = DittoConfigConnect::SmallPeersOnly { private_key: None };
    let mut config =
        DittoConfig::new(app_id.to_string(), connect).with_persistence_directory(root.root_path());

    // Configure system parameters for UDP+NGN
    config
        .system_parameters
        .insert("network_enable_ngn".into(), serde_json::Value::Bool(true));
    config
        .system_parameters
        .insert("replication_over_ngn".into(), serde_json::Value::Bool(true));
    config.system_parameters.insert(
        "transports_udp_port".into(),
        serde_json::Value::Number(0.into()),
    );
    config
        .system_parameters
        .insert("udp_server_enabled".into(), serde_json::Value::Bool(true));

    let ditto = Ditto::open(config).await?;
    DittoLogger::set_minimum_log_level(LogLevel::Debug);

    // Set license
    ditto.set_license_from_env("DITTO_LICENSE")?;

    // Configure transports: enable mDNS, disable everything else
    // Note: UDP server is enabled via system parameters above (udp_server_enabled,
    // transports_udp_port)
    let mut tc = TransportConfig::new();

    // Enable LAN with mDNS for peer discovery
    tc.peer_to_peer.lan.enabled = true;
    tc.peer_to_peer.lan.mdns_enabled = true;
    tc.peer_to_peer.lan.multicast_enabled = false; // Use mDNS only

    // Disable all other P2P transports (awdl and wifi_aware are not available)
    tc.peer_to_peer.bluetooth_le.enabled = false;

    // Disable TCP listener (UDP is enabled via system parameters)
    tc.listen.tcp.enabled = false;
    tc.listen.http.enabled = false;

    ditto.set_transport_config(tc);

    Ok((ditto, root))
}
