gopher64-netplay/src/netplay.rs

use crate::device;
use crate::savestates;
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,
    /// Session-resume (issue #4): exchange the saved session state with all
    /// peers before the frame loop starts. Host-only payload: the
    /// `.netplay-state` file to broadcast.
    pub resume: bool,
    pub state_path: Option<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,
    /// Session-resume (issue #4): whether the session-state exchange runs
    /// before the frame loop.
    pub resume: bool,
    /// Session-resume (issue #4), host only: the `.netplay-state` file to
    /// broadcast to peers.
    pub state_path: Option<std::path::PathBuf>,
    /// Session-resume (issue #4): the GGRS frame the resumed session was
    /// saved at, recorded by both sides once the state has been sent
    /// (host) or loaded (peer). Issue #5's frame reset will consume this.
    pub resume_frame: Option<i32>,
}

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

/// Reliable-channel message name for a full session state transfer.
pub const NETPLAY_STATE_MESSAGE: &str = "netplay_state";
/// Reliable-channel ack name confirming a state transfer was applied.
#[allow(dead_code)] // consumed by the upcoming session save/resume lanes
pub const NETPLAY_STATE_READY_MESSAGE: &str = "netplay_state_ready";
/// Reliable-channel message name for a desync resync state transfer.
#[allow(dead_code)] // consumed by the upcoming desync resync lanes
pub const RESYNC_STATE_MESSAGE: &str = "resync_state";
/// Reliable-channel ack name confirming a resync state was applied.
#[allow(dead_code)] // consumed by the upcoming desync resync lanes
pub const RESYNC_READY_MESSAGE: &str = "resync_ready";

/// Timeout for waiting on reliable-channel messages.
const MESSAGE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);

/// Serialize a message and split it into 16 KiB transport chunks.
fn encode_message_chunks(name: &str, data: &[u8]) -> Vec<Vec<u8>> {
    let message = NetplayMessage {
        name: name.to_string(),
        data: data.to_vec(),
    };
    let encoded = postcard::to_stdvec(&message).unwrap();
    encoded.chunks(16384).map(|chunk| chunk.to_vec()).collect()
}

/// Decode a complete message from accumulated incoming bytes, returning
/// `None` while the message is still arriving in chunks.
fn decode_message(buffer: &[u8]) -> Option<(String, Vec<u8>)> {
    postcard::from_bytes::<NetplayMessage>(buffer)
        .ok()
        .map(|message| (message.name, message.data))
}

/// Accumulate received packets into `incoming_message` until complete messages
/// parse, returning every message decoded from the batch.
fn decode_packets(
    incoming_message: &mut Vec<u8>,
    received_data: &mut std::collections::VecDeque<Vec<u8>>,
    packets: Vec<Vec<u8>>,
) -> Vec<(String, Vec<u8>)> {
    let mut decoded = Vec::new();
    received_data.extend(packets);
    while !received_data.is_empty() {
        if let Some(data) = received_data.pop_front() {
            incoming_message.extend(data);
            if let Some(message) = decode_message(incoming_message) {
                decoded.push(message);
                incoming_message.clear();
            }
        }
    }
    decoded
}

fn send_message(netplay: &mut Netplay, message: NetplayMessage) {
    for chunk in encode_message_chunks(&message.name, &message.data) {
        for peer in netplay.peers.iter() {
            if let Err(e) = netplay.reliable_channel.try_send(chunk.clone().into(), *peer) {
                eprintln!("Failed to send message: {}", e);
            }
        }
    }
}

/// Drain the reliable channel and route every complete message into
/// `netplay.messages`, returning the decoded batch so callers can count
/// messages that share a name: several peers may ack within a single
/// channel poll, and the by-name `messages` map would collapse them into
/// one entry.
fn process_reliable_messages(netplay: &mut Netplay) -> Vec<(String, Vec<u8>)> {
    let packets = netplay
        .reliable_channel
        .receive()
        .iter()
        .map(|(_, data)| data.to_vec())
        .collect::<Vec<Vec<u8>>>();

    let mut decoded = Vec::new();
    for (name, data) in decode_packets(
        &mut netplay.incoming_message,
        &mut netplay.received_data,
        packets,
    ) {
        netplay.messages.insert(name.clone(), data.clone());
        check_input_delay(netplay);
        check_disconnect(netplay);
        decoded.push((name, data));
    }
    decoded
}

/// Poll `messages` for `name` until it arrives or `timeout` elapses, calling
/// `poll` on every iteration to route newly received data into `messages`.
/// Panics with the standard reliable-channel pattern when the message never
/// arrives.
fn wait_for_message(
    name: &str,
    messages: &mut std::collections::HashMap<String, Vec<u8>>,
    timeout: std::time::Duration,
    mut poll: impl FnMut(&mut std::collections::HashMap<String, Vec<u8>>),
) -> Vec<u8> {
    let now = std::time::Instant::now();

    loop {
        poll(messages);
        if let Some(data) = 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 receive_message(netplay: &mut Netplay, name: &str) -> Vec<u8> {
    let Netplay {
        messages,
        reliable_channel,
        incoming_message,
        received_data,
        input_delay,
        session,
        disconnected,
        ..
    } = netplay;

    wait_for_message(name, messages, MESSAGE_TIMEOUT, |messages| {
        let packets = reliable_channel
            .receive()
            .iter()
            .map(|(_, data)| data.to_vec())
            .collect::<Vec<Vec<u8>>>();

        for (message_name, data) in decode_packets(incoming_message, received_data, packets) {
            // Preserve the side effects the original receive path applied to
            // every decoded message (input-delay changes and disconnects).
            // These control messages are consumed, never stored in `messages`
            // (mirroring check_input_delay/check_disconnect removing them).
            if message_name == "input_delay" {
                let new_delay = u64::from_be_bytes(data[..].try_into().unwrap()) as usize;
                if new_delay != *input_delay {
                    *input_delay = new_delay;
                    for handle in session.local_player_handles() {
                        if let Err(e) = session.set_input_delay(handle, new_delay) {
                            eprintln!("Error setting input delay: {}", e);
                        } else {
                            ui::video::onscreen_message(
                                &format!("Input delay: {}", new_delay),
                                ui::video::MESSAGE_LENGTH_MESSAGE_VERY_SHORT,
                            );
                        }
                    }
                }
            } else if message_name == "disconnect" {
                if !*disconnected {
                    *disconnected = true;
                    ui::video::onscreen_message(
                        "Player disconnected, session has ended",
                        ui::video::MESSAGE_LENGTH_MESSAGE_LONG,
                    );
                }
            } else {
                messages.insert(message_name, data);
            }
        }
    })
}

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;
}

/// Send a full session state over the reliable channel as chunked messages.
#[allow(dead_code)] // consumed by the upcoming session save/resume lanes
pub fn send_netplay_state(netplay: &mut Netplay, data: &[u8]) {
    let message = NetplayMessage {
        name: NETPLAY_STATE_MESSAGE.to_string(),
        data: data.to_vec(),
    };
    send_message(netplay, message);
}

/// Receive a full session state from the reliable channel.
#[allow(dead_code)] // consumed by the upcoming session save/resume lanes
pub fn receive_netplay_state(netplay: &mut Netplay) -> Vec<u8> {
    receive_message(netplay, NETPLAY_STATE_MESSAGE)
}

/// Send a ready-ack message with `name` over the reliable channel.
#[allow(dead_code)] // consumed by the upcoming session save/resume lanes
pub fn send_ready(netplay: &mut Netplay, name: &str) {
    let message = NetplayMessage {
        name: name.to_string(),
        data: vec![],
    };
    send_message(netplay, message);
}

/// Wait for the ready-ack message with `name` (standard 10s timeout).
#[allow(dead_code)] // consumed by the upcoming session save/resume lanes
pub fn wait_for_ready(netplay: &mut Netplay, name: &str) {
    receive_message(netplay, name);
}

/// On-disk representation of a saved netplay session (host side).
/// `state` holds the compressed savestate payload produced by
/// `SavestateData::save_to_memory` (zip entries for device, saves,
/// rdp_state and ra_state); the whole struct is postcard-encoded and
/// stored zstd-compressed under a single entry inside the
/// `.netplay-state` file.
#[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct NetplaySessionSave {
    pub state: Vec<u8>,
    /// GGRS frame the session was saved at, so a resume can restart here.
    pub frame: i32,
    pub input_delay: usize,
    pub rom_hash: String,
    pub players: usize,
}

const NETPLAY_STATE_ENTRY: &str = "netplay_state";

/// Postcard-encode + compress a netplay session save into its on-disk form.
pub fn encode_netplay_session(save: &NetplaySessionSave) -> Option<Vec<u8>> {
    let data = postcard::to_stdvec(save).ok()?;
    ui::storage::compress_file(&[(&data, NETPLAY_STATE_ENTRY)]).ok()
}

/// Decompress + decode a .netplay-state file back into its session save.
/// Used by the tests today; a future resume feature consumes it.
#[cfg_attr(not(test), allow(dead_code))]
pub fn decode_netplay_session(file_data: &[u8]) -> Option<NetplaySessionSave> {
    let data = ui::storage::decompress_file(file_data, NETPLAY_STATE_ENTRY).ok()?;
    postcard::from_bytes(&data).ok()
}

fn write_atomic(path: &std::path::Path, data: &[u8]) -> std::io::Result<()> {
    let tmp_path = path.with_extension("netplay-state.tmp");
    std::fs::write(&tmp_path, data)?;
    std::fs::rename(&tmp_path, path)
}

/// Write a netplay session save to disk atomically (tmp + rename).
pub fn write_netplay_session(
    path: &std::path::Path,
    save: &NetplaySessionSave,
) -> std::io::Result<()> {
    let data = encode_netplay_session(save).ok_or_else(|| {
        std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            "failed to encode netplay session",
        )
    })?;
    write_atomic(path, &data)
}

/// Host-side session save: capture the current emulator state together with
/// the GGRS frame and session metadata into a `.netplay-state` file so the
/// session can be resumed later. Only the host (player_number 0) can save;
/// everyone else just gets an onscreen notice.
pub fn save_netplay_session(device: &mut device::Device, path: &std::path::Path) {
    let Some(netplay) = device.netplay.as_ref() else {
        return;
    };
    if netplay.player_number != 0 {
        ui::video::onscreen_message(
            "Only the host can save the session",
            ui::video::MESSAGE_LENGTH_MESSAGE_VERY_SHORT,
        );
        return;
    }

    let Some(state_data) = savestates::create_savestate_in_memory(device) else {
        eprintln!("Error capturing netplay session state");
        ui::video::onscreen_message(
            "Failed to save netplay session",
            ui::video::MESSAGE_LENGTH_MESSAGE_SHORT,
        );
        return;
    };
    // save_to_memory (issue #2 API) returns an empty Vec on serialization
    // failure rather than an Option.
    let state_zip = state_data.save_to_memory();
    if state_zip.is_empty() {
        eprintln!("Error serializing netplay session state");
        ui::video::onscreen_message(
            "Failed to save netplay session",
            ui::video::MESSAGE_LENGTH_MESSAGE_SHORT,
        );
        return;
    }

    let netplay = device.netplay.as_ref().unwrap();
    let save = NetplaySessionSave {
        state: state_zip,
        frame: netplay.session.current_frame(),
        input_delay: netplay.input_delay,
        rom_hash: device.ui.game_hash.clone(),
        players: netplay.peers.len() + 1,
    };

    match write_netplay_session(path, &save) {
        Ok(()) => ui::video::onscreen_message(
            "Session saved — safe to quit",
            ui::video::MESSAGE_LENGTH_MESSAGE_SHORT,
        ),
        Err(e) => {
            eprintln!("Error writing netplay session: {e}");
            ui::video::onscreen_message(
                "Failed to save netplay session",
                ui::video::MESSAGE_LENGTH_MESSAGE_SHORT,
            );
        }
    }
}

// ---- Session resume (issue #4) ----

/// How long the host waits for every peer to acknowledge receiving and
/// loading the broadcast session state before giving up on that peer.
const NETPLAY_STATE_ACK_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);

/// Count how many messages in a decoded batch carry `name`. Used for
/// ready-acks, where one ack per peer is expected and several peers can
/// acknowledge within a single channel poll.
fn count_messages_named(decoded: &[(String, Vec<u8>)], name: &str) -> usize {
    decoded.iter().filter(|(decoded_name, _)| decoded_name == name).count()
}

/// Wait for `expected` acknowledgements named `name`, polling the reliable
/// channel until all of them arrive or `timeout` elapses.
///
/// Acks are counted, not attributed to a specific peer: `send_ready`
/// broadcasts to every peer, so the host receives one copy per peer. The
/// count comes from the decoded batch of every poll — not from the
/// by-name `messages` map — so acks arriving together are all counted.
fn wait_for_ready_all(
    netplay: &mut Netplay,
    name: &str,
    expected: usize,
    timeout: std::time::Duration,
) -> bool {
    let now = std::time::Instant::now();
    let mut received = 0;
    while received < expected {
        received += count_messages_named(&process_reliable_messages(netplay), name);
        if received >= expected {
            return true;
        }
        if now.elapsed() > timeout {
            return false;
        }
        std::thread::sleep(std::time::Duration::from_millis(1));
    }
    true
}

/// Session-resume exchange (issue #4). Runs once, right after the initial
/// RNG/RTC/savestate handshake and before the GGRS frame loop starts:
///
/// - host (player 0): reads the `.netplay-state` file, validates its ROM
///   hash, loads the embedded savestate into its own device, broadcasts the
///   session-save bytes to every peer via `netplay_state` and waits for each
///   peer's `netplay_state_ready` ack (30 s timeout, then the lagging peers
///   are disconnected and the session is abandoned);
/// - peers: receive `netplay_state`, decode it with the same
///   `decode_netplay_session` the host used to read the file, load the
///   embedded savestate via `SavestateData::load_from_memory` +
///   `savestates::load_savestate_from_data`, then acknowledge with
///   `netplay_state_ready`.
///
/// Both sides then reset the GGRS session frame counter to the saved frame
/// with issue #5's `netplay::reset_frame`, so the frame loop resumes at the
/// saved frame instead of 0.
///
/// On any failure the session is flagged disconnected (mirroring the
/// existing peer-disconnect path) so the frame loop never starts desynced.
pub fn resume_session(device: &mut device::Device) -> Result<(), String> {
    let Some(netplay) = device.netplay.as_ref() else {
        return Ok(());
    };
    if !netplay.resume {
        return Ok(());
    }

    if netplay.player_number == 0 {
        resume_session_host(device)
    } else {
        resume_session_peer(device)
    }
}

/// Load a decoded `.netplay-state` envelope into the local device (its
/// savestate payload via `load_savestate_from_data`) and reset the GGRS
/// session frame counter to the saved frame (issue #5's `reset_frame`).
fn apply_resume_state(device: &mut device::Device, save: &NetplaySessionSave) -> Result<(), String> {
    let state_data = savestates::SavestateData::load_from_memory(&save.state)
        .ok_or_else(|| "Session state could not be deserialized".to_string())?;

    if !savestates::load_savestate_from_data(device, &state_data) {
        return Err("Session state could not be applied".to_string());
    }

    if let Some(netplay) = device.netplay.as_mut() {
        netplay.resume_frame = Some(save.frame);
        reset_frame(&mut netplay.session, save.frame);
    }
    Ok(())
}

fn resume_session_host(device: &mut device::Device) -> Result<(), String> {
    ui::video::onscreen_message(
        "Syncing session state…",
        ui::video::MESSAGE_LENGTH_MESSAGE_SHORT,
    );

    let (state_path, expected_peers, game_hash) = {
        let netplay = device.netplay.as_ref().unwrap();
        (
            netplay
                .state_path
                .clone()
                .ok_or_else(|| "Host resume requires a .netplay-state file".to_string())?,
            netplay.peers.len(),
            device.ui.game_hash.clone(),
        )
    };

    let file_data = std::fs::read(&state_path)
        .map_err(|e| format!("Could not read session state {}: {}", state_path.display(), e))?;
    let save = decode_netplay_session(&file_data)
        .ok_or_else(|| format!("Could not decode session state file {}", state_path.display()))?;

    if !save.rom_hash.is_empty() && save.rom_hash != game_hash {
        return Err(format!(
            "Session state was saved for a different ROM ({} vs {})",
            save.rom_hash, game_hash
        ));
    }

    // The host resumes at the saved state too: load the savestate into the
    // local device and reset the frame counter before the handshake, exactly
    // as every peer will.
    apply_resume_state(device, &save)?;

    // Broadcast the file bytes unmodified: `encode_netplay_session` produces
    // exactly the on-disk format, so every peer decodes what the host read.
    let broadcast = encode_netplay_session(&save)
        .ok_or_else(|| "Could not encode session state for broadcast".to_string())?;

    let netplay = device.netplay.as_mut().unwrap();
    send_netplay_state(netplay, &broadcast);

    if !wait_for_ready_all(
        netplay,
        NETPLAY_STATE_READY_MESSAGE,
        expected_peers,
        NETPLAY_STATE_ACK_TIMEOUT,
    ) {
        // One or more peers did not load the state in time. Acks are not
        // peer-attributed, so the lagging peer cannot be singled out: drop
        // the whole session rather than start it out of sync.
        for handle in 0..expected_peers + 1 {
            if handle != netplay.player_number {
                let _ = netplay.session.disconnect_player(handle);
            }
        }
        netplay.disconnected = true;
        return Err(format!(
            "Timed out waiting for {} peer(s) to sync the session state",
            expected_peers
        ));
    }

    ui::video::onscreen_message(
        "Session synced — play resumes",
        ui::video::MESSAGE_LENGTH_MESSAGE_SHORT,
    );
    Ok(())
}

fn resume_session_peer(device: &mut device::Device) -> Result<(), String> {
    ui::video::onscreen_message(
        "Syncing session state…",
        ui::video::MESSAGE_LENGTH_MESSAGE_SHORT,
    );

    let state_bytes = {
        let netplay = device.netplay.as_mut().unwrap();
        // Standard reliable-channel timeout (10 s); the host is the sender.
        receive_netplay_state(netplay)
    };

    let save = decode_netplay_session(&state_bytes)
        .ok_or_else(|| "Could not decode session state received from host".to_string())?;

    apply_resume_state(device, &save)?;

    let netplay = device.netplay.as_mut().unwrap();
    send_ready(netplay, NETPLAY_STATE_READY_MESSAGE);

    ui::video::onscreen_message(
        "Session synced — play resumes",
        ui::video::MESSAGE_LENGTH_MESSAGE_SHORT,
    );
    Ok(())
}

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,
        );
    }
}

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(netplay: &mut Netplay) {
    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",
                    ui::video::MESSAGE_LENGTH_MESSAGE_LONG,
                );
            }
        }
    }
}

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

    poll_clients(netplay);
    process_reliable_messages(netplay);
    advance_frame(device);
}

fn advance_frame(device: &mut device::Device) {
    let netplay = device.netplay.as_mut().unwrap();
    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
    {
        poll_clients(netplay);
    }

    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(),
        resume: netplay_config.resume,
        state_path: netplay_config.state_path.clone(),
        resume_frame: None,
    })
}

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);
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn state_message_constants_match_protocol() {
        assert_eq!(NETPLAY_STATE_MESSAGE, "netplay_state");
        assert_eq!(NETPLAY_STATE_READY_MESSAGE, "netplay_state_ready");
        assert_eq!(RESYNC_STATE_MESSAGE, "resync_state");
        assert_eq!(RESYNC_READY_MESSAGE, "resync_ready");
    }

    #[test]
    fn large_payload_round_trips_through_chunked_codec() {
        let payload: Vec<u8> = (0..5 * 1024 * 1024).map(|i| (i % 251) as u8).collect();

        let chunks = encode_message_chunks(NETPLAY_STATE_MESSAGE, &payload);
        assert!(chunks.len() > 1, "5MB payload must span multiple chunks");

        // Mirror what process_reliable_messages does in production: accumulate
        // chunks until the full postcard message parses.
        let mut incoming_message = Vec::new();
        let mut received_data = std::collections::VecDeque::new();
        let decoded = decode_packets(&mut incoming_message, &mut received_data, chunks);

        assert_eq!(
            decoded.len(),
            1,
            "exactly one complete message should decode after all chunks"
        );
        let (name, data) = &decoded[0];
        assert_eq!(name, NETPLAY_STATE_MESSAGE);
        assert_eq!(data, &payload);
    }

    #[test]
    fn ready_ack_send_receive_within_timeout() {
        let timeout = std::time::Duration::from_secs(10);
        let start = std::time::Instant::now();

        // Send side: chunked encoding of the ack message, exactly like send_ready.
        let chunks = encode_message_chunks(NETPLAY_STATE_READY_MESSAGE, &[]);

        // Receive side: accumulate + decode, exactly like the production path.
        let mut incoming_message = Vec::new();
        let mut received_data = std::collections::VecDeque::new();
        let decoded = decode_packets(&mut incoming_message, &mut received_data, chunks);

        assert!(
            start.elapsed() < timeout,
            "ack must arrive within the timeout window"
        );
        assert_eq!(decoded.len(), 1);
        let (name, data) = &decoded[0];
        assert_eq!(name, NETPLAY_STATE_READY_MESSAGE);
        assert!(data.is_empty());

        // The wait API itself must return promptly when the ack is already queued.
        let mut messages = std::collections::HashMap::from([(
            NETPLAY_STATE_READY_MESSAGE.to_string(),
            Vec::<u8>::new(),
        )]);
        let ack = wait_for_message(NETPLAY_STATE_READY_MESSAGE, &mut messages, timeout, |_| {});
        assert!(ack.is_empty());
        assert!(start.elapsed() < timeout);
    }

    #[test]
    fn missing_ack_times_out_with_standard_panic() {
        let mut messages = std::collections::HashMap::new();
        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            wait_for_message(
                NETPLAY_STATE_READY_MESSAGE,
                &mut messages,
                std::time::Duration::from_millis(50),
                |_| {},
            );
        }));

        let payload = result.expect_err("waiting for a missing ack must panic");
        let message = payload
            .downcast_ref::<String>()
            .map(String::as_str)
            .or_else(|| payload.downcast_ref::<&str>().copied())
            .unwrap_or("");
        assert_eq!(message, "Could not receive message for netplay_state_ready");
    }

    fn test_save(frame: i32) -> NetplaySessionSave {
        NetplaySessionSave {
            state: (0..256u32).map(|i| (i.wrapping_mul(13) % 251) as u8).collect(),
            frame,
            input_delay: 2,
            rom_hash: "12345678-90ABCDEF-C:45".to_string(),
            players: 2,
        }
    }

    #[test]
    fn netplay_session_roundtrips_through_compress_decompress() {
        let save = test_save(123_456);
        let encoded = encode_netplay_session(&save).expect("encode should succeed");
        let decoded = decode_netplay_session(&encoded).expect("decode should succeed");
        assert_eq!(decoded, save);
    }

    #[test]
    fn netplay_session_file_has_correct_metadata() {
        static COUNTER: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0);
        let save = test_save(4321);
        let path = std::env::temp_dir().join(format!(
            "gopher64-netplay-state-test-{}-{}.bin",
            std::process::id(),
            COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
        ));
        write_netplay_session(&path, &save).expect("write should succeed");

        let file_data = std::fs::read(&path).expect("file should exist on disk");
        let parsed = decode_netplay_session(&file_data).expect("file should parse");
        assert_eq!(parsed.frame, 4321);
        assert_eq!(parsed.rom_hash, "12345678-90ABCDEF-C:45");
        assert_eq!(parsed.input_delay, 2);
        assert_eq!(parsed.players, 2);
        assert_eq!(parsed, save);

        let _ = std::fs::remove_file(&path);
    }

    /// No-op network socket: satisfies ggrs' `NonBlockingSocket` bound so a
    /// local-only session can be constructed without any real network.
    struct TestSocket;

    impl ggrs::NonBlockingSocket<matchbox_socket::PeerId> for TestSocket {
        fn send_to(&mut self, _msg: &ggrs::Message, _addr: &matchbox_socket::PeerId) {}
        fn receive_all_messages(&mut self) -> Vec<(matchbox_socket::PeerId, ggrs::Message)> {
            Vec::new()
        }
    }

    /// Build a real `Netplay` value offline (mirrors the savestates test
    /// pattern): local-only ggrs session in `Running` state, WebRTC channels
    /// from a builder that never connects.
    fn mock_netplay(player_number: usize) -> Netplay {
        let (mut socket, _loop_future) = matchbox_socket::WebRtcSocketBuilder::new(
            "ws://127.0.0.1:9/room",
        )
        .add_unreliable_channel()
        .add_reliable_channel()
        .build();
        let reliable_channel = socket.take_channel(1).unwrap();

        let mut builder = ggrs::SessionBuilder::<GgrsConfig>::new()
            .with_num_players(player_number + 1)
            .unwrap();
        // GGRS requires every handle to be registered: the local player at
        // `player_number`, Remote placeholders (a dummy nil PeerId — the
        // socket never sends to it) for the rest.
        let dummy_peer: matchbox_socket::PeerId =
            serde_json::from_str("\"00000000-0000-0000-0000-000000000000\"")
                .expect("nil PeerId parses");
        for handle in 0..=player_number {
            let player_type = if handle == player_number {
                ggrs::PlayerType::Local
            } else {
                ggrs::PlayerType::Remote(dummy_peer)
            };
            builder = builder.add_player(player_type, handle).unwrap();
        }
        let session = builder.start_p2p_session(TestSocket).unwrap();

        Netplay {
            disconnected: false,
            session,
            reliable_channel,
            peers: vec![],
            player_number,
            connected: [true, false, false, false],
            input_delay: 0,
            messages: Default::default(),
            received_data: Default::default(),
            inputs: vec![],
            requests: Default::default(),
            incoming_message: vec![],
            ice_config_path: std::path::PathBuf::new(),
            resume: true,
            state_path: None,
            resume_frame: None,
        }
    }

    /// Mock 2-peer resume flow (issue #4): the host's `.netplay-state`
    /// envelope travels over the `netplay_state` reliable message, the peer
    /// decodes it and loads the embedded savestate, and both sides agree on
    /// the frame metadata. The peer's GGRS session frame counter is reset to
    /// the saved frame (issue #5's `reset_frame`) so the frame loop resumes
    /// there, not at 0.
    #[test]
    fn resume_state_reaches_peer_with_same_frame_metadata() {
        use crate::savestates::create_savestate_in_memory;

        // Host: a session with distinctive emulator state, captured in
        // memory (issue #2 API) and wrapped in the issue #3 on-disk
        // `.netplay-state` envelope.
        let mut host_device = crate::device::Device::new(false);
        host_device.cpu.gpr[1] = 0x1234_5678_9abc_def0;
        host_device.cpu.gpr[3] = 0x0fed_cba9_8765_4321;
        host_device.rdram.mem.resize(0x800, 0);
        host_device.rdram.mem[0] = 0xAB;
        host_device.rdram.mem[0x3FF] = 0xCD;
        host_device.vi.regs[0] = 0x1122_3344;

        let snapshot =
            create_savestate_in_memory(&mut host_device).expect("create in-memory savestate");
        let save = NetplaySessionSave {
            state: snapshot.save_to_memory(),
            frame: 12_345,
            input_delay: 2,
            rom_hash: "DEADBEEF-C01DF00D".to_string(),
            players: 2,
        };

        // The host broadcasts exactly the on-disk file bytes.
        let broadcast = encode_netplay_session(&save).expect("encode session save");
        let chunks = encode_message_chunks(NETPLAY_STATE_MESSAGE, &broadcast);

        // Peer accumulates the chunked message, exactly like
        // process_reliable_messages does in production.
        let mut incoming_message = Vec::new();
        let mut received_data = std::collections::VecDeque::new();
        let decoded = decode_packets(&mut incoming_message, &mut received_data, chunks);
        assert_eq!(decoded.len(), 1, "one complete netplay_state message");
        let (name, data) = &decoded[0];
        assert_eq!(name, NETPLAY_STATE_MESSAGE);

        // Peer decodes with the same envelope the host used.
        let peer_save = decode_netplay_session(data).expect("peer decodes session state");
        assert_eq!(peer_save.frame, 12_345, "both sides report the saved frame");
        assert_eq!(peer_save.input_delay, 2);
        assert_eq!(peer_save.players, 2);
        assert_eq!(peer_save.rom_hash, "DEADBEEF-C01DF00D");
        assert_eq!(peer_save.state, save.state, "state payload travels unmodified");

        // Peer loads the embedded savestate into a fresh device (no disk),
        // exactly like resume_session_peer: load_from_memory ->
        // load_savestate_from_data, then issue #5's frame reset. The RDRAM
        // backing store is pre-sized like a real emulated device, matching
        // the saved session's RDRAM.
        let mut peer_device = crate::device::Device::new(false);
        peer_device.rdram.mem.resize(0x800, 0);
        peer_device.netplay = Some(mock_netplay(1));
        apply_resume_state(&mut peer_device, &peer_save).expect("peer applies the state");

        assert_eq!(peer_device.cpu.gpr[1], 0x1234_5678_9abc_def0);
        assert_eq!(peer_device.cpu.gpr[3], 0x0fed_cba9_8765_4321);
        assert_eq!(peer_device.rdram.mem[0], 0xAB);
        assert_eq!(peer_device.rdram.mem[0x3FF], 0xCD);
        assert_eq!(peer_device.vi.regs[0], 0x1122_3344);
        assert!(
            peer_device.netplay.is_some(),
            "the live netplay session survives the state load"
        );

        // Both sides agree on the frame metadata: the resumed frame is the
        // saved frame.
        let peer_netplay = peer_device.netplay.as_ref().unwrap();
        assert_eq!(peer_netplay.resume_frame, Some(12_345));
        assert_eq!(
            peer_netplay.session.current_frame(),
            12_345,
            "GGRS frame counter resumes at the saved frame, not 0"
        );
    }

    /// Host resume path (issue #4): a `.netplay-state` file on disk is
    /// decoded by `resume_session_host`, its savestate loaded into the
    /// host's own device, and the GGRS frame counter reset to the saved
    /// frame before broadcast.
    #[test]
    fn resume_host_loads_state_from_disk_and_resets_frame() {
        use crate::device::Device;
        use crate::savestates::create_savestate_in_memory;

        let mut host_device = Device::new(false);
        host_device.cpu.gpr[5] = 0xDEAD_BEEF;
        host_device.rdram.mem.resize(0x800, 0);
        host_device.rdram.mem[0x100] = 0x42;
        host_device.netplay = Some(mock_netplay(0));
        host_device.ui.game_hash = "DEADBEEF-C01DF00D".to_string();

        let snapshot =
            create_savestate_in_memory(&mut host_device).expect("create in-memory savestate");
        let save = NetplaySessionSave {
            state: snapshot.save_to_memory(),
            frame: 77_007,
            input_delay: 2,
            rom_hash: "DEADBEEF-C01DF00D".to_string(),
            players: 2,
        };

        static COUNTER: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0);
        let path = std::env::temp_dir().join(format!(
            "gopher64-netplay-resume-test-{}-{}.netplay-state",
            std::process::id(),
            COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
        ));
        write_netplay_session(&path, &save).expect("write session save");

        // Fresh device stands in for the host's just-launched session; the
        // savestate was captured from a different Device instance above.
        // RDRAM is pre-sized like a real emulated device so the saved
        // session's RDRAM can be restored.
        let mut host_session = Device::new(false);
        host_session.rdram.mem.resize(0x800, 0);
        host_session.ui.game_hash = "DEADBEEF-C01DF00D".to_string();
        host_session.netplay = Some(mock_netplay(0));
        host_session.netplay.as_mut().unwrap().state_path = Some(path.clone());

        let frame_before = host_session.netplay.as_ref().unwrap().session.current_frame();
        resume_session_host(&mut host_session).expect("host resume succeeds");

        assert_eq!(
            host_session.netplay.as_ref().unwrap().resume_frame,
            Some(77_007)
        );
        assert_eq!(
            host_session.netplay.as_ref().unwrap().session.current_frame(),
            77_007,
            "host frame counter resumes at the saved frame (was {frame_before})"
        );
        assert_eq!(host_session.cpu.gpr[5], 0xDEAD_BEEF);
        assert_eq!(host_session.rdram.mem[0x100], 0x42);
        let _ = std::fs::remove_file(&path);
    }

    /// Ack counting is per decoded batch: several peers acknowledging
    /// within a single channel poll are all counted, and unrelated
    /// messages are ignored.
    #[test]
    fn ack_batch_counts_every_ack() {
        let batch = vec![
            (NETPLAY_STATE_READY_MESSAGE.to_string(), vec![]),
            (NETPLAY_STATE_READY_MESSAGE.to_string(), vec![]),
            ("unrelated".to_string(), vec![1]),
        ];
        assert_eq!(
            count_messages_named(&batch, NETPLAY_STATE_READY_MESSAGE),
            2,
            "both acks in one poll are counted; unrelated messages are not"
        );
    }

    /// The host's 30s ack wait times out (returning false) when acks are
    /// missing instead of hanging forever.
    #[test]
    fn wait_for_ready_all_counts_acks_and_times_out() {
        // Missing acks (the mock channel never delivers): gives up after
        // the timeout.
        let mut netplay = mock_netplay(0);
        assert!(!wait_for_ready_all(
            &mut netplay,
            NETPLAY_STATE_READY_MESSAGE,
            1,
            std::time::Duration::from_millis(50),
        ));
    }

    /// resume_session is a no-op outside resume sessions, so regular netplay
    /// is untouched (host and peer paths).
    #[test]
    fn resume_session_skips_non_resume_sessions() {
        let mut device = crate::device::Device::new(false);
        let mut netplay = mock_netplay(0);
        netplay.resume = false;
        device.netplay = Some(netplay);
        assert!(resume_session(&mut device).is_ok());
        assert_eq!(device.netplay.as_ref().unwrap().resume_frame, None);
    }

    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);
    }
}