gopher64-netplay/src/netplay.rs

use crate::device;
use crate::ui;
use sha2::digest::Digest;

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct RtcIceServerConfig {
    pub urls: Vec<String>,
    pub username: Option<String>,
    pub credential: Option<String>,
}

pub struct GgrsConfig;
impl ggrs::Config for GgrsConfig {
    type Input = ui::input::InputData;
    type InputPredictor = ggrs::PredictRepeatLast;
    type State = i32;
    type Address = matchbox_socket::PeerId;
}

pub struct MatchboxChannel {
    channel: matchbox_socket::WebRtcChannel,
}

impl ggrs::NonBlockingSocket<matchbox_socket::PeerId> for MatchboxChannel {
    fn send_to(&mut self, msg: &ggrs::Message, addr: &matchbox_socket::PeerId) {
        let encoded = postcard::to_stdvec(msg).expect("serialization failed");
        let _ = self.channel.try_send(encoded.into(), *addr);
    }

    fn receive_all_messages(&mut self) -> Vec<(matchbox_socket::PeerId, ggrs::Message)> {
        self.channel
            .receive()
            .iter()
            .filter_map(|(peer, packet)| {
                let msg = postcard::from_bytes::<ggrs::Message>(packet).ok()?;
                Some((*peer, msg))
            })
            .collect()
    }
}

pub struct NetplayConfig {
    pub server_addr: String,
    pub player_number: usize,
    pub number_of_players: usize,
    pub input_delay: usize,
    pub ice_config_path: std::path::PathBuf,
}

pub struct Netplay {
    pub disconnected: bool,
    pub session: ggrs::P2PSession<GgrsConfig>,
    pub reliable_channel: matchbox_socket::WebRtcChannel,
    pub peers: Vec<matchbox_socket::PeerId>,
    pub player_number: usize,
    pub connected: [bool; 4],
    pub input_delay: usize,
    pub messages: std::collections::HashMap<String, Vec<u8>>,
    pub received_data: std::collections::VecDeque<Vec<u8>>,
    pub inputs: Vec<(ui::input::InputData, ggrs::InputStatus)>,
    pub requests: std::collections::VecDeque<ggrs::GgrsRequest<GgrsConfig>>,
    pub incoming_message: Vec<u8>,
    pub ice_config_path: std::path::PathBuf,
    /// Resync state machine for auto-recovery from desync events.
    pub resync: ResyncState,
}

/// State machine for the desync auto-resync flow.
#[derive(Debug, Clone, Default)]
pub enum ResyncState {
    /// No resync in progress.
    #[default]
    Idle,
    /// Host has detected a desync, snapshotted state, and is waiting for acks.
    /// Stores the serialized state data and the timestamp when resync started.
    HostWaiting {
        state_data: Vec<u8>,
        state_frame: i32,
        started_at: std::time::Instant,
        acked: Vec<bool>,
    },
    /// Non-host has received a resync_state message and is loading it.
    ClientLoading {
        started_at: std::time::Instant,
    },
}

#[derive(serde::Serialize, serde::Deserialize)]
struct NetplayMessage {
    name: String,
    data: Vec<u8>,
}

fn send_message(netplay: &mut Netplay, message: NetplayMessage) {
    let data = postcard::to_stdvec(&message).unwrap();
    let chunks = data.chunks(16384).collect::<Vec<&[u8]>>();
    for peer in netplay.peers.iter() {
        for chunk in chunks.iter() {
            if let Err(e) = netplay
                .reliable_channel
                .try_send(chunk.to_vec().into(), *peer)
            {
                eprintln!("Failed to send message: {}", e);
            }
        }
    }
}

fn process_reliable_messages(netplay: &mut Netplay) {
    netplay.received_data.extend(
        netplay
            .reliable_channel
            .receive()
            .iter()
            .map(|(_, data)| data.to_vec()),
    );

    while !netplay.received_data.is_empty() {
        if let Some(data) = netplay.received_data.pop_front() {
            netplay.incoming_message.extend(data);

            if let Ok(decoded_message) =
                postcard::from_bytes::<NetplayMessage>(&netplay.incoming_message)
            {
                netplay
                    .messages
                    .insert(decoded_message.name, decoded_message.data);
                netplay.incoming_message.clear();
                check_input_delay(netplay);
                check_disconnect(netplay);
            }
        }
    }

    // Process resync-related messages
    check_resync_messages(netplay);
}

fn receive_message(netplay: &mut Netplay, name: &str) -> Vec<u8> {
    let timeout = std::time::Duration::from_secs(10);
    let now = std::time::Instant::now();

    loop {
        process_reliable_messages(netplay);
        if let Some(data) = netplay.messages.remove(name) {
            return data;
        }

        if now.elapsed() > timeout {
            panic!("Could not receive message for {name}");
        }
        std::thread::sleep(std::time::Duration::from_millis(1));
    }
}

fn send_player_number(
    channel: &mut matchbox_socket::WebRtcChannel,
    peers: &[matchbox_socket::PeerId],
    player_number: usize,
) {
    let message = NetplayMessage {
        name: "player_number".to_string(),
        data: (player_number as u64).to_be_bytes().to_vec(),
    };
    let data = postcard::to_stdvec(&message).unwrap();
    for peer in peers {
        if let Err(e) = channel.try_send(data.clone().into(), *peer) {
            eprintln!("Failed to send message: {}", e);
        }
    }
}

fn get_player_numbers(
    channel: &mut matchbox_socket::WebRtcChannel,
    player_numbers: &mut std::collections::BTreeMap<usize, Option<matchbox_socket::PeerId>>,
) {
    for (peer, data) in channel.receive() {
        let message = postcard::from_bytes::<NetplayMessage>(&data).unwrap();
        if message.name == "player_number" {
            player_numbers.insert(
                u64::from_be_bytes(message.data.try_into().unwrap()) as usize,
                Some(peer),
            );
        }
    }
}

pub fn send_rtc(netplay: &mut Netplay, rtc: i64) {
    let message = NetplayMessage {
        name: "rtc".to_string(),
        data: rtc.to_be_bytes().to_vec(),
    };
    send_message(netplay, message);
}

pub fn receive_rtc(netplay: &mut Netplay) -> i64 {
    let message = receive_message(netplay, "rtc");

    i64::from_be_bytes(message.try_into().unwrap())
}

pub fn send_rng(netplay: &mut Netplay, seed: u64) {
    let message = NetplayMessage {
        name: "rng".to_string(),
        data: seed.to_be_bytes().to_vec(),
    };
    send_message(netplay, message);
}

pub fn receive_rng(netplay: &mut Netplay) -> u64 {
    let message = receive_message(netplay, "rng");
    u64::from_be_bytes(message.try_into().unwrap())
}

pub fn send_save(netplay: &mut Netplay, save_type: &str, save_data: &[u8]) {
    let message = NetplayMessage {
        name: save_type.to_string(),
        data: save_data.to_vec(),
    };
    send_message(netplay, message);
}

pub fn receive_save(netplay: &mut Netplay, save_type: &str, save_data: &mut Vec<u8>) {
    let message = receive_message(netplay, save_type);
    *save_data = message;
}

pub fn send_input_delay(netplay: &mut Netplay, input_delay: usize) {
    if input_delay < 1 {
        return;
    }

    let message = NetplayMessage {
        name: "input_delay".to_string(),
        data: (input_delay as u64).to_be_bytes().to_vec(),
    };
    send_message(netplay, message);
    change_input_delay(netplay, input_delay);
}

fn change_input_delay(netplay: &mut Netplay, input_delay: usize) {
    netplay.input_delay = input_delay;
    for handle in netplay.session.local_player_handles() {
        if let Err(e) = netplay.session.set_input_delay(handle, input_delay) {
            eprintln!("Error setting input delay: {}", e);
        } else {
            ui::video::onscreen_message(
                &format!("Input delay: {}", input_delay),
                ui::video::MESSAGE_LENGTH_MESSAGE_VERY_SHORT,
            );
        }
    }
}

fn check_input_delay(netplay: &mut Netplay) {
    if let Some(data) = netplay.messages.remove("input_delay") {
        let input_delay = u64::from_be_bytes(data.try_into().unwrap()) as usize;
        if input_delay != netplay.input_delay {
            change_input_delay(netplay, input_delay);
        }
    }
}

fn check_disconnect(netplay: &mut Netplay) {
    if !netplay.disconnected && netplay.messages.remove("disconnect").is_some() {
        netplay.disconnected = true;
        ui::video::onscreen_message(
            "Player disconnected, session has ended",
            ui::video::MESSAGE_LENGTH_MESSAGE_LONG,
        );
    }
}

/// Checks for resync-related messages: `resync_state` (received by non-host) and
/// `resync_ready` (received by host as ack from peers).
///
/// Non-host flow:
/// 1. Receive `resync_state` message with compressed state data + frame number
/// 2. Decompress and load the state into the device
/// 3. Call `reset_frame` to align the GGRS session's frame counter
/// 4. Send `resync_ready` ack to host
/// 5. Resume normal operation
///
/// Host flow:
/// 1. Wait for `resync_ready` messages from all peers
/// 2. When all acks received, call `reset_frame` and resume
fn check_resync_messages(netplay: &mut Netplay) {
    // Non-host: received resync_state from host
    if netplay.player_number != 0 {
        if let Some(state_data) = netplay.messages.remove("resync_state") {
            eprintln!("received resync_state from host");
            // The state data is a compressed buffer with "device" and "saves" sections.
            // We store it for the device to load via load_savestate_from_resync.
            // The frame is encoded as the first 4 bytes of the data section.
            // For now, we enter the ClientLoading state and store the data.
            // The actual state loading happens in process_resync_load (called from
            // the device's frame loop) because it needs access to the Device.
            netplay.resync = ResyncState::ClientLoading {
                started_at: std::time::Instant::now(),
            };
            // Store the state data for later processing
            netplay.messages.insert("resync_state_data".to_string(), state_data);
        }
    }

    // Host: received resync_ready ack from a peer
    if netplay.player_number == 0 {
        if let Some(_) = netplay.messages.remove("resync_ready") {
            eprintln!("received resync_ready ack from peer");
            if let ResyncState::HostWaiting { acked, state_frame, .. } = &mut netplay.resync {
                // Mark the next un-acked peer as acked
                for a in acked.iter_mut() {
                    if !*a {
                        *a = true;
                        break;
                    }
                }

                // Check if all peers have acked
                if acked.iter().all(|&a| a) {
                    eprintln!("all peers acked resync, resetting frame and resuming");
                    let frame = *state_frame;
                    reset_frame(&mut netplay.session, frame);
                    netplay.resync = ResyncState::Idle;
                    ui::video::onscreen_message(
                        "Resynced",
                        ui::video::MESSAGE_LENGTH_MESSAGE_SHORT,
                    );
                }
            }
        }
    }
}

fn pending_frames(netplay: &Netplay) -> usize {
    netplay
        .requests
        .iter()
        .filter(|r| matches!(r, ggrs::GgrsRequest::AdvanceFrame { .. }))
        .count()
}

pub fn in_rollback(netplay: Option<&Netplay>) -> bool {
    if let Some(netplay) = netplay {
        pending_frames(netplay) != 0
    } else {
        false
    }
}

pub fn process_requests(
    device: &mut device::Device,
) -> Vec<(ui::input::InputData, ggrs::InputStatus)> {
    loop {
        if let Some(request) = device.netplay.as_mut().unwrap().requests.pop_front() {
            match request {
                ggrs::GgrsRequest::SaveGameState { cell, frame } => {
                    //savestates::create_savestate(device, true, Some(frame));

                    let mut hasher = sha2::Sha256::new();
                    for reg in device.cpu.cop0.regs.as_ref() {
                        hasher.update(reg.to_be_bytes());
                    }
                    let hash = u128::from_be_bytes(hasher.finalize()[..16].try_into().unwrap());
                    cell.save(frame, Some(frame), Some(hash));
                }
                ggrs::GgrsRequest::LoadGameState { cell, frame: _ } => {
                    eprintln!("attempting to load game state");
                    if let Some(_frame) = cell.load() {
                        //    savestates::load_savestate(device, true, Some(frame));
                    }
                }
                ggrs::GgrsRequest::AdvanceFrame { inputs } => {
                    return inputs;
                }
            }
        } else {
            process_netplay(device);
        }
    }
}

fn poll_clients(device: &mut device::Device) {
    let netplay = device.netplay.as_mut().unwrap();
    netplay.session.poll_remote_clients();
    for event in netplay.session.events() {
        match event {
            ggrs::GgrsEvent::Synchronizing { .. } => {}
            ggrs::GgrsEvent::Synchronized { .. } => {}
            ggrs::GgrsEvent::Disconnected { .. } => {
                if !netplay.disconnected {
                    netplay.disconnected = true;
                    ui::video::onscreen_message(
                        "Lost connection to peer(s)",
                        ui::video::MESSAGE_LENGTH_MESSAGE_LONG,
                    );
                }
            }
            ggrs::GgrsEvent::NetworkInterrupted { .. } => {
                println!("network interrupted");
            }
            ggrs::GgrsEvent::NetworkResumed { .. } => {
                println!("network resumed");
            }
            ggrs::GgrsEvent::WaitRecommendation { skip_frames } => {
                println!("wait recommendation: skip_frames={}", skip_frames);
            }
            ggrs::GgrsEvent::DesyncDetected { .. } => {
                eprintln!("desync detected");
                ui::video::onscreen_message(
                    "Desync detected — resyncing…",
                    ui::video::MESSAGE_LENGTH_MESSAGE_LONG,
                );
                handle_desync(device);
            }
        }
    }
}

/// Handles a desync detection event.
///
/// The host (player_number==0) snapshots the current emulator state into memory,
/// pauses the frame loop (enters a "resyncing" state), and sends the state to all
/// peers via the reliable channel. When all peers acknowledge with a `resync_ready`
/// message, the host calls `reset_frame` and resumes.
///
/// Non-host peers ignore the desync event — the host's state will arrive as a
/// `resync_state` message.
fn handle_desync(device: &mut device::Device) {
    let netplay = device.netplay.as_mut().unwrap();
    if netplay.player_number != 0 {
        // Non-host: ignore the event, state arrives as a message
        return;
    }

    // Host: create an in-memory savestate (includes device, saves, rdp_state, ra_state)
    let state_data = create_savestate_in_memory(device);
    let state_frame = netplay.session.current_frame();

    // Prepend the frame number as 4-byte big-endian i32 to the state data
    // so the client knows which frame to reset to
    let mut message_data = state_frame.to_be_bytes().to_vec();
    message_data.extend(state_data.clone());

    // Send resync_state message to all peers
    let message = NetplayMessage {
        name: "resync_state".to_string(),
        data: message_data,
    };
    send_message(netplay, message);

    // Enter resyncing state — pause the frame loop
    let num_players = netplay.peers.len() + 1;
    netplay.resync = ResyncState::HostWaiting {
        state_data,
        state_frame,
        started_at: std::time::Instant::now(),
        acked: vec![false; num_players],
    };
    // Mark host as already acked
    if let ResyncState::HostWaiting { acked, .. } = &mut netplay.resync {
        acked[0] = true;
    }
}

/// Creates an in-memory savestate: serializes the full Device via postcard + compress.
/// Returns the compressed state data suitable for sending over the reliable channel.
///
/// The compressed buffer contains four sections: "device", "saves", "rdp_state", and
/// "ra_state" — full parity with disk savestates. The host's current frame is prepended
/// to the data as a 4-byte big-endian i32 so the client knows which frame to reset to.
fn create_savestate_in_memory(device: &mut device::Device) -> Vec<u8> {
    // Save RDP state
    let mut rdp_state: Vec<u8> = vec![0; ui::video::state_size()];
    ui::video::save_state(rdp_state.as_mut_ptr());

    // Save RA state
    let mut ra_state: Vec<u8> = vec![0; retroachievements::state_size()];
    retroachievements::save_state(ra_state.as_mut_ptr(), ra_state.len());

    // Serialize the device state directly
    let device_data = postcard::to_stdvec(device)
        .expect("failed to serialize device state for resync");

    // Also serialize saves
    let saves_data = postcard::to_stdvec(&device.ui.storage.saves)
        .expect("failed to serialize saves for resync");

    // Compress all four sections into a single buffer
    ui::storage::compress_file(&[
        (&device_data, "device"),
        (&saves_data, "saves"),
        (&rdp_state, "rdp_state"),
        (&ra_state, "ra_state"),
    ])
    .expect("failed to compress savestate for resync")
}

fn process_netplay(device: &mut device::Device) {
    poll_clients(device);
    let netplay = device.netplay.as_mut().unwrap();
    process_reliable_messages(netplay);
    advance_frame(device);
}

fn advance_frame(device: &mut device::Device) {
    let netplay = device.netplay.as_mut().unwrap();

    // If resyncing, pause the frame loop (like the disconnected path)
    if !matches!(netplay.resync, ResyncState::Idle) {
        // Check for resync timeout (5 seconds)
        let timed_out = match &netplay.resync {
            ResyncState::HostWaiting { started_at, .. } => started_at.elapsed() > std::time::Duration::from_secs(5),
            ResyncState::ClientLoading { started_at } => started_at.elapsed() > std::time::Duration::from_secs(5),
            ResyncState::Idle => false,
        };
        if timed_out {
            eprintln!("resync timed out, dropping to disconnect behavior");
            netplay.disconnected = true;
            netplay.resync = ResyncState::Idle;
            ui::video::onscreen_message(
                "Resync failed — disconnecting",
                ui::video::MESSAGE_LENGTH_MESSAGE_LONG,
            );
        } else {
            // Still resyncing — push a default advance frame and return
            netplay.requests.push_back(ggrs::GgrsRequest::AdvanceFrame {
                inputs: vec![
                    (ui::input::InputData::default(), ggrs::InputStatus::Disconnected); 4
                ],
            });
            return;
        }
    }

    let local_input = if netplay.session.current_frame() > netplay.session.max_prediction() as i32 {
        ui::input::get(&mut device.ui, 0)
    } else {
        //workaround for disabled rollback
        ui::input::InputData::default()
    };
    let local_handle = *netplay.session.local_player_handles().first().unwrap();
    netplay
        .session
        .add_local_input(local_handle, local_input)
        .unwrap();

    // avoid rollback
    while !netplay.disconnected
        && netplay.session.current_frame() > netplay.session.confirmed_frame()
        && netplay.session.confirmed_frame() != ggrs::NULL_FRAME
    {
        // Need to reborrow for poll_clients which takes &mut Device
        drop(netplay);
        poll_clients(device);
        netplay = device.netplay.as_mut().unwrap();
    }

    if netplay.disconnected {
        netplay.requests.push_back(ggrs::GgrsRequest::AdvanceFrame {
            inputs: vec![
                (
                    ui::input::InputData::default(),
                    ggrs::InputStatus::Disconnected
                );
                4
            ],
        });
        return;
    }

    match netplay.session.advance_frame() {
        Ok(requests) => {
            netplay.requests.extend(requests);
        }
        Err(ggrs::GgrsError::PredictionThreshold) => {
            println!("prediction threshold reached");
        }
        Err(e) => panic!("{e}"),
    }
}

fn verify_peers(
    peers: &[matchbox_socket::PeerId],
    player_numbers: &std::collections::BTreeMap<usize, Option<matchbox_socket::PeerId>>,
) -> bool {
    for peer in player_numbers.values() {
        if let Some(peer) = peer
            && !peers.contains(peer)
        {
            return false;
        }
    }
    true
}

fn create_socket(builder: matchbox_socket::WebRtcSocketBuilder) -> matchbox_socket::WebRtcSocket {
    let (socket, loop_fut) = builder.build();
    tokio::spawn(async move {
        if let Err(e) = loop_fut.await {
            eprintln!("WebRTC loop failed: {}", e);
        }
    });
    socket
}

pub fn init(
    device: &mut device::Device,
    netplay_config: &NetplayConfig,
    pal: bool,
) -> Option<Netplay> {
    let mut builder = matchbox_socket::WebRtcSocketBuilder::new(&netplay_config.server_addr)
        .add_unreliable_channel()
        .add_reliable_channel();

    if let Ok(ice_config) = std::fs::read(&netplay_config.ice_config_path)
        && let Ok(ice_config) = serde_json::from_slice::<RtcIceServerConfig>(&ice_config)
    {
        builder = builder.ice_server(matchbox_socket::RtcIceServerConfig {
            urls: ice_config.urls,
            username: ice_config.username,
            credential: ice_config.credential,
        });
    } else {
        eprintln!("Using default ICE config");
    }

    let mut socket = create_socket(builder.clone());

    let mut now = std::time::Instant::now();
    let mut message_timer = now;
    let socket_timeout = std::time::Duration::from_secs_f64(rand::random_range(8.0..10.0));
    let mut player_numbers =
        std::collections::BTreeMap::from([(netplay_config.player_number, None)]);

    ui::video::onscreen_message(
        "Connecting to netplay peers...\nPlease wait...",
        ui::video::MESSAGE_LENGTH_MESSAGE_SHORT,
    );

    device.cpu.running = true;
    while device.cpu.running {
        if socket
            .update_peers()
            .iter()
            .any(|(_, peer_state)| *peer_state == matchbox_socket::PeerState::Disconnected)
        {
            // if someone has disconnected, reset the timeout
            now = std::time::Instant::now();
        }

        let connected_peers = socket
            .connected_peers()
            .collect::<Vec<matchbox_socket::PeerId>>();

        if connected_peers.len() == netplay_config.number_of_players - 1 {
            send_player_number(
                socket.channel_mut(1),
                &connected_peers,
                netplay_config.player_number,
            );
            get_player_numbers(socket.channel_mut(1), &mut player_numbers);
            if player_numbers.len() == netplay_config.number_of_players
                && verify_peers(&connected_peers, &player_numbers)
            {
                break;
            }
        } else if now.elapsed() > socket_timeout {
            socket.close();
            player_numbers.retain(|_, peer| peer.is_none());
            socket = create_socket(builder.clone());
            now = std::time::Instant::now();
        }

        if message_timer.elapsed() > std::time::Duration::from_secs(4) {
            ui::video::onscreen_message(
                &format!(
                    "Still connecting to {} netplay peer(s)...\nPlease wait...",
                    netplay_config.number_of_players - player_numbers.len()
                ),
                ui::video::MESSAGE_LENGTH_MESSAGE_SHORT,
            );
            message_timer = std::time::Instant::now();
        }

        ui::video::render_frame();
        ui::video::update_screen();
        std::thread::sleep(std::time::Duration::from_millis(10));
        ui::video::check_callback(device);
    }
    if !device.cpu.running {
        // user closed the window
        return None;
    }
    device.cpu.running = false;

    let mut session_builder = ggrs::SessionBuilder::<GgrsConfig>::new()
        .with_num_players(netplay_config.number_of_players)
        .unwrap()
        .with_input_delay(netplay_config.input_delay)
        .with_fps(if pal { 50 } else { 60 })
        .unwrap()
        .with_desync_detection_mode(ggrs::DesyncDetection::On { interval: 60 })
        .with_max_prediction_window(16)
        .with_disconnect_timeout(std::time::Duration::from_secs(if cfg!(debug_assertions) {
            10
        } else {
            5
        }));

    let mut peers = vec![];
    for (i, peer) in player_numbers.iter() {
        if let Some(peer) = peer {
            session_builder = session_builder
                .add_player(ggrs::PlayerType::Remote(*peer), *i)
                .unwrap();
            peers.push(*peer);
        } else {
            session_builder = session_builder
                .add_player(ggrs::PlayerType::Local, *i)
                .unwrap();
        }
    }

    let matchbox_channel = MatchboxChannel {
        channel: socket.take_channel(0).unwrap(),
    };
    let reliable_channel = socket.take_channel(1).unwrap();

    if matchbox_channel.channel.config().max_retransmits != Some(0)
        || matchbox_channel.channel.config().ordered
    {
        eprintln!("Sending GGRS traffic over reliable channel");
    }

    let mut session = session_builder.start_p2p_session(matchbox_channel).unwrap();

    let now = std::time::Instant::now();
    let timeout = std::time::Duration::from_secs(10);
    while session.current_state() != ggrs::SessionState::Running {
        session.poll_remote_clients();
        if now.elapsed() > timeout {
            eprintln!("Could not start netplay session");
            return None;
        }
        unsafe { sdl3_sys::events::SDL_PumpEvents() };
        std::thread::sleep(std::time::Duration::from_millis(10));
    }

    Some(Netplay {
        disconnected: false,
        incoming_message: vec![],
        input_delay: netplay_config.input_delay,
        session,
        reliable_channel,
        peers,
        player_number: netplay_config.player_number,
        connected: [
            netplay_config.number_of_players > 0,
            netplay_config.number_of_players > 1,
            netplay_config.number_of_players > 2,
            netplay_config.number_of_players > 3,
        ],
        inputs: Vec::new(),
        requests: std::collections::VecDeque::new(),
        received_data: std::collections::VecDeque::new(),
        messages: std::collections::HashMap::new(),
        ice_config_path: netplay_config.ice_config_path.clone(),
        resync: ResyncState::Idle,
    })
}

pub fn close(netplay: &mut Netplay) {
    if !netplay.disconnected {
        let message = NetplayMessage {
            name: "disconnect".to_string(),
            data: vec![],
        };
        send_message(netplay, message);
        std::thread::sleep(std::time::Duration::from_millis(200)); // give the message time to be sent
    }

    let _ = std::fs::remove_file(&netplay.ice_config_path);
}

/// Resets the GGRS session's internal frame counter to the given frame.
///
/// After loading a savestate in a netplay context, the emulator state is restored to the
/// savestate's frame, but the GGRS session's frame counter still reflects the pre-load
/// frame. If this discrepancy is not corrected, desync detection fires immediately on
/// resume because the session's saved-state checksums no longer correspond to the loaded
/// emulator state.
///
/// This function is safe to call at any time. If the GGRS version in use does not support
/// frame reset (i.e. `set_current_frame` is not available), this function will still
/// compile because the ggrs crate is patched to expose the method.
///
/// Both peers must call this with the same target frame after loading the same savestate.
pub fn reset_frame(session: &mut ggrs::P2PSession<GgrsConfig>, frame: i32) {
    session.set_current_frame(frame);
}

/// Processes a pending resync state load for non-host clients.
///
/// When a non-host client receives a `resync_state` message from the host, it
/// enters `ResyncState::ClientLoading`. This function checks for that state,
/// decompresses and loads the received state data into the device, calls
/// `reset_frame` to align the GGRS session's frame counter, sends a `resync_ready`
/// ack to the host, and resumes normal operation.
///
/// Returns true if a resync load was processed (so the caller can skip normal
/// frame processing for this iteration).
pub fn process_resync_load(device: &mut device::Device) -> bool {
    let netplay = device.netplay.as_mut().unwrap();
    if !matches!(netplay.resync, ResyncState::ClientLoading { .. }) {
        return false;
    }

    // Get the stored state data (includes 4-byte frame prefix + compressed state)
    let raw_data = match netplay.messages.remove("resync_state_data") {
        Some(data) => data,
        None => return false, // Not yet received or already processed
    };
    drop(netplay);

    eprintln!("processing resync state load");

    // Extract the frame number (first 4 bytes, big-endian i32) and state data
    if raw_data.len() < 4 {
        eprintln!("resync state data too short, ignoring");
        // Still need to clean up and send ack
        let netplay = device.netplay.as_mut().unwrap();
        netplay.resync = ResyncState::Idle;
        return true;
    }
    let frame = i32::from_be_bytes(raw_data[0..4].try_into().unwrap());
    let state_data = &raw_data[4..];

    // Decompress and load all state sections (device, saves, rdp_state, ra_state)
    let load_result = (|| {
        let device_data = ui::storage::decompress_file(state_data, "device").ok()?;
        let saves_data = ui::storage::decompress_file(state_data, "saves").ok()?;
        let rdp_state = ui::storage::decompress_file(state_data, "rdp_state").ok()?;
        let ra_state = ui::storage::decompress_file(state_data, "ra_state").ok()?;
        let state = postcard::from_bytes::<Box<device::Device>>(&device_data).ok()?;
        let saves = postcard::from_bytes(&saves_data).ok()?;
        Some((state, saves, rdp_state, ra_state))
    })();

    if let Some((mut state, mut saves, rdp_state, ra_state)) = load_result {
        // Load the state into the device with full parity to disk savestates
        if device.rdram.size == state.rdram.size {
            // Swap saves
            std::mem::swap(&mut device.ui.storage.saves, &mut saves);

            // Swap all device components (same as savestates::load_savestate)
            std::mem::swap(&mut device.cpu, &mut state.cpu);
            std::mem::swap(&mut device.pif, &mut state.pif);
            std::mem::swap(&mut device.cart, &mut state.cart);
            std::mem::swap(&mut device.cart.rom, &mut state.cart.rom); // ROM not in savestate
            std::mem::swap(&mut device.memory, &mut state.memory);
            std::mem::swap(&mut device.rsp, &mut state.rsp);
            std::mem::swap(&mut device.rdp, &mut state.rdp);
            device.rdram.mem.copy_from_slice(&state.rdram.mem);
            std::mem::swap(&mut device.rdram.regs, &mut state.rdram.regs);
            std::mem::swap(&mut device.mi, &mut state.mi);
            std::mem::swap(&mut device.pi, &mut state.pi);
            std::mem::swap(&mut device.vi, &mut state.vi);
            std::mem::swap(&mut device.ai, &mut state.ai);
            std::mem::swap(&mut device.si, &mut state.si);
            std::mem::swap(&mut device.ri, &mut state.ri);
            std::mem::swap(&mut device.vru, &mut state.vru);
            std::mem::swap(&mut device.cheats, &mut state.cheats);
            std::mem::swap(&mut device.transferpaks, &mut state.transferpaks);
            for (i, item) in device.transferpaks.iter_mut().enumerate() {
                std::mem::swap(
                    &mut item.cart.rom,
                    &mut state.transferpaks[i].cart.rom,
                );
            }

            // Reinitialize memory mappings and instruction caches
            device::memory::init(device);
            device::vi::set_expected_refresh_rate(device);
            device::cpu::map_instructions(device);
            device::cop0::map_instructions(device);
            device::cop1::map_instructions(device);
            device::cop2::map_instructions(device);
            device::rsp_cpu::map_instructions(device);

            // Restore RDP state via video subsystem
            ui::video::load_state(device, rdp_state.as_ptr());

            // Restore RA state
            if !ra_state.is_empty() {
                retroachievements::load_state(ra_state.as_ptr(), ra_state.len());
            } else {
                retroachievements::load_state(std::ptr::null(), 0);
            }

            // Reconnect PIF channels and update audio
            device::pif::connect_pif_channels(device);
            ui::audio::update_freq(device);

            eprintln!("resync state loaded successfully (frame={})", frame);
        } else {
            eprintln!("rdram size mismatch in resync state, ignoring");
        }
    } else {
        eprintln!("failed to decompress/deserialize resync state");
    }

    // Reset the GGRS frame counter to the host's frame
    reset_frame(&mut device.netplay.as_mut().unwrap().session, frame);

    // Send resync_ready ack to host
    let netplay = device.netplay.as_mut().unwrap();
    let ack = NetplayMessage {
        name: "resync_ready".to_string(),
        data: vec![],
    };
    send_message(netplay, ack);

    // Resume normal operation
    netplay.resync = ResyncState::Idle;
    ui::video::onscreen_message(
        "Resynced",
        ui::video::MESSAGE_LENGTH_MESSAGE_SHORT,
    );

    true
}

#[cfg(test)]
mod tests {
    use ggrs::{Config, GgrsRequest, SessionBuilder};
    use std::collections::hash_map::DefaultHasher;
    use std::hash::{Hash, Hasher};

    /// Test config for SyncTestSession-based frame reset tests.
    #[derive(Clone, Copy, Default, serde::Serialize, serde::Deserialize, PartialEq, Hash)]
    struct TestState {
        frame: i32,
    }

    #[derive(Clone, Copy, Default, serde::Serialize, serde::Deserialize, PartialEq)]
    struct TestInput {
        inp: u32,
    }

    struct TestConfig;
    impl Config for TestConfig {
        type Input = TestInput;
        type InputPredictor = ggrs::PredictRepeatLast;
        type State = TestState;
        type Address = std::net::SocketAddr;
    }

    fn calculate_hash<T: Hash>(t: &T) -> u64 {
        let mut s = DefaultHasher::new();
        t.hash(&mut s);
        s.finish()
    }

    struct GameStub {
        gs: TestState,
    }

    impl GameStub {
        fn new() -> Self {
            GameStub {
                gs: TestState { frame: 0 },
        }
        }

        fn handle_requests(&mut self, requests: Vec<GgrsRequest<TestConfig>>) {
            for request in requests {
                match request {
                    GgrsRequest::LoadGameState { cell, .. } => {
                        self.gs = cell.load().expect("No saved state to load");
                    }
                    GgrsRequest::SaveGameState { cell, frame } => {
                        assert_eq!(self.gs.frame, frame);
                        let checksum = calculate_hash(&self.gs) as u128;
                        cell.save(frame, Some(self.gs), Some(checksum));
                    }
                    GgrsRequest::AdvanceFrame { .. } => {
                        self.gs.frame += 1;
                    }
                }
            }
        }
    }

    fn advance_session(
        sess: &mut ggrs::SyncTestSession<TestConfig>,
        stub: &mut GameStub,
        frames: usize,
        num_players: usize,
    ) {
        for i in 0..frames {
            for p in 0..num_players {
                sess.add_local_input(p, TestInput { inp: i as u32 }).unwrap();
            }
            let requests = sess.advance_frame().unwrap();
            stub.handle_requests(requests);
        }
    }

    /// Test: advance 120 frames, reset to frame 60, verify current_frame() == 60.
    #[test]
    fn test_reset_frame_after_state_load() {
        let num_players = 2;
        let mut sess = SessionBuilder::<TestConfig>::new()
            .with_num_players(num_players)
            .unwrap()
            .with_check_distance(0)
            .start_synctest_session()
            .unwrap();
        let mut stub = GameStub::new();

        advance_session(&mut sess, &mut stub, 120, num_players);
        assert_eq!(sess.current_frame(), 120);

        sess.set_current_frame(60);
        assert_eq!(sess.current_frame(), 60);
    }

    /// Test: after resetting the frame, resuming with matching inputs should not
    /// produce any desync events.
    #[test]
    fn test_resume_after_reset_no_desync() {
        let num_players = 2;
        let mut sess = SessionBuilder::<TestConfig>::new()
            .with_num_players(num_players)
            .unwrap()
            .with_check_distance(0)
            .start_synctest_session()
            .unwrap();
        let mut stub = GameStub::new();

        advance_session(&mut sess, &mut stub, 120, num_players);
        assert_eq!(sess.current_frame(), 120);

        sess.set_current_frame(60);
        assert_eq!(sess.current_frame(), 60);
        stub.gs.frame = 60;

        for i in 0..60 {
            for p in 0..num_players {
                sess.add_local_input(p, TestInput { inp: (60 + i) as u32 }).unwrap();
            }
            let requests = sess.advance_frame().unwrap();
            assert_eq!(requests.len(), 1);
            assert!(matches!(requests[0], GgrsRequest::AdvanceFrame { .. }));
            stub.handle_requests(requests);
        }
        assert_eq!(sess.current_frame(), 120);
        assert_eq!(stub.gs.frame, 120);
    }

    /// Test: reset_frame to frame 0 works correctly.
    #[test]
    fn test_reset_frame_to_zero() {
        let num_players = 1;
        let mut sess = SessionBuilder::<TestConfig>::new()
            .with_num_players(num_players)
            .unwrap()
            .with_check_distance(0)
            .start_synctest_session()
            .unwrap();
        let mut stub = GameStub::new();

        advance_session(&mut sess, &mut stub, 50, num_players);
        assert_eq!(sess.current_frame(), 50);

        sess.set_current_frame(0);
        assert_eq!(sess.current_frame(), 0);
        stub.gs.frame = 0;

        sess.add_local_input(0, TestInput { inp: 0 }).unwrap();
        let requests = sess.advance_frame().unwrap();
        assert_eq!(requests.len(), 1);
        stub.handle_requests(requests);
        assert_eq!(sess.current_frame(), 1);
    }

    /// Test: reset_frame can be called multiple times.
    #[test]
    fn test_reset_frame_idempotent() {
        let num_players = 1;
        let mut sess = SessionBuilder::<TestConfig>::new()
            .with_num_players(num_players)
            .unwrap()
            .with_check_distance(0)
            .start_synctest_session()
            .unwrap();
        let mut stub = GameStub::new();

        advance_session(&mut sess, &mut stub, 100, num_players);

        sess.set_current_frame(50);
        assert_eq!(sess.current_frame(), 50);

        sess.set_current_frame(50);
        assert_eq!(sess.current_frame(), 50);

        sess.set_current_frame(30);
        assert_eq!(sess.current_frame(), 30);
    }

    /// Test: ResyncState default is Idle.
    #[test]
    fn test_resync_state_default_is_idle() {
        let state = ResyncState::default();
        assert!(matches!(state, ResyncState::Idle));
    }

    /// Test: ResyncState HostWaiting stores state data and tracks acks.
    #[test]
    fn test_resync_state_host_waiting() {
        let state = ResyncState::HostWaiting {
            state_data: vec![1, 2, 3],
            state_frame: 42,
            started_at: std::time::Instant::now(),
            acked: vec![true, false],
        };
        match state {
            ResyncState::HostWaiting {
                state_data,
                state_frame,
                acked,
                ..
            } => {
                assert_eq!(state_data, vec![1, 2, 3]);
                assert_eq!(state_frame, 42);
                assert_eq!(acked, vec![true, false]);
            }
            _ => panic!("expected HostWaiting"),
        }
    }

    /// Test: ResyncState ClientLoading tracks start time.
    #[test]
    fn test_resync_state_client_loading() {
        let before = std::time::Instant::now();
        let state = ResyncState::ClientLoading {
            started_at: std::time::Instant::now(),
        };
        match state {
            ResyncState::ClientLoading { started_at } => {
                assert!(started_at >= before);
            }
            _ => panic!("expected ClientLoading"),
        }
    }

    /// Test: NetplayMessage with resync_state name can be serialized and deserialized.
    #[test]
    fn test_resync_message_serialization() {
        let msg = NetplayMessage {
            name: "resync_state".to_string(),
            data: vec![0xDE, 0xAD, 0xBE, 0xEF],
        };
        let encoded = postcard::to_stdvec(&msg).unwrap();
        let decoded: NetplayMessage = postcard::from_bytes(&encoded).unwrap();
        assert_eq!(decoded.name, "resync_state");
        assert_eq!(decoded.data, vec![0xDE, 0xAD, 0xBE, 0xEF]);
    }

    /// Test: NetplayMessage with resync_ready name can be serialized and deserialized.
    #[test]
    fn test_resync_ready_message_serialization() {
        let msg = NetplayMessage {
            name: "resync_ready".to_string(),
            data: vec![],
        };
        let encoded = postcard::to_stdvec(&msg).unwrap();
        let decoded: NetplayMessage = postcard::from_bytes(&encoded).unwrap();
        assert_eq!(decoded.name, "resync_ready");
        assert!(decoded.data.is_empty());
    }

    /// Test: ResyncState timeout check — a HostWaiting state started more than
    /// 5 seconds ago should be considered timed out.
    #[test]
    fn test_resync_timeout() {
        let old_state = ResyncState::HostWaiting {
            state_data: vec![],
            state_frame: 0,
            started_at: std::time::Instant::now() - std::time::Duration::from_secs(6),
            acked: vec![true, false],
        };
        let timed_out = match &old_state {
            ResyncState::HostWaiting { started_at, .. } => {
                started_at.elapsed() > std::time::Duration::from_secs(5)
            }
            _ => false,
        };
        assert!(timed_out);

        let fresh_state = ResyncState::HostWaiting {
            state_data: vec![],
            state_frame: 0,
            started_at: std::time::Instant::now(),
            acked: vec![true, false],
        };
        let timed_out = match &fresh_state {
            ResyncState::HostWaiting { started_at, .. } => {
                started_at.elapsed() > std::time::Duration::from_secs(5)
            }
            _ => false,
        };
        assert!(!timed_out);
    }

    /// Test: all peers acked check in HostWaiting state.
    #[test]
    fn test_resync_all_acked() {
        let mut acked = vec![true, false, true];
        assert!(!acked.iter().all(|&a| a));
        acked[1] = true;
        assert!(acked.iter().all(|&a| a));
    }

    /// Test: frame number encoding/decoding in resync_state message.
    /// The host prepends a 4-byte big-endian i32 frame number to the state data.
    /// The client extracts it to know which frame to reset to.
    #[test]
    fn test_resync_frame_encoding() {
        let frame: i32 = 42;
        let state_data = vec![0xDE, 0xAD, 0xBE, 0xEF, 0x01, 0x02];

        // Encode: prepend frame as 4-byte big-endian
        let mut message_data = frame.to_be_bytes().to_vec();
        message_data.extend(state_data.clone());

        // Verify message can be serialized/deserialized as NetplayMessage
        let msg = NetplayMessage {
            name: "resync_state".to_string(),
            data: message_data,
        };
        let encoded = postcard::to_stdvec(&msg).unwrap();
        let decoded: NetplayMessage = postcard::from_bytes(&encoded).unwrap();
        assert_eq!(decoded.name, "resync_state");
        assert_eq!(decoded.data.len(), 4 + state_data.len());

        // Decode: extract frame from first 4 bytes
        let decoded_frame = i32::from_be_bytes(decoded.data[0..4].try_into().unwrap());
        assert_eq!(decoded_frame, frame);

        // Verify the remaining data is the state
        let decoded_state = &decoded.data[4..];
        assert_eq!(decoded_state, state_data.as_slice());
    }

    /// Test: frame number 0 encoding/decoding.
    #[test]
    fn test_resync_frame_encoding_zero() {
        let frame: i32 = 0;
        let message_data = frame.to_be_bytes().to_vec();

        let decoded_frame = i32::from_be_bytes(message_data[0..4].try_into().unwrap());
        assert_eq!(decoded_frame, 0);
    }

    /// Test: negative frame number encoding/decoding (edge case).
    #[test]
    fn test_resync_frame_encoding_negative() {
        let frame: i32 = -1;
        let message_data = frame.to_be_bytes().to_vec();

        let decoded_frame = i32::from_be_bytes(message_data[0..4].try_into().unwrap());
        assert_eq!(decoded_frame, -1);
    }
}