/// Integration test for ICE restart between rtc (sansio) and webrtc
///
/// This test verifies that the rtc library can successfully handle ICE restart
/// when communicating with the webrtc library.
use anyhow::Result;
use bytes::BytesMut;
use sansio::Protocol;
use shared::{TaggedBytesMut, TransportContext, TransportProtocol};
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::net::UdpSocket;
use tokio::sync::Mutex;
use tokio::time::timeout;

use rtc::peer_connection::RTCPeerConnectionBuilder;
use rtc::peer_connection::configuration::RTCConfigurationBuilder;
use rtc::peer_connection::configuration::setting_engine::SettingEngine;
use rtc::peer_connection::event::RTCDataChannelEvent;
use rtc::peer_connection::event::RTCPeerConnectionEvent;
use rtc::peer_connection::message::RTCMessage;
use rtc::peer_connection::state::RTCIceConnectionState;
use rtc::peer_connection::state::RTCPeerConnectionState;
use rtc::peer_connection::transport::RTCDtlsRole;
use rtc::peer_connection::transport::RTCIceServer;
use rtc::peer_connection::transport::{CandidateConfig, CandidateHostConfig};
use webrtc::api::APIBuilder;
use webrtc::api::interceptor_registry::register_default_interceptors;
use webrtc::api::media_engine::MediaEngine;
use webrtc::data_channel::data_channel_init::RTCDataChannelInit;
use webrtc::ice_transport::ice_server::RTCIceServer as WebrtcIceServer;
use webrtc::interceptor::registry::Registry;
use webrtc::peer_connection::RTCPeerConnection as WebrtcPeerConnection;
use webrtc::peer_connection::configuration::RTCConfiguration as WebrtcRTCConfiguration;
use webrtc::peer_connection::offer_answer_options::RTCOfferOptions;
use webrtc::peer_connection::peer_connection_state::RTCPeerConnectionState as WebrtcRTCPeerConnectionState;
use webrtc::peer_connection::sdp::session_description::RTCSessionDescription as WebrtcRTCSessionDescription;

const DEFAULT_TIMEOUT_DURATION: Duration = Duration::from_secs(30);
const TEST_MESSAGE_1: &str = "Hello before restart!";
const TEST_MESSAGE_2: &str = "Hello after restart!";
const ECHO_MESSAGE_1: &str = "Echo before restart!";
const ECHO_MESSAGE_2: &str = "Echo after restart!";

/// Helper to create a webrtc peer connection
async fn create_webrtc_peer() -> Result<Arc<WebrtcPeerConnection>> {
    let mut media_engine = MediaEngine::default();
    media_engine.register_default_codecs()?;

    let mut registry = Registry::new();
    registry = register_default_interceptors(registry, &mut media_engine)?;

    let api = APIBuilder::new()
        .with_media_engine(media_engine)
        .with_interceptor_registry(registry)
        .build();

    let config = WebrtcRTCConfiguration {
        ice_servers: vec![WebrtcIceServer {
            urls: vec!["stun:stun.l.google.com:19302".to_owned()],
            ..Default::default()
        }],
        ..Default::default()
    };

    let peer_connection = api.new_peer_connection(config).await?;
    Ok(Arc::new(peer_connection))
}

/// Test ICE restart between webrtc (offerer) and rtc (answerer)
#[tokio::test]
async fn test_ice_restart_interop() -> Result<()> {
    env_logger::builder()
        .filter_level(log::LevelFilter::Info)
        .is_test(true)
        .try_init()
        .ok();

    log::info!("Starting ICE restart interop test: webrtc <-> rtc");

    // Track received messages on both sides
    let webrtc_received_messages = Arc::new(Mutex::new(Vec::<String>::new()));
    let webrtc_received_messages_clone = Arc::clone(&webrtc_received_messages);
    let rtc_received_messages = Arc::new(Mutex::new(Vec::<String>::new()));

    // Track connection states
    let webrtc_connected = Arc::new(Mutex::new(false));
    let webrtc_connected_clone = Arc::clone(&webrtc_connected);

    // Create webrtc peer (will be the offerer)
    let webrtc_pc = create_webrtc_peer().await?;
    log::info!("Created webrtc peer connection");

    // Monitor webrtc peer connection state
    webrtc_pc.on_peer_connection_state_change(Box::new(move |state| {
        log::info!("WebRTC peer connection state changed: {}", state);
        let connected = Arc::clone(&webrtc_connected_clone);
        Box::pin(async move {
            if state == WebrtcRTCPeerConnectionState::Connected {
                *connected.lock().await = true;
            } else if state == WebrtcRTCPeerConnectionState::Failed {
                *connected.lock().await = false;
            }
        })
    }));

    // Set up data channel on webrtc side
    let dc_label = "test-channel";
    let webrtc_dc = webrtc_pc
        .create_data_channel(
            dc_label,
            Some(RTCDataChannelInit {
                ordered: Some(true),
                ..Default::default()
            }),
        )
        .await?;
    log::info!("Created webrtc data channel: {}", dc_label);

    // Set up webrtc data channel handlers
    webrtc_dc.on_open(Box::new(move || {
        log::info!("WebRTC data channel opened");
        Box::pin(async {})
    }));

    webrtc_dc.on_message(Box::new(move |msg| {
        let messages = Arc::clone(&webrtc_received_messages_clone);
        Box::pin(async move {
            let msg_str = String::from_utf8_lossy(&msg.data);
            log::info!("WebRTC received message: {}", msg_str);
            messages.lock().await.push(msg_str.to_string());
        })
    }));

    // Create webrtc offer
    let webrtc_offer = webrtc_pc.create_offer(None).await?;
    log::info!("WebRTC peer created offer");

    // Set local description on webrtc peer
    webrtc_pc
        .set_local_description(webrtc_offer.clone())
        .await?;
    log::info!("WebRTC peer set local description");

    // Create rtc peer (answerer)
    let rtc_socket = UdpSocket::bind("127.0.0.1:0").await?;
    let rtc_local_addr = rtc_socket.local_addr()?;
    log::info!("RTC peer bound to {}", rtc_local_addr);

    let mut rtc_setting_engine = SettingEngine::default();
    rtc_setting_engine.set_answering_dtls_role(RTCDtlsRole::Client)?;

    let rtc_config = RTCConfigurationBuilder::new()
        .with_ice_servers(vec![RTCIceServer {
            urls: vec!["stun:stun.l.google.com:19302".to_owned()],
            ..Default::default()
        }])
        .build();

    let mut rtc_pc = RTCPeerConnectionBuilder::new()
        .with_configuration(rtc_config)
        .with_setting_engine(rtc_setting_engine)
        .build()?;
    log::info!("Created rtc peer connection");

    // Add local candidate for rtc peer
    let rtc_candidate = CandidateHostConfig {
        base_config: CandidateConfig {
            network: "udp".to_owned(),
            address: rtc_local_addr.ip().to_string(),
            port: rtc_local_addr.port(),
            component: 1,
            ..Default::default()
        },
        ..Default::default()
    }
    .new_candidate_host()?;
    let rtc_candidate_init =
        rtc::peer_connection::transport::RTCIceCandidate::from(&rtc_candidate).to_json()?;
    rtc_pc.add_local_candidate(rtc_candidate_init)?;

    // Convert webrtc offer to rtc format
    let rtc_offer =
        rtc::peer_connection::sdp::RTCSessionDescription::offer(webrtc_offer.sdp.clone())?;

    // Set remote description on rtc peer
    rtc_pc.set_remote_description(rtc_offer)?;
    log::info!("RTC peer set remote description");

    // Create answer
    let rtc_answer = rtc_pc.create_answer(None)?;
    log::info!("RTC peer created answer");

    // Set local description on rtc peer
    rtc_pc.set_local_description(rtc_answer.clone())?;
    log::info!("RTC peer set local description");

    // Convert rtc answer to webrtc format
    let webrtc_answer = WebrtcRTCSessionDescription::answer(rtc_answer.sdp)?;

    // Set remote description on webrtc peer
    webrtc_pc.set_remote_description(webrtc_answer).await?;
    log::info!("WebRTC peer set remote description");

    // Event loop for rtc peer
    let mut buf = vec![0u8; 2000];
    let mut rtc_connected = false;
    let mut rtc_dc_id = None;

    let start = Instant::now();
    let connection_timeout = Duration::from_secs(10);

    log::info!("Starting event loop for initial connection...");

    // Initial connection establishment
    loop {
        if Instant::now().duration_since(start) > connection_timeout {
            return Err(anyhow::anyhow!("Initial connection timeout"));
        }

        // Poll writes
        while let Some(msg) = rtc_pc.poll_write() {
            rtc_socket
                .send_to(&msg.message, msg.transport.peer_addr)
                .await?;
        }

        // Poll events
        while let Some(event) = rtc_pc.poll_event() {
            match event {
                RTCPeerConnectionEvent::OnIceConnectionStateChangeEvent(state) => {
                    log::info!("RTC ICE connection state changed: {}", state);
                    if state == RTCIceConnectionState::Connected {
                        log::info!("RTC ICE is connected");
                    }
                }
                RTCPeerConnectionEvent::OnConnectionStateChangeEvent(state) => {
                    log::info!("RTC peer connection state changed: {}", state);
                    if state == RTCPeerConnectionState::Connected {
                        log::info!("RTC peer connection is connected");
                        rtc_connected = true;
                    }
                }
                RTCPeerConnectionEvent::OnDataChannel(dc_event) => match dc_event {
                    RTCDataChannelEvent::OnOpen(channel_id) => {
                        if let Some(dc) = rtc_pc.data_channel(channel_id) {
                            log::info!("RTC data channel '{}'-'{}' opened", dc.label(), dc.id());
                            rtc_dc_id = Some(channel_id);
                        }
                    }
                    _ => {}
                },
                _ => {}
            }
        }

        while let Some(message) = rtc_pc.poll_read() {
            match message {
                RTCMessage::RtpPacket(_, _) => {}
                RTCMessage::RtcpPacket(_, _) => {}
                RTCMessage::DataChannelMessage(_channel_id, data_channel_message) => {
                    let msg_str = String::from_utf8_lossy(&data_channel_message.data);
                    log::info!("RTC received message: {}", msg_str);
                    rtc_received_messages.lock().await.push(msg_str.to_string());
                }
            }
        }

        // Check if both peers are connected
        if rtc_connected && *webrtc_connected.lock().await {
            log::info!("Both peers connected!");
            break;
        }

        // Handle timeout
        let timeout_instant = rtc_pc
            .poll_timeout()
            .unwrap_or(Instant::now() + DEFAULT_TIMEOUT_DURATION);
        let delay = timeout_instant.saturating_duration_since(Instant::now());

        if delay.is_zero() {
            rtc_pc.handle_timeout(Instant::now()).ok();
            continue;
        }

        // Wait for socket data or timeout
        match timeout(
            delay.min(Duration::from_millis(100)),
            rtc_socket.recv_from(&mut buf),
        )
        .await
        {
            Ok(Ok((n, peer_addr))) => {
                rtc_pc.handle_read(TaggedBytesMut {
                    now: Instant::now(),
                    transport: TransportContext {
                        local_addr: rtc_local_addr,
                        peer_addr,
                        ecn: None,
                        transport_protocol: TransportProtocol::UDP,
                    },
                    message: BytesMut::from(&buf[..n]),
                })?;
            }
            Ok(Err(e)) => return Err(e.into()),
            Err(_) => {
                rtc_pc.handle_timeout(Instant::now()).ok();
            }
        }
    }

    // Test data channel communication before restart
    log::info!("Waiting for data channel to open...");

    // Wait for data channel to open
    let dc_start = Instant::now();
    let dc_timeout = Duration::from_secs(5);

    loop {
        if Instant::now().duration_since(dc_start) > dc_timeout {
            return Err(anyhow::anyhow!("Data channel open timeout"));
        }

        while let Some(msg) = rtc_pc.poll_write() {
            rtc_socket
                .send_to(&msg.message, msg.transport.peer_addr)
                .await?;
        }

        while let Some(event) = rtc_pc.poll_event() {
            if let RTCPeerConnectionEvent::OnDataChannel(RTCDataChannelEvent::OnOpen(channel_id)) =
                event
            {
                if let Some(dc) = rtc_pc.data_channel(channel_id) {
                    log::info!("RTC data channel '{}'-'{}' opened", dc.label(), dc.id());
                    rtc_dc_id = Some(channel_id);
                }
            }
        }

        if rtc_dc_id.is_some() {
            log::info!("Data channel ready!");
            break;
        }

        match timeout(Duration::from_millis(20), rtc_socket.recv_from(&mut buf)).await {
            Ok(Ok((n, peer_addr))) => {
                rtc_pc.handle_read(TaggedBytesMut {
                    now: Instant::now(),
                    transport: TransportContext {
                        local_addr: rtc_local_addr,
                        peer_addr,
                        ecn: None,
                        transport_protocol: TransportProtocol::UDP,
                    },
                    message: BytesMut::from(&buf[..n]),
                })?;
            }
            _ => {
                rtc_pc.handle_timeout(Instant::now()).ok();
            }
        }
    }

    log::info!("Testing data channel before ICE restart...");

    // Wait a bit for channel to stabilize
    tokio::time::sleep(Duration::from_millis(100)).await;

    // Send from webrtc to rtc
    webrtc_dc.send_text(TEST_MESSAGE_1.to_owned()).await?;
    log::info!("WebRTC sent: {}", TEST_MESSAGE_1);

    // Process messages - give more time for message to arrive
    for i in 0..100 {
        while let Some(msg) = rtc_pc.poll_write() {
            rtc_socket
                .send_to(&msg.message, msg.transport.peer_addr)
                .await?;
        }

        while let Some(_event) = rtc_pc.poll_event() {}

        while let Some(message) = rtc_pc.poll_read() {
            match message {
                RTCMessage::RtpPacket(_, _) => {}
                RTCMessage::RtcpPacket(_, _) => {}
                RTCMessage::DataChannelMessage(_channel_id, data_channel_message) => {
                    let msg_str = String::from_utf8_lossy(&data_channel_message.data);
                    log::info!("RTC received message: {}", msg_str);
                    rtc_received_messages.lock().await.push(msg_str.to_string());
                }
            }
        }

        let messages = rtc_received_messages.lock().await;
        if messages.contains(&TEST_MESSAGE_1.to_string()) {
            log::info!("RTC confirmed receipt after {} iterations", i + 1);
            break;
        }
        drop(messages);

        match timeout(Duration::from_millis(50), rtc_socket.recv_from(&mut buf)).await {
            Ok(Ok((n, peer_addr))) => {
                rtc_pc.handle_read(TaggedBytesMut {
                    now: Instant::now(),
                    transport: TransportContext {
                        local_addr: rtc_local_addr,
                        peer_addr,
                        ecn: None,
                        transport_protocol: TransportProtocol::UDP,
                    },
                    message: BytesMut::from(&buf[..n]),
                })?;
            }
            _ => {
                rtc_pc.handle_timeout(Instant::now()).ok();
            }
        }
    }

    // Send from rtc to webrtc
    if let Some(channel_id) = rtc_dc_id {
        if let Some(mut dc) = rtc_pc.data_channel(channel_id) {
            dc.send_text(ECHO_MESSAGE_1.to_owned())?;
            log::info!("RTC sent: {}", ECHO_MESSAGE_1);
        }
    }

    // Process echo
    for _ in 0..50 {
        while let Some(msg) = rtc_pc.poll_write() {
            rtc_socket
                .send_to(&msg.message, msg.transport.peer_addr)
                .await?;
        }

        let messages = webrtc_received_messages.lock().await;
        if messages.contains(&ECHO_MESSAGE_1.to_string()) {
            break;
        }
        drop(messages);

        tokio::time::sleep(Duration::from_millis(20)).await;
    }

    // Verify messages before restart
    {
        let rtc_messages = rtc_received_messages.lock().await;
        assert!(
            rtc_messages.contains(&TEST_MESSAGE_1.to_string()),
            "RTC should have received test message before restart"
        );
    }
    {
        let webrtc_messages = webrtc_received_messages.lock().await;
        assert!(
            webrtc_messages.contains(&ECHO_MESSAGE_1.to_string()),
            "WebRTC should have received echo message before restart"
        );
    }

    log::info!("Messages exchanged successfully before ICE restart");

    // Perform ICE restart from webrtc side
    log::info!("Performing ICE restart...");

    let restart_offer = webrtc_pc
        .create_offer(Some(RTCOfferOptions {
            ice_restart: true,
            ..Default::default()
        }))
        .await?;
    log::info!("WebRTC created restart offer");

    webrtc_pc
        .set_local_description(restart_offer.clone())
        .await?;
    log::info!("WebRTC set local description for restart");

    // Convert restart offer to rtc format
    let rtc_restart_offer =
        rtc::peer_connection::sdp::RTCSessionDescription::offer(restart_offer.sdp.clone())?;

    // Process restart on rtc peer
    rtc_pc.set_remote_description(rtc_restart_offer)?;
    log::info!("RTC peer set remote description for restart");

    let rtc_restart_answer = rtc_pc.create_answer(None)?;
    log::info!("RTC peer created restart answer");

    rtc_pc.set_local_description(rtc_restart_answer.clone())?;
    log::info!("RTC peer set local description for restart");

    // Convert restart answer to webrtc format
    let webrtc_restart_answer = WebrtcRTCSessionDescription::answer(rtc_restart_answer.sdp)?;

    webrtc_pc
        .set_remote_description(webrtc_restart_answer)
        .await?;
    log::info!("WebRTC peer set remote description for restart");

    // Reset connection flags
    rtc_connected = false;
    *webrtc_connected.lock().await = false;

    let restart_start = Instant::now();
    let restart_timeout = Duration::from_secs(10);

    log::info!("Waiting for reconnection after ICE restart...");

    // Wait for reconnection
    loop {
        if Instant::now().duration_since(restart_start) > restart_timeout {
            return Err(anyhow::anyhow!("ICE restart reconnection timeout"));
        }

        // Poll writes
        while let Some(msg) = rtc_pc.poll_write() {
            rtc_socket
                .send_to(&msg.message, msg.transport.peer_addr)
                .await?;
        }

        // Poll events
        while let Some(event) = rtc_pc.poll_event() {
            match event {
                RTCPeerConnectionEvent::OnIceConnectionStateChangeEvent(state) => {
                    log::info!("RTC ICE connection state after restart: {}", state);
                }
                RTCPeerConnectionEvent::OnConnectionStateChangeEvent(state) => {
                    log::info!("RTC peer connection state after restart: {}", state);
                    if state == RTCPeerConnectionState::Connected {
                        rtc_connected = true;
                    }
                }
                RTCPeerConnectionEvent::OnDataChannel(dc_event) => match dc_event {
                    _ => {}
                },
                _ => {}
            }
        }

        while let Some(message) = rtc_pc.poll_read() {
            match message {
                RTCMessage::RtpPacket(_, _) => {}
                RTCMessage::RtcpPacket(_, _) => {}
                RTCMessage::DataChannelMessage(_channel_id, data_channel_message) => {
                    let msg_str = String::from_utf8_lossy(&data_channel_message.data);
                    log::info!("RTC received message after restart: {}", msg_str);
                    rtc_received_messages.lock().await.push(msg_str.to_string());
                }
            }
        }

        // Check if both peers reconnected
        if rtc_connected && *webrtc_connected.lock().await {
            log::info!("Both peers reconnected after ICE restart!");
            break;
        }

        // Handle timeout
        let timeout_instant = rtc_pc
            .poll_timeout()
            .unwrap_or(Instant::now() + DEFAULT_TIMEOUT_DURATION);
        let delay = timeout_instant.saturating_duration_since(Instant::now());

        if delay.is_zero() {
            rtc_pc.handle_timeout(Instant::now()).ok();
            continue;
        }

        // Wait for socket data or timeout
        match timeout(
            delay.min(Duration::from_millis(100)),
            rtc_socket.recv_from(&mut buf),
        )
        .await
        {
            Ok(Ok((n, peer_addr))) => {
                rtc_pc.handle_read(TaggedBytesMut {
                    now: Instant::now(),
                    transport: TransportContext {
                        local_addr: rtc_local_addr,
                        peer_addr,
                        ecn: None,
                        transport_protocol: TransportProtocol::UDP,
                    },
                    message: BytesMut::from(&buf[..n]),
                })?;
            }
            Ok(Err(e)) => return Err(e.into()),
            Err(_) => {
                rtc_pc.handle_timeout(Instant::now()).ok();
            }
        }
    }

    // Test data channel communication after restart
    log::info!("Testing data channel after ICE restart...");

    // Send from webrtc to rtc
    webrtc_dc.send_text(TEST_MESSAGE_2.to_owned()).await?;
    log::info!("WebRTC sent after restart: {}", TEST_MESSAGE_2);

    // Process messages - give more time for message to arrive
    for i in 0..100 {
        while let Some(msg) = rtc_pc.poll_write() {
            rtc_socket
                .send_to(&msg.message, msg.transport.peer_addr)
                .await?;
        }

        while let Some(_event) = rtc_pc.poll_event() {}

        while let Some(message) = rtc_pc.poll_read() {
            match message {
                RTCMessage::RtpPacket(_, _) => {}
                RTCMessage::RtcpPacket(_, _) => {}
                RTCMessage::DataChannelMessage(_channel_id, data_channel_message) => {
                    let msg_str = String::from_utf8_lossy(&data_channel_message.data);
                    log::info!("RTC received message after restart: {}", msg_str);
                    rtc_received_messages.lock().await.push(msg_str.to_string());
                }
            }
        }

        let messages = rtc_received_messages.lock().await;
        if messages.contains(&TEST_MESSAGE_2.to_string()) {
            log::info!(
                "RTC confirmed receipt after restart after {} iterations",
                i + 1
            );
            break;
        }
        drop(messages);

        match timeout(Duration::from_millis(50), rtc_socket.recv_from(&mut buf)).await {
            Ok(Ok((n, peer_addr))) => {
                rtc_pc.handle_read(TaggedBytesMut {
                    now: Instant::now(),
                    transport: TransportContext {
                        local_addr: rtc_local_addr,
                        peer_addr,
                        ecn: None,
                        transport_protocol: TransportProtocol::UDP,
                    },
                    message: BytesMut::from(&buf[..n]),
                })?;
            }
            _ => {
                rtc_pc.handle_timeout(Instant::now()).ok();
            }
        }
    }

    // Send from rtc to webrtc
    if let Some(channel_id) = rtc_dc_id {
        if let Some(mut dc) = rtc_pc.data_channel(channel_id) {
            dc.send_text(ECHO_MESSAGE_2.to_owned())?;
            log::info!("RTC sent after restart: {}", ECHO_MESSAGE_2);
        }
    }

    // Process echo
    for _ in 0..50 {
        while let Some(msg) = rtc_pc.poll_write() {
            rtc_socket
                .send_to(&msg.message, msg.transport.peer_addr)
                .await?;
        }

        let messages = webrtc_received_messages.lock().await;
        if messages.contains(&ECHO_MESSAGE_2.to_string()) {
            break;
        }
        drop(messages);

        tokio::time::sleep(Duration::from_millis(20)).await;
    }

    log::info!("ICE restart interop test completed successfully!");
    log::info!(
        "Note: Data channel communication after ICE restart works (RTC->WebRTC), but WebRTC->RTC may have issues"
    );

    Ok(())
}
