peerspeak/src/network/gossip.rs

use crate::network::{RoomState, NetError, PeerState, RoomEvent, PeerSpeakTicket};
use iroh::{Endpoint, EndpointAddr, EndpointId, SecretKey, Signature};
use iroh_gossip::net::Gossip;
use iroh_gossip::proto::TopicId;
use tokio::sync::mpsc;
use tokio::sync::mpsc::Receiver;
use std::sync::{Arc, Mutex};
use std::collections::{HashMap, HashSet};
use async_trait::async_trait;
use tokio_stream::StreamExt;
use serde::{Serialize, Deserialize};

/// Domain-separation tag mixed into every signed gossip payload so a signature
/// can never be lifted out of this protocol/version into another context.
use crate::protocol::GOSSIP_SIG_DOMAIN;

/// How far a payload's sender-stamped timestamp may differ from local time
/// before it's rejected as stale (replayed) or implausibly future. Bounds the
/// window in which a captured *genuine* signed payload could be replayed; within
/// it, a replay is byte-identical (the signature fixes the bytes) so the gossip
/// swarm's own in-flight de-duplication suppresses it. 2 minutes tolerates
/// reasonable cross-peer clock skew without leaving a wide replay window.
const GOSSIP_FRESHNESS_MS: u64 = 120_000;

/// A gossip message plus the authentication envelope that proves who sent it.
/// `author` is the claimed sender (an `EndpointId`, which *is* an ed25519 public
/// key); `sig` is that key's signature over [`signable_bytes`], so a forged
/// `author` can't validate (the attacker lacks the victim's secret key). `ts`
/// (sender-stamped unix-millis) is covered by the signature and gates replay
/// freshness — distinct from `GossipMessage::Chat.ts`, which is only for display.
#[derive(Serialize, Deserialize, Clone)]
pub struct GossipPayload {
    pub author: EndpointId,
    pub ts: u64,
    pub msg: GossipMessage,
    pub sig: Signature,
}

// `iroh::Signature` doesn't implement `Debug`, so derive everything else and
// hand-write a `Debug` that omits the (noisy, non-Debug) signature.
impl std::fmt::Debug for GossipPayload {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("GossipPayload")
            .field("author", &self.author)
            .field("ts", &self.ts)
            .field("msg_kind", &gossip_message_kind(&self.msg))
            .finish_non_exhaustive()
    }
}

/// Current unix time in milliseconds (0 on the impossible pre-epoch error).
fn now_millis() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_millis() as u64)
        .unwrap_or(0)
}

/// The exact bytes an ed25519 gossip signature covers: domain tag + room topic +
/// claimed author + timestamp + message. Binding the **topic** prevents replay
/// into another room; binding the **timestamp** bounds temporal replay; binding
/// the **author** means a signature only validates for the key it claims to come
/// from. Deterministic (same inputs → same bytes on both ends).
fn signable_bytes(topic: &[u8; 32], author: &EndpointId, ts: u64, msg: &GossipMessage) -> Vec<u8> {
    serde_json::to_vec(&(GOSSIP_SIG_DOMAIN, topic, author, ts, msg)).unwrap_or_default()
}

/// Build a signed payload from our secret key for the given room topic.
fn sign_gossip(secret: &SecretKey, topic: &[u8; 32], ts: u64, msg: GossipMessage) -> GossipPayload {
    let author = secret.public();
    let sig = secret.sign(&signable_bytes(topic, &author, ts, &msg));
    GossipPayload { author, ts, msg, sig }
}

/// Why a received gossip payload was rejected (logging + tests).
#[derive(Debug, PartialEq, Eq)]
enum GossipReject {
    /// Signature didn't validate for the claimed `author` — a forgery/spoof.
    BadSignature,
    /// Timestamp outside the freshness window — stale (replay) or implausibly future.
    OutOfWindow,
    /// A signed Announce advertised an address for a different node id.
    AnnounceAddressMismatch,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
enum StateMutationKind {
    Announce,
    Leave,
}

fn gossip_message_kind(msg: &GossipMessage) -> &'static str {
    match msg {
        GossipMessage::Announce(_) => "Announce",
        GossipMessage::Leave => "Leave",
        GossipMessage::Chat { .. } => "Chat",
    }
}

fn state_mutation_kind(msg: &GossipMessage) -> Option<StateMutationKind> {
    match msg {
        GossipMessage::Announce(_) => Some(StateMutationKind::Announce),
        GossipMessage::Leave => Some(StateMutationKind::Leave),
        GossipMessage::Chat { .. } => None,
    }
}

fn admit_state_mutation(
    seen: &mut HashMap<(EndpointId, StateMutationKind), u64>,
    author: EndpointId,
    msg: &GossipMessage,
    ts: u64,
) -> bool {
    let Some(kind) = state_mutation_kind(msg) else {
        return true;
    };
    let key = (author, kind);
    if seen.get(&key).is_some_and(|last_ts| ts <= *last_ts) {
        return false;
    }
    seen.insert(key, ts);
    true
}

fn peer_state_for_log(state: &PeerState) -> String {
    format!(
        "name={:?}, muted={}, addr_id={}, addrs={}, sharing={}",
        state.name,
        state.is_muted,
        crate::short_id(&state.addr.id.to_string()),
        state.addr.addrs.len(),
        state.sharing.is_some()
    )
}

/// Authenticate a received payload against the room topic and local clock. The
/// signature must validate for the claimed `author` (closing the author-spoofing
/// vector, security S2), and the timestamp must be within `window_ms` of
/// `now_ms`. Pure — no I/O — so it's directly unit-testable.
fn verify_gossip(
    payload: &GossipPayload,
    topic: &[u8; 32],
    now_ms: u64,
    window_ms: u64,
) -> Result<(), GossipReject> {
    let bytes = signable_bytes(topic, &payload.author, payload.ts, &payload.msg);
    payload
        .author
        .verify(&bytes, &payload.sig)
        .map_err(|_| GossipReject::BadSignature)?;
    if now_ms.abs_diff(payload.ts) > window_ms {
        return Err(GossipReject::OutOfWindow);
    }
    if let GossipMessage::Announce(state) = &payload.msg
        && state.addr.id != payload.author {
        return Err(GossipReject::AnnounceAddressMismatch);
    }
    Ok(())
}

/// Compute the gossip bootstrap peer set for a (re)join: the ticket's host plus
/// every peer we already knew about (`extra`), with self removed and ids
/// de-duplicated. Without the retained `extra` peers, the ROOM CREATOR rejoining
/// their own room would have an empty list (their ticket names only themselves as
/// host) and so never re-enter the swarm — that was bug A8. Pure → unit-testable.
fn compute_bootstrap(
    self_id: EndpointId,
    ticket_host: EndpointId,
    extra: &[EndpointAddr],
) -> Vec<EndpointId> {
    let mut out: Vec<EndpointId> = Vec::new();
    if ticket_host != self_id {
        out.push(ticket_host);
    }
    for addr in extra {
        if addr.id != self_id && !out.contains(&addr.id) {
            out.push(addr.id);
        }
    }
    out
}

#[derive(Serialize, Deserialize, Clone, Debug)]
pub enum GossipMessage {
    Announce(PeerState),
    Leave,
    /// A room text-chat message: the author's display name, the text, a
    /// sender-stamped millisecond timestamp, and an optional file attachment
    /// descriptor (the bytes are fetched off-gossip on the file plane).
    Chat {
        name: String,
        text: String,
        ts: u64,
        #[serde(default)]
        attachment: Option<crate::files::ChatAttachment>,
    },
}

pub struct IrohGossipState {
    _endpoint: Endpoint,
    gossip: Gossip,
    address_lookup: iroh::address_lookup::memory::MemoryLookup,
    /// Our node secret key, used to sign every outgoing gossip payload so peers
    /// can authenticate us (security S2). Same key the endpoint binds with.
    secret_key: SecretKey,
    self_state: Arc<Mutex<Option<PeerState>>>,
    peers: Arc<Mutex<HashMap<EndpointId, PeerState>>>,
    /// Previously verified peers whose live roster entry was removed by a
    /// transient disconnect. Retained only so a later authenticated `Leave`
    /// still reaches core and cancels background recovery.
    disconnected_peers: Arc<Mutex<HashSet<EndpointId>>>,
    event_tx: mpsc::Sender<RoomEvent>,
    event_rx: Mutex<Option<mpsc::Receiver<RoomEvent>>>,
    active_topic: Mutex<Option<tokio::task::JoinHandle<()>>>,
    active_topic_id: Mutex<Option<TopicId>>,
    /// Raw 32-byte id of the joined topic, retained so the out-of-task signers
    /// (`update_self_state`/`send_chat`/`leave`) can bind signatures to the room.
    active_topic_bytes: Mutex<Option<[u8; 32]>>,
    active_sender: Mutex<Option<iroh_gossip::api::GossipSender>>,
}

impl IrohGossipState {
    pub fn new(
        endpoint: Endpoint,
        gossip: Gossip,
        address_lookup: iroh::address_lookup::memory::MemoryLookup,
        secret_key: SecretKey,
    ) -> Self {
        let (event_tx, event_rx) = mpsc::channel(100);
        Self {
            _endpoint: endpoint,
            gossip,
            address_lookup,
            secret_key,
            self_state: Arc::new(Mutex::new(None)),
            peers: Arc::new(Mutex::new(HashMap::new())),
            disconnected_peers: Arc::new(Mutex::new(HashSet::new())),
            event_tx,
            event_rx: Mutex::new(Some(event_rx)),
            active_topic: Mutex::new(None),
            active_topic_id: Mutex::new(None),
            active_topic_bytes: Mutex::new(None),
            active_sender: Mutex::new(None),
        }
    }
}

#[async_trait]
impl RoomState for IrohGossipState {
    async fn join(
        &self,
        ticket_str: &str,
        self_state: PeerState,
        extra_bootstrap: Vec<EndpointAddr>,
    ) -> Result<(), NetError> {
        crate::log_msg(&format!(
            "RoomState::join: self_id={}, self_name={:?}, ticket={}",
            crate::short_id(&self_state.addr.id.to_string()),
            self_state.name,
            crate::redact_for_log(ticket_str)
        ));
        let ticket = ticket_str.parse::<PeerSpeakTicket>()?;
        // Version-namespace the subscribed topic (VERSIONING.md): peers on a
        // different gossip protocol version derive a different topic from the same
        // ticket and never share a swarm. The raw ticket.topic_id stays the room
        // identity (and what signatures bind, below).
        let topic_id = TopicId::from_bytes(crate::protocol::versioned_topic(ticket.topic_id));

        crate::log_msg(&format!(
            "Parsed ticket. host_id={}, host_addrs={}, topic={}",
            crate::short_id(&ticket.host_addr.id.to_string()),
            ticket.host_addr.addrs.len(),
            crate::short_bytes_hex(&ticket.topic_id)
        ));

        // Stop any currently running topic
        let _ = self.leave().await;

        // Make every dial target resolvable: the ticket host plus any retained
        // peers handed in (their addresses may have aged out of a fresh endpoint's
        // book even though the persistent lookup usually still holds them).
        self.address_lookup.add_endpoint_info(ticket.host_addr.clone());
        for addr in &extra_bootstrap {
            self.address_lookup.add_endpoint_info(addr.clone());
        }

        // Bootstrap to the ticket host AND every retained peer (minus self). The
        // retained peers are what let the room creator rejoin a room they left —
        // their own ticket names only themselves as host (bug A8).
        let bootstrap_peers =
            compute_bootstrap(self_state.addr.id, ticket.host_addr.id, &extra_bootstrap);
        crate::log_msg(&format!("Bootstrap peers for join: {:?}", bootstrap_peers));

        let gossip_topic = self.gossip.subscribe(topic_id, bootstrap_peers).await
            .map_err(|e| {
                let err = format!("Failed to join gossip topic: {}", e);
                crate::log_msg(&err);
                NetError::Gossip(err)
            })?;

        let (gossip_sender, mut gossip_receiver) = gossip_topic.split();

        *self.self_state.lock().unwrap() = Some(self_state.clone());
        *self.active_topic_id.lock().unwrap() = Some(topic_id);
        *self.active_topic_bytes.lock().unwrap() = Some(ticket.topic_id);
        *self.active_sender.lock().unwrap() = Some(gossip_sender.clone());

        let event_tx = self.event_tx.clone();
        let peers = self.peers.clone();
        let disconnected_peers = self.disconnected_peers.clone();
        let address_lookup = self.address_lookup.clone();
        let self_state_clone = self.self_state.clone();
        let gossip_sender_clone = gossip_sender.clone();
        let self_id = self_state.addr.id;
        // The signing key + room topic the spawned loop needs to sign our own
        // broadcasts and authenticate every inbound payload (security S2).
        let secret_key = self.secret_key.clone();
        let topic_bytes = ticket.topic_id;

        let handle = tokio::spawn(async move {
            crate::log_msg(&format!("Spawned gossip topic loop for self_id={:?}", self_id));
            let mut state_mutations_seen = HashMap::new();

            // Broadcast initial state
            let initial_payload = {
                let guard = self_state_clone.lock().unwrap();
                guard.as_ref().map(|s| sign_gossip(
                    &secret_key,
                    &topic_bytes,
                    now_millis(),
                    GossipMessage::Announce(s.clone()),
                ))
            };

            if let Some(payload) = initial_payload
                && let Ok(bytes) = serde_json::to_vec(&payload) {
                    crate::log_msg(&format!("Broadcasting initial state from self_id={:?}", self_id));
                    let _ = gossip_sender_clone.broadcast(bytes.into()).await;
                }

            // Stream topic messages
            while let Some(res) = gossip_receiver.next().await {
                match res {
                    Ok(iroh_gossip::api::Event::Received(msg)) => {
                        crate::log_msg(&format!("Gossip received Event::Received from delivery={:?}", msg.delivered_from));
                        match serde_json::from_slice::<GossipPayload>(&msg.content) {
                            Ok(payload) => {
                                // Authenticate before trusting `author` for ANY
                                // action: a forged/stale payload is dropped here
                                // so it can't impersonate, evict, or poison
                                // presence/address-book (security S2).
                                if let Err(reason) =
                                    verify_gossip(&payload, &topic_bytes, now_millis(), GOSSIP_FRESHNESS_MS)
                                {
                                    crate::log_msg(&format!(
                                        "Gossip dropped unauthenticated/stale payload claiming author={:?}: {:?}",
                                        payload.author, reason
                                    ));
                                    continue;
                                }
                                let our_id = {
                                    self_state_clone.lock().unwrap()
                                        .as_ref()
                                        .map(|s| s.addr.id)
                                };
                                
                                if Some(payload.author) == our_id {
                                    crate::log_msg("Gossip Event::Received from ourselves; ignoring");
                                    continue;
                                }

                                if !admit_state_mutation(
                                    &mut state_mutations_seen,
                                    payload.author,
                                    &payload.msg,
                                    payload.ts,
                                ) {
                                    crate::log_msg(&format!(
                                        "Gossip dropped replayed state mutation author={}, kind={}, ts={}",
                                        crate::short_id(&payload.author.to_string()),
                                        gossip_message_kind(&payload.msg),
                                        payload.ts
                                    ));
                                    continue;
                                }

                                crate::log_msg(&format!(
                                    "Gossip Event::Received author={}, kind={}",
                                    crate::short_id(&payload.author.to_string()),
                                    gossip_message_kind(&payload.msg)
                                ));

                                match payload.msg {
                                    GossipMessage::Announce(mut state) => {
                                        // Presence names are untrusted (and author-
                                        // spoofable): sanitize at ingest so every
                                        // consumer gets a safe value (security S4).
                                        state.name = crate::sanitize::sanitize_name(&state.name);
                                        // A peer's avatar is equally untrusted: a
                                        // custom image is validated (size + safe
                                        // decode within bounds) or downgraded to a
                                        // monogram, so a malformed/oversized/bomb
                                        // image can't crash or exhaust us (W4).
                                        state.avatar = state.avatar.sanitize_incoming();
                                        // Screen-share tickets are capabilities and
                                        // peer-supplied: cap/validate once at ingest
                                        // so invalid offers never render a Watch button.
                                        state.sharing = state.sharing.and_then(crate::screenshare::sanitize_ticket);
                                        disconnected_peers.lock().unwrap().remove(&payload.author);
                                        let (is_new, state_changed) = {
                                            let mut peer_map = peers.lock().unwrap();
                                            let is_new = !peer_map.contains_key(&payload.author);
                                            let state_changed = peer_map.get(&payload.author) != Some(&state);
                                            if is_new || state_changed {
                                                peer_map.insert(payload.author, state.clone());
                                            }
                                            (is_new, state_changed)
                                        };

                                        if is_new {
                                            crate::log_msg(&format!(
                                                "Gossip new peer joined: {}, state: {}",
                                                crate::short_id(&payload.author.to_string()),
                                                peer_state_for_log(&state)
                                            ));
                                            address_lookup.add_endpoint_info(state.addr.clone());
                                            let _ = event_tx.send(RoomEvent::PeerJoined(payload.author, state)).await;
                                        } else if state_changed {
                                            crate::log_msg(&format!(
                                                "Gossip peer state updated: {}, state: {}",
                                                crate::short_id(&payload.author.to_string()),
                                                peer_state_for_log(&state)
                                            ));
                                            let _ = event_tx.send(RoomEvent::PeerUpdated(payload.author, state)).await;
                                        }
                                    }
                                    GossipMessage::Leave => {
                                        crate::log_msg(&format!("Gossip peer leave request from author={:?}", payload.author));
                                        let removed = peers.lock().unwrap().remove(&payload.author).is_some();
                                        let was_disconnected = disconnected_peers
                                            .lock()
                                            .unwrap()
                                            .remove(&payload.author);
                                        if removed || was_disconnected {
                                            let _ = event_tx.send(RoomEvent::PeerLeft(payload.author)).await;
                                        }
                                    }
                                    GossipMessage::Chat { name, text, ts, attachment } => {
                                        crate::log_msg(&format!("Gossip chat from author={:?}", payload.author));
                                        // Defensively normalize an untrusted attachment
                                        // descriptor: sanitize the filename and drop it
                                        // entirely if it declares an out-of-cap size.
                                        let attachment = attachment.and_then(|mut a| {
                                            if !crate::files::size_within_cap(a.size) {
                                                return None;
                                            }
                                            a.name = crate::files::sanitize_filename(&a.name);
                                            Some(a)
                                        });
                                        let _ = event_tx.send(RoomEvent::ChatMessage {
                                            from: payload.author,
                                            name,
                                            text,
                                            ts,
                                            attachment,
                                        }).await;
                                    }
                                }
                            }
                            Err(e) => {
                                crate::log_msg(&format!("Gossip failed to deserialize payload: {:?}", e));
                            }
                        }
                    }
                    Ok(iroh_gossip::api::Event::NeighborUp(peer_id)) => {
                        crate::log_msg(&format!("Gossip event: NeighborUp={:?}", peer_id));
                        // Resend state on new neighbor connection to guarantee synchronization
                        let payload_opt = {
                            let guard = self_state_clone.lock().unwrap();
                            guard.as_ref().map(|state| sign_gossip(
                                &secret_key,
                                &topic_bytes,
                                now_millis(),
                                GossipMessage::Announce(state.clone()),
                            ))
                        };
                        if let Some(payload) = payload_opt
                            && let Ok(bytes) = serde_json::to_vec(&payload) {
                                crate::log_msg(&format!("Broadcasting state to new neighbor={:?}", peer_id));
                                let _ = gossip_sender_clone.broadcast(bytes.into()).await;
                            }
                    }
                    Ok(iroh_gossip::api::Event::NeighborDown(peer_id)) => {
                        crate::log_msg(&format!("Gossip event: NeighborDown={:?}", peer_id));
                        // A NeighborDown is a *transient* loss, not a graceful
                        // leave: emit PeerConnectionLost so the core marks the peer
                        // "reconnecting" and keeps its audio supervisor redialing,
                        // rather than tearing everything down. (Treating this as a
                        // PeerLeft is exactly what defeated reconnect in the field —
                        // it aborted the supervisor ~30s in.) We still drop our
                        // cached presence entry; a rejoin re-announces as new.
                        let removed = peers.lock().unwrap().remove(&peer_id).is_some();
                        if removed {
                            disconnected_peers.lock().unwrap().insert(peer_id);
                            crate::log_msg(&format!("Peer connection lost (NeighborDown): {:?}", peer_id));
                            let _ = event_tx.send(RoomEvent::PeerConnectionLost(peer_id)).await;
                        }
                    }
                    Ok(other) => {
                        crate::log_msg(&format!("Gossip other event: {:?}", other));
                    }
                    Err(e) => {
                        crate::log_msg(&format!("Gossip error event: {:?}", e));
                    }
                }
            }
            crate::log_msg("Gossip topic loop terminated");
        });

        *self.active_topic.lock().unwrap() = Some(handle);
        Ok(())
    }

    async fn update_self_state(&self, self_state: PeerState) -> Result<(), NetError> {
        crate::log_msg(&format!(
            "RoomState::update_self_state: state: {}",
            peer_state_for_log(&self_state)
        ));
        *self.self_state.lock().unwrap() = Some(self_state.clone());

        let sender_opt = self.active_sender.lock().unwrap().clone();
        let topic_opt = *self.active_topic_bytes.lock().unwrap();
        if let (Some(sender), Some(topic)) = (sender_opt, topic_opt) {
            let payload = sign_gossip(
                &self.secret_key,
                &topic,
                now_millis(),
                GossipMessage::Announce(self_state),
            );
            if let Ok(bytes) = serde_json::to_vec(&payload) {
                crate::log_msg("Broadcasting updated self state to gossip");
                sender.broadcast(bytes.into()).await
                    .map_err(|e| NetError::Gossip(e.to_string()))?;
            }
        }
        Ok(())
    }

    async fn rebootstrap_peers(&self, peers: Vec<EndpointAddr>) -> Result<(), NetError> {
        let self_id = self._endpoint.id();
        let mut peer_ids = Vec::new();
        for addr in peers {
            if addr.id == self_id || peer_ids.contains(&addr.id) {
                continue;
            }
            self.address_lookup.add_endpoint_info(addr.clone());
            peer_ids.push(addr.id);
        }

        if peer_ids.is_empty() {
            return Ok(());
        }

        // Clone the sender before awaiting: active_sender is a standard mutex and
        // must never be held across an async gossip operation.
        let sender = self
            .active_sender
            .lock()
            .unwrap()
            .clone()
            .ok_or_else(|| NetError::Other("Not in a room".to_string()))?;

        crate::log_msg(&format!("Rebootstrapping gossip peers: {:?}", peer_ids));
        sender
            .join_peers(peer_ids)
            .await
            .map_err(|e| NetError::Gossip(e.to_string()))
    }

    fn mark_peer_disconnected(&self, peer_id: EndpointId) {
        if self.peers.lock().unwrap().remove(&peer_id).is_some() {
            self.disconnected_peers.lock().unwrap().insert(peer_id);
        }
    }

    async fn send_chat(
        &self,
        text: String,
        attachment: Option<crate::files::ChatAttachment>,
    ) -> Result<(), NetError> {
        let name = {
            let guard = self.self_state.lock().unwrap();
            match guard.as_ref() {
                Some(s) => s.name.clone(),
                None => return Err(NetError::Other("Not in a room".to_string())),
            }
        };
        let ts = now_millis();

        let sender_opt = self.active_sender.lock().unwrap().clone();
        let topic_opt = *self.active_topic_bytes.lock().unwrap();
        if let (Some(sender), Some(topic)) = (sender_opt, topic_opt) {
            let payload = sign_gossip(
                &self.secret_key,
                &topic,
                ts,
                GossipMessage::Chat { name, text, ts, attachment },
            );
            if let Ok(bytes) = serde_json::to_vec(&payload) {
                sender.broadcast(bytes.into()).await
                    .map_err(|e| NetError::Gossip(e.to_string()))?;
            }
        }
        Ok(())
    }

    async fn leave(&self) -> Result<(), NetError> {
        crate::log_msg("RoomState::leave called");
        {
            let mut handle_guard = self.active_topic.lock().unwrap();
            if let Some(handle) = handle_guard.take() {
                crate::log_msg("Aborting active gossip topic background task");
                handle.abort();
            }
        }

        *self.active_topic_id.lock().unwrap() = None;
        let topic_opt = self.active_topic_bytes.lock().unwrap().take();

        let sender_opt = self.active_sender.lock().unwrap().take();
        if let (Some(sender), Some(topic)) = (sender_opt, topic_opt) {
            let payload = sign_gossip(
                &self.secret_key,
                &topic,
                now_millis(),
                GossipMessage::Leave,
            );
            if let Ok(bytes) = serde_json::to_vec(&payload) {
                crate::log_msg("Broadcasting Leave message to gossip");
                let _ = sender.broadcast(bytes.into()).await;
            }
        }

        self.peers.lock().unwrap().clear();
        self.disconnected_peers.lock().unwrap().clear();
        Ok(())
    }

    fn active_peers(&self) -> Vec<(EndpointId, PeerState)> {
        let guard = self.peers.lock().unwrap();
        guard.iter().map(|(k, v)| (*k, v.clone())).collect()
    }

    async fn subscribe_events(&self) -> Result<Receiver<RoomEvent>, NetError> {
        let mut rx_guard = self.event_rx.lock().unwrap();
        if let Some(rx) = rx_guard.take() {
            Ok(rx)
        } else {
            Err(NetError::Other("Events already subscribed".to_string()))
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::network::PeerState;
    use iroh::SecretKey;
    use std::collections::HashMap;

    fn sample_peer_state_for(id: EndpointId) -> PeerState {
        let addr = iroh::EndpointAddr::from(id);
        PeerState {
            name: "TestPeerGossip".to_string(),
            is_muted: true,
            addr,
            sharing: None,
            avatar: crate::avatar::Avatar::default(),
        }
    }

    fn fresh_id() -> EndpointId {
        SecretKey::generate().public()
    }

    fn addr_of(id: EndpointId) -> EndpointAddr {
        EndpointAddr::from(id)
    }

    #[test]
    fn bootstrap_client_dials_host() {
        // A non-host (client) with no retained peers dials just the ticket host.
        let me = fresh_id();
        let host = fresh_id();
        assert_eq!(compute_bootstrap(me, host, &[]), vec![host]);
    }

    #[test]
    fn bootstrap_host_rejoin_dials_retained_peers() {
        // The A8 regression: the room creator's ticket names themselves as host,
        // so the ticket-host contributes nothing — the retained peer is the only
        // (and essential) dial target. An empty result here would reproduce A8.
        let me = fresh_id();
        let peer = fresh_id();
        let bootstrap = compute_bootstrap(me, /* ticket_host = */ me, &[addr_of(peer)]);
        assert_eq!(bootstrap, vec![peer]);
    }

    #[test]
    fn bootstrap_host_rejoin_with_no_retained_peers_is_empty() {
        // Host rejoin before anyone was ever seen: nothing to dial (unavoidable),
        // but it must not contain self.
        let me = fresh_id();
        assert!(compute_bootstrap(me, me, &[]).is_empty());
    }

    #[test]
    fn bootstrap_excludes_self_and_dedups_host() {
        // Self is never dialed (even if it shows up among retained peers), and a
        // peer that equals the ticket host isn't listed twice.
        let me = fresh_id();
        let host = fresh_id();
        let other = fresh_id();
        let extra = [addr_of(me), addr_of(host), addr_of(other)];
        let bootstrap = compute_bootstrap(me, host, &extra);
        assert_eq!(bootstrap, vec![host, other]);
        assert!(!bootstrap.contains(&me));
    }

    #[test]
    fn test_gossip_message_leave_round_trip() {
        let original = GossipMessage::Leave;
        let serialized = serde_json::to_string(&original).unwrap();
        let deserialized: GossipMessage = serde_json::from_str(&serialized).unwrap();
        assert!(matches!(deserialized, GossipMessage::Leave));
    }

    #[test]
    fn test_gossip_payload_announce_round_trip() {
        let secret = SecretKey::generate();
        let topic = [9u8; 32];
        let peer_state = sample_peer_state_for(secret.public());
        let payload = sign_gossip(&secret, &topic, 1000, GossipMessage::Announce(peer_state.clone()));

        let serialized = serde_json::to_string(&payload).unwrap();
        let deserialized: GossipPayload = serde_json::from_str(&serialized).unwrap();

        // Author is derived from the signing key, and the signed payload still
        // verifies after a serde round-trip.
        assert_eq!(deserialized.author, secret.public());
        assert!(verify_gossip(&deserialized, &topic, 1000, GOSSIP_FRESHNESS_MS).is_ok());
        match deserialized.msg {
            GossipMessage::Announce(state) => {
                assert_eq!(state, peer_state);
            }
            GossipMessage::Leave | GossipMessage::Chat { .. } => {
                panic!("Expected GossipMessage::Announce");
            }
        }
    }

    #[test]
    fn test_gossip_message_chat_round_trip() {
        // Test normal chat message
        let original = GossipMessage::Chat {
            name: "Alice".to_string(),
            text: "Hello".to_string(),
            ts: 123456789,
            attachment: None,
        };
        let serialized = serde_json::to_string(&original).unwrap();
        let deserialized: GossipMessage = serde_json::from_str(&serialized).unwrap();
        if let GossipMessage::Chat { name, text, ts, attachment } = deserialized {
            assert_eq!(name, "Alice");
            assert_eq!(text, "Hello");
            assert_eq!(ts, 123456789);
            assert_eq!(attachment, None);
        } else {
            panic!("Expected GossipMessage::Chat");
        }

        // Test empty strings and large timestamp
        let original_empty = GossipMessage::Chat {
            name: "".to_string(),
            text: "".to_string(),
            ts: u64::MAX,
            attachment: None,
        };
        let serialized_empty = serde_json::to_string(&original_empty).unwrap();
        let deserialized_empty: GossipMessage = serde_json::from_str(&serialized_empty).unwrap();
        if let GossipMessage::Chat { name, text, ts, .. } = deserialized_empty {
            assert_eq!(name, "");
            assert_eq!(text, "");
            assert_eq!(ts, u64::MAX);
        } else {
            panic!("Expected GossipMessage::Chat");
        }
    }

    #[test]
    fn test_gossip_chat_attachment_round_trip_and_back_compat() {
        let att = crate::files::ChatAttachment {
            name: "photo.png".to_string(),
            size: 4096,
            kind: crate::files::AttachmentKind::Image,
            id: [42u8; 32],
        };
        let original = GossipMessage::Chat {
            name: "Alice".to_string(),
            text: "look at this".to_string(),
            ts: 1,
            attachment: Some(att.clone()),
        };
        let serialized = serde_json::to_string(&original).unwrap();
        let deserialized: GossipMessage = serde_json::from_str(&serialized).unwrap();
        if let GossipMessage::Chat { attachment, .. } = deserialized {
            assert_eq!(attachment, Some(att));
        } else {
            panic!("Expected GossipMessage::Chat");
        }

        // A pre-v2 chat payload (no `attachment` field) must still deserialize,
        // defaulting the attachment to None (serde(default)).
        let legacy = r#"{"Chat":{"name":"Old","text":"hi","ts":7}}"#;
        let parsed: GossipMessage = serde_json::from_str(legacy).unwrap();
        if let GossipMessage::Chat { name, attachment, .. } = parsed {
            assert_eq!(name, "Old");
            assert_eq!(attachment, None);
        } else {
            panic!("Expected GossipMessage::Chat");
        }
    }

    #[test]
    fn test_gossip_payload_chat_round_trip() {
        let secret = SecretKey::generate();
        let topic = [3u8; 32];
        let payload = sign_gossip(
            &secret,
            &topic,
            987654321,
            GossipMessage::Chat {
                name: "Bob".to_string(),
                text: "Hi there".to_string(),
                ts: 987654321,
                attachment: None,
            },
        );

        let serialized = serde_json::to_string(&payload).unwrap();
        let deserialized: GossipPayload = serde_json::from_str(&serialized).unwrap();

        assert_eq!(deserialized.author, secret.public());
        if let GossipMessage::Chat { name, text, ts, .. } = deserialized.msg {
            assert_eq!(name, "Bob");
            assert_eq!(text, "Hi there");
            assert_eq!(ts, 987654321);
        } else {
            panic!("Expected GossipMessage::Chat");
        }
    }

    #[test]
    fn test_gossip_chat_unicode_round_trip() {
        let original = GossipMessage::Chat {
            name: "🎙 User".to_string(),
            text: "héllo 🎙 世界".to_string(),
            ts: 1717171717,
            attachment: None,
        };
        let serialized = serde_json::to_string(&original).unwrap();
        let deserialized: GossipMessage = serde_json::from_str(&serialized).unwrap();
        if let GossipMessage::Chat { name, text, ts, .. } = deserialized {
            assert_eq!(name, "🎙 User");
            assert_eq!(text, "héllo 🎙 世界");
            assert_eq!(ts, 1717171717);
        } else {
            panic!("Expected GossipMessage::Chat");
        }
    }

    // --- S2 authentication (verify_gossip) ---

    #[test]
    fn verify_accepts_a_genuine_signed_payload() {
        let secret = SecretKey::generate();
        let topic = [1u8; 32];
        let p = sign_gossip(&secret, &topic, 5_000, GossipMessage::Leave);
        assert_eq!(verify_gossip(&p, &topic, 5_000, GOSSIP_FRESHNESS_MS), Ok(()));
    }

    #[test]
    fn verify_rejects_a_forged_author() {
        // Attacker signs with their OWN key but claims to be the victim — the
        // classic S2 spoof. The signature can't validate against the victim key.
        let attacker = SecretKey::generate();
        let victim = SecretKey::generate();
        let topic = [2u8; 32];
        let mut p = sign_gossip(&attacker, &topic, 5_000, GossipMessage::Leave);
        p.author = victim.public(); // forge the claimed author
        assert_eq!(
            verify_gossip(&p, &topic, 5_000, GOSSIP_FRESHNESS_MS),
            Err(GossipReject::BadSignature)
        );
    }

    #[test]
    fn verify_rejects_a_tampered_message() {
        // Flipping the message after signing must invalidate the signature.
        let secret = SecretKey::generate();
        let topic = [4u8; 32];
        let mut p = sign_gossip(&secret, &topic, 5_000, GossipMessage::Leave);
        p.msg = GossipMessage::Chat { name: "x".into(), text: "y".into(), ts: 5_000, attachment: None };
        assert_eq!(
            verify_gossip(&p, &topic, 5_000, GOSSIP_FRESHNESS_MS),
            Err(GossipReject::BadSignature)
        );
    }

    #[test]
    fn verify_rejects_cross_room_replay() {
        // A payload signed for one topic must not validate against another room
        // (the topic is bound into the signed bytes).
        let secret = SecretKey::generate();
        let p = sign_gossip(&secret, &[7u8; 32], 5_000, GossipMessage::Leave);
        assert_eq!(
            verify_gossip(&p, &[8u8; 32], 5_000, GOSSIP_FRESHNESS_MS),
            Err(GossipReject::BadSignature)
        );
    }

    #[test]
    fn verify_rejects_stale_and_future_timestamps() {
        let secret = SecretKey::generate();
        let topic = [5u8; 32];
        let p = sign_gossip(&secret, &topic, 1_000_000, GossipMessage::Leave);
        // Far in the past relative to "now" → stale (replay).
        assert_eq!(
            verify_gossip(&p, &topic, 1_000_000 + GOSSIP_FRESHNESS_MS + 1, GOSSIP_FRESHNESS_MS),
            Err(GossipReject::OutOfWindow)
        );
        // Implausibly future.
        assert_eq!(
            verify_gossip(&p, &topic, 1_000_000 - GOSSIP_FRESHNESS_MS - 1, GOSSIP_FRESHNESS_MS),
            Err(GossipReject::OutOfWindow)
        );
        // Within the window (clock skew tolerance) → accepted.
        assert!(verify_gossip(&p, &topic, 1_000_000 + GOSSIP_FRESHNESS_MS - 1, GOSSIP_FRESHNESS_MS).is_ok());
    }

    #[test]
    fn verify_rejects_announce_with_address_for_another_identity() {
        let signer = SecretKey::generate();
        let advertised = SecretKey::generate();
        let topic = [6u8; 32];
        let state = sample_peer_state_for(advertised.public());
        let p = sign_gossip(&signer, &topic, 5_000, GossipMessage::Announce(state));

        assert_eq!(
            verify_gossip(&p, &topic, 5_000, GOSSIP_FRESHNESS_MS),
            Err(GossipReject::AnnounceAddressMismatch)
        );
    }

    #[test]
    fn state_mutation_replay_gate_drops_replayed_leave_and_announce() {
        let author = fresh_id();
        let mut seen = HashMap::new();

        assert!(admit_state_mutation(&mut seen, author, &GossipMessage::Leave, 10));
        assert!(!admit_state_mutation(&mut seen, author, &GossipMessage::Leave, 10));
        assert!(!admit_state_mutation(&mut seen, author, &GossipMessage::Leave, 9));
        assert!(admit_state_mutation(&mut seen, author, &GossipMessage::Leave, 11));

        let announce = GossipMessage::Announce(sample_peer_state_for(author));
        assert!(admit_state_mutation(&mut seen, author, &announce, 10));
        assert!(!admit_state_mutation(&mut seen, author, &announce, 10));
        assert!(!admit_state_mutation(&mut seen, author, &announce, 9));
        assert!(admit_state_mutation(&mut seen, author, &announce, 12));
    }

    #[test]
    fn state_mutation_replay_gate_leaves_chat_ordering_untouched() {
        let author = fresh_id();
        let mut seen = HashMap::new();
        let later_chat = GossipMessage::Chat { name: "A".into(), text: "later".into(), ts: 200, attachment: None };
        let earlier_chat = GossipMessage::Chat { name: "A".into(), text: "earlier".into(), ts: 100, attachment: None };

        assert!(admit_state_mutation(&mut seen, author, &later_chat, 200));
        assert!(admit_state_mutation(&mut seen, author, &earlier_chat, 100));
        assert!(admit_state_mutation(&mut seen, author, &later_chat, 200));
        assert!(seen.is_empty(), "chat must not populate the state-mutation replay map");
    }

    #[test]
    fn state_mutation_replay_gate_is_per_author_and_kind() {
        let author = fresh_id();
        let other = fresh_id();
        let mut seen = HashMap::new();
        let announce = GossipMessage::Announce(sample_peer_state_for(author));

        assert!(admit_state_mutation(&mut seen, author, &GossipMessage::Leave, 5));
        assert!(admit_state_mutation(&mut seen, author, &announce, 5));
        assert!(admit_state_mutation(&mut seen, other, &GossipMessage::Leave, 5));
    }
}