//! Relay room state and admission control.
//!
//! The QUIC server loop admits peers through this room registry, while the
//! registry itself stays socket-free so the relay invariants remain directly
//! testable.

mod config;
mod server;

use std::{collections::HashMap, time::Instant};

use lanparty_ctrl::{
    ControlError, EndpointHello, PeerInfo, Reject, RejectReason, Role, RoomCode, ServerWelcome,
};
use lanparty_obs::{DropReason, FrameAction};
use lanparty_proto::{
    EthernetFrame, MacAddr, ethernet_frame_exceeds_tap_mtu, gateway_lan_safety_drop_reason,
    recommended_tap_mtu, remote_client_safety_drop_reason,
};
use thiserror::Error;

pub use config::{ConfigError, DEFAULT_RELAY_PORT, ListenEndpoint, RelayArgs, RelayConfig};
pub use server::RelayServer;

pub const DEFAULT_MAX_CLIENTS_PER_ROOM: usize = 16;
const MEBIBYTE: u64 = 1024 * 1024;
const CLIENT_MULTICAST_BURST_FRAMES: u32 = 64;
const CLIENT_MULTICAST_REFILL_FRAMES_PER_SECOND: u32 = 32;
const CLIENT_UNKNOWN_UNICAST_BURST_FRAMES: u32 = 64;
const CLIENT_UNKNOWN_UNICAST_REFILL_FRAMES_PER_SECOND: u32 = 32;
const CLIENT_TOTAL_BANDWIDTH_BURST_BYTES: u64 = 4 * MEBIBYTE;
const CLIENT_TOTAL_BANDWIDTH_REFILL_BYTES_PER_SECOND: u64 = 2 * MEBIBYTE;
#[cfg(test)]
const ETHERTYPE_IPV4: u16 = 0x0800;
#[cfg(test)]
const ETHERTYPE_LLDP: u16 = 0x88cc;
#[cfg(test)]
const ETHERTYPE_IPV6: u16 = 0x86dd;
#[cfg(test)]
const ETHERTYPE_8021Q: u16 = 0x8100;
#[cfg(test)]
const ETHERTYPE_8021AD: u16 = 0x88a8;
#[cfg(test)]
const IP_PROTOCOL_UDP: u8 = 17;
#[cfg(test)]
const IPV6_NEXT_HEADER_FRAGMENT: u8 = 44;
#[cfg(test)]
const IPV6_NEXT_HEADER_DESTINATION_OPTIONS: u8 = 60;
#[cfg(test)]
const IPV6_NEXT_HEADER_ICMPV6: u8 = 58;
#[cfg(test)]
const DHCPV4_SERVER_PORT: u16 = 67;
#[cfg(test)]
const DHCPV4_CLIENT_PORT: u16 = 68;
#[cfg(test)]
const DHCPV6_CLIENT_PORT: u16 = 546;
#[cfg(test)]
const DHCPV6_SERVER_PORT: u16 = 547;
#[cfg(test)]
const ICMPV6_ROUTER_ADVERTISEMENT: u8 = 134;

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct JoinAccepted {
    welcome: ServerWelcome,
    peer: PeerInfo,
}

impl JoinAccepted {
    #[must_use]
    const fn new(welcome: ServerWelcome, peer: PeerInfo) -> Self {
        Self { welcome, peer }
    }

    #[must_use]
    pub const fn welcome(&self) -> &ServerWelcome {
        &self.welcome
    }

    #[must_use]
    pub const fn peer(&self) -> &PeerInfo {
        &self.peer
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RoomSnapshot {
    room_id: u64,
    effective_tap_mtu: u16,
    gateway: Option<PeerInfo>,
    clients: Vec<PeerInfo>,
    last_seen_by_peer: HashMap<u32, Instant>,
}

impl RoomSnapshot {
    #[must_use]
    pub const fn room_id(&self) -> u64 {
        self.room_id
    }

    #[must_use]
    pub const fn effective_tap_mtu(&self) -> u16 {
        self.effective_tap_mtu
    }

    #[must_use]
    pub const fn gateway(&self) -> Option<&PeerInfo> {
        self.gateway.as_ref()
    }

    #[must_use]
    pub fn clients(&self) -> &[PeerInfo] {
        &self.clients
    }

    #[must_use]
    pub fn last_seen(&self, peer_id: u32) -> Option<Instant> {
        self.last_seen_by_peer.get(&peer_id).copied()
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LeaveResult {
    peer: PeerInfo,
    room_removed: bool,
}

impl LeaveResult {
    #[must_use]
    const fn new(peer: PeerInfo, room_removed: bool) -> Self {
        Self { peer, room_removed }
    }

    #[must_use]
    pub const fn peer(&self) -> &PeerInfo {
        &self.peer
    }

    #[must_use]
    pub const fn room_removed(&self) -> bool {
        self.room_removed
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ForwardingDecision {
    targets: Vec<u32>,
    action: FrameAction,
    drop_reason: Option<DropReason>,
}

impl ForwardingDecision {
    #[must_use]
    fn forwarded(targets: Vec<u32>) -> Self {
        Self {
            targets,
            action: FrameAction::Forwarded,
            drop_reason: None,
        }
    }

    #[must_use]
    fn dropped(drop_reason: DropReason) -> Self {
        Self {
            targets: Vec::new(),
            action: FrameAction::Dropped,
            drop_reason: Some(drop_reason),
        }
    }

    #[must_use]
    fn filtered(drop_reason: DropReason) -> Self {
        Self {
            targets: Vec::new(),
            action: FrameAction::Filtered,
            drop_reason: Some(drop_reason),
        }
    }

    #[must_use]
    fn rate_limited() -> Self {
        Self {
            targets: Vec::new(),
            action: FrameAction::RateLimited,
            drop_reason: Some(DropReason::RateLimit),
        }
    }

    #[must_use]
    pub fn targets(&self) -> &[u32] {
        &self.targets
    }

    #[must_use]
    pub const fn action(&self) -> FrameAction {
        self.action
    }

    #[must_use]
    pub const fn drop_reason(&self) -> Option<DropReason> {
        self.drop_reason
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Error)]
pub enum ForwardingError {
    #[error("room {0} does not exist")]
    UnknownRoom(RoomCode),
    #[error("ingress peer {peer_id} is not present in room {room}")]
    UnknownIngressPeer { room: RoomCode, peer_id: u32 },
}

#[derive(Debug, Clone)]
pub struct RoomRegistry {
    rooms: HashMap<RoomCode, Room>,
    next_room_id: u64,
    max_clients_per_room: usize,
}

impl Default for RoomRegistry {
    fn default() -> Self {
        Self::new(DEFAULT_MAX_CLIENTS_PER_ROOM)
    }
}

impl RoomRegistry {
    #[must_use]
    pub fn new(max_clients_per_room: usize) -> Self {
        assert!(
            max_clients_per_room > 0,
            "max_clients_per_room must be greater than zero"
        );

        Self {
            rooms: HashMap::new(),
            next_room_id: 1,
            max_clients_per_room,
        }
    }

    pub fn join(&mut self, hello: EndpointHello) -> Result<JoinAccepted, Reject> {
        hello.validate().map_err(reject_control_error)?;

        let supported_tap_mtu = recommended_tap_mtu(usize::from(hello.max_datagram_size()))
            .map_err(|error| Reject::new(RejectReason::MtuTooSmall, error.to_string()))?
            as u16;
        let room_code = hello.room().clone();

        if !self.rooms.contains_key(&room_code) {
            let room_id = self.allocate_room_id()?;
            self.rooms.insert(
                room_code.clone(),
                Room::new(room_id, self.max_clients_per_room),
            );
        }

        self.rooms
            .get_mut(&room_code)
            .expect("room was inserted before lookup")
            .join(hello, supported_tap_mtu)
    }

    #[must_use]
    pub fn room_count(&self) -> usize {
        self.rooms.len()
    }

    #[must_use]
    pub fn snapshot(&self, room: &RoomCode) -> Option<RoomSnapshot> {
        self.rooms.get(room).map(Room::snapshot)
    }

    pub fn leave(&mut self, room: &RoomCode, peer_id: u32) -> Result<LeaveResult, ForwardingError> {
        let room_state = self
            .rooms
            .get_mut(room)
            .ok_or_else(|| ForwardingError::UnknownRoom(room.clone()))?;
        let peer = room_state.leave(room, peer_id)?;
        let room_removed = room_state.is_empty();

        if room_removed {
            self.rooms.remove(room);
        }

        Ok(LeaveResult::new(peer, room_removed))
    }

    pub fn forward_ethernet(
        &mut self,
        room: &RoomCode,
        ingress_peer_id: u32,
        frame_bytes: &[u8],
    ) -> Result<ForwardingDecision, ForwardingError> {
        self.forward_ethernet_at(room, ingress_peer_id, frame_bytes, Instant::now())
    }

    fn forward_ethernet_at(
        &mut self,
        room: &RoomCode,
        ingress_peer_id: u32,
        frame_bytes: &[u8],
        now: Instant,
    ) -> Result<ForwardingDecision, ForwardingError> {
        self.rooms
            .get_mut(room)
            .ok_or_else(|| ForwardingError::UnknownRoom(room.clone()))?
            .forward_ethernet(room, ingress_peer_id, frame_bytes, now)
    }

    fn allocate_room_id(&mut self) -> Result<u64, Reject> {
        let room_id = self.next_room_id;
        self.next_room_id = self
            .next_room_id
            .checked_add(1)
            .ok_or_else(|| Reject::new(RejectReason::InternalError, "room id space exhausted"))?;

        Ok(room_id)
    }
}

#[derive(Debug, Clone)]
struct Room {
    room_id: u64,
    next_peer_id: u32,
    max_clients: usize,
    effective_tap_mtu: Option<u16>,
    gateway: Option<PeerEntry>,
    clients: HashMap<u32, PeerEntry>,
    clients_by_mac: HashMap<MacAddr, u32>,
    client_multicast_limits: HashMap<u32, TokenBucket>,
    client_unknown_unicast_limits: HashMap<u32, TokenBucket>,
    client_total_bandwidth_limits: HashMap<u32, TokenBucket>,
}

#[derive(Debug, Clone)]
struct PeerEntry {
    info: PeerInfo,
    last_seen: Instant,
}

impl PeerEntry {
    const fn new(info: PeerInfo, last_seen: Instant) -> Self {
        Self { info, last_seen }
    }
}

impl Room {
    fn new(room_id: u64, max_clients: usize) -> Self {
        Self {
            room_id,
            next_peer_id: 1,
            max_clients,
            effective_tap_mtu: None,
            gateway: None,
            clients: HashMap::new(),
            clients_by_mac: HashMap::new(),
            client_multicast_limits: HashMap::new(),
            client_unknown_unicast_limits: HashMap::new(),
            client_total_bandwidth_limits: HashMap::new(),
        }
    }

    fn join(
        &mut self,
        hello: EndpointHello,
        supported_tap_mtu: u16,
    ) -> Result<JoinAccepted, Reject> {
        self.validate_role_capacity(&hello)?;
        let effective_tap_mtu = self.accept_effective_mtu(supported_tap_mtu)?;

        let peer_id = self.allocate_peer_id()?;
        let peer =
            PeerInfo::new(peer_id, hello.role(), hello.announced_mac()).map_err(|error| {
                Reject::new(
                    RejectReason::MalformedHello,
                    format!("invalid peer: {error}"),
                )
            })?;
        let gateway_peer_id = match peer.role() {
            Role::Gateway => Some(peer.peer_id()),
            Role::Client => self.gateway.as_ref().map(|gateway| gateway.info.peer_id()),
        };
        let welcome = ServerWelcome::new(self.room_id, peer_id, effective_tap_mtu)
            .map(|welcome| welcome.with_gateway_peer_id(gateway_peer_id))
            .map_err(|error| {
                Reject::new(
                    RejectReason::InternalError,
                    format!("welcome failed: {error}"),
                )
            })?;

        let joined_at = Instant::now();

        match peer.role() {
            Role::Gateway => {
                self.gateway = Some(PeerEntry::new(peer.clone(), joined_at));
            }
            Role::Client => {
                let mac = peer.mac().expect("client peer info has MAC");
                self.clients_by_mac.insert(mac, peer.peer_id());
                self.clients
                    .insert(peer.peer_id(), PeerEntry::new(peer.clone(), joined_at));
                self.client_multicast_limits
                    .insert(peer.peer_id(), client_multicast_limit());
                self.client_unknown_unicast_limits
                    .insert(peer.peer_id(), client_unknown_unicast_limit());
                self.client_total_bandwidth_limits
                    .insert(peer.peer_id(), client_total_bandwidth_limit());
            }
        }

        Ok(JoinAccepted::new(welcome, peer))
    }

    fn accept_effective_mtu(&mut self, supported_tap_mtu: u16) -> Result<u16, Reject> {
        match self.effective_tap_mtu {
            Some(existing) if supported_tap_mtu < existing => Err(Reject::new(
                RejectReason::MtuTooSmall,
                format!("peer supports TAP MTU {supported_tap_mtu}, room requires {existing}"),
            )),
            Some(existing) => Ok(existing),
            None => {
                self.effective_tap_mtu = Some(supported_tap_mtu);
                Ok(supported_tap_mtu)
            }
        }
    }

    fn validate_role_capacity(&self, hello: &EndpointHello) -> Result<(), Reject> {
        match hello.role() {
            Role::Gateway if self.gateway.is_some() => Err(Reject::new(
                RejectReason::GatewayAlreadyConnected,
                "room already has a gateway",
            )),
            Role::Gateway => Ok(()),
            Role::Client if self.clients.len() >= self.max_clients => {
                Err(Reject::new(RejectReason::RoomFull, "room is full"))
            }
            Role::Client => {
                let mac = hello
                    .announced_mac()
                    .expect("validated client hello has MAC address");

                if self.clients_by_mac.contains_key(&mac) {
                    return Err(Reject::new(
                        RejectReason::DuplicateMac,
                        format!("client MAC {mac} is already present"),
                    ));
                }

                Ok(())
            }
        }
    }

    fn allocate_peer_id(&mut self) -> Result<u32, Reject> {
        let peer_id = self.next_peer_id;
        self.next_peer_id = self
            .next_peer_id
            .checked_add(1)
            .ok_or_else(|| Reject::new(RejectReason::InternalError, "peer id space exhausted"))?;

        Ok(peer_id)
    }

    fn snapshot(&self) -> RoomSnapshot {
        let mut clients: Vec<_> = self
            .clients
            .values()
            .map(|entry| entry.info.clone())
            .collect();
        clients.sort_by_key(PeerInfo::peer_id);
        let mut last_seen_by_peer =
            HashMap::with_capacity(self.clients.len() + usize::from(self.gateway.is_some()));

        if let Some(gateway) = &self.gateway {
            last_seen_by_peer.insert(gateway.info.peer_id(), gateway.last_seen);
        }
        last_seen_by_peer.extend(
            self.clients
                .values()
                .map(|client| (client.info.peer_id(), client.last_seen)),
        );

        RoomSnapshot {
            room_id: self.room_id,
            effective_tap_mtu: self.effective_tap_mtu.unwrap_or_default(),
            gateway: self.gateway.as_ref().map(|gateway| gateway.info.clone()),
            clients,
            last_seen_by_peer,
        }
    }

    fn leave(&mut self, room_code: &RoomCode, peer_id: u32) -> Result<PeerInfo, ForwardingError> {
        if self
            .gateway
            .as_ref()
            .is_some_and(|gateway| gateway.info.peer_id() == peer_id)
        {
            return Ok(self.gateway.take().expect("gateway exists").info);
        }

        let entry =
            self.clients
                .remove(&peer_id)
                .ok_or_else(|| ForwardingError::UnknownIngressPeer {
                    room: room_code.clone(),
                    peer_id,
                })?;
        let peer = entry.info;
        let mac = peer.mac().expect("client peer info has MAC");
        self.clients_by_mac.remove(&mac);
        self.client_multicast_limits.remove(&peer.peer_id());
        self.client_unknown_unicast_limits.remove(&peer.peer_id());
        self.client_total_bandwidth_limits.remove(&peer.peer_id());

        Ok(peer)
    }

    fn is_empty(&self) -> bool {
        self.gateway.is_none() && self.clients.is_empty()
    }

    fn forward_ethernet(
        &mut self,
        room_code: &RoomCode,
        ingress_peer_id: u32,
        frame_bytes: &[u8],
        now: Instant,
    ) -> Result<ForwardingDecision, ForwardingError> {
        let ingress =
            self.peer(ingress_peer_id)
                .ok_or_else(|| ForwardingError::UnknownIngressPeer {
                    room: room_code.clone(),
                    peer_id: ingress_peer_id,
                })?;
        let ingress_role = ingress.role();
        let ingress_mac = ingress.mac();
        let frame = match EthernetFrame::parse(frame_bytes) {
            Ok(frame) => frame,
            Err(_) => return Ok(ForwardingDecision::dropped(DropReason::Malformed)),
        };

        if !frame.source().is_valid_unicast() {
            return Ok(ForwardingDecision::filtered(DropReason::InvalidSourceMac));
        }

        if ingress_role == Role::Client {
            let expected_source = ingress_mac.expect("client peers have MAC addresses");
            if frame.source() != expected_source {
                return Ok(ForwardingDecision::filtered(
                    DropReason::UnauthorizedSourceMac,
                ));
            }
        }

        self.mark_peer_seen(ingress_peer_id, now);

        let safety_drop_reason = match ingress_role {
            Role::Client => remote_client_safety_drop_reason(frame),
            Role::Gateway => gateway_lan_safety_drop_reason(frame),
        };
        if let Some(drop_reason) = safety_drop_reason {
            return Ok(ForwardingDecision::filtered(DropReason::from(drop_reason)));
        }
        if let Some(effective_tap_mtu) = self.effective_tap_mtu
            && ethernet_frame_exceeds_tap_mtu(frame, usize::from(effective_tap_mtu))
        {
            return Ok(ForwardingDecision::dropped(DropReason::TapMtuExceeded));
        }

        if ingress_role == Role::Client
            && !self.allow_client_total_bandwidth(ingress_peer_id, frame.len() as u64, now)
        {
            return Ok(ForwardingDecision::rate_limited());
        }

        if ingress_role == Role::Client
            && frame.destination().is_multicast()
            && !self.allow_client_multicast(ingress_peer_id, now)
        {
            return Ok(ForwardingDecision::rate_limited());
        }

        let destination = frame.destination();
        let targets = if destination.is_multicast() {
            self.all_peer_ids_except(ingress_peer_id)
        } else if let Some(client_peer_id) = self.clients_by_mac.get(&destination) {
            if *client_peer_id == ingress_peer_id {
                Vec::new()
            } else {
                vec![*client_peer_id]
            }
        } else {
            if ingress_role == Role::Client
                && !self.allow_client_unknown_unicast(ingress_peer_id, now)
            {
                return Ok(ForwardingDecision::rate_limited());
            }
            match ingress_role {
                Role::Client => self.gateway_peer_id_except(ingress_peer_id),
                Role::Gateway => Vec::new(),
            }
        };

        if targets.is_empty() {
            return Ok(ForwardingDecision::dropped(DropReason::UnknownDestination));
        }

        Ok(ForwardingDecision::forwarded(targets))
    }

    fn allow_client_multicast(&mut self, peer_id: u32, now: Instant) -> bool {
        self.client_multicast_limits
            .entry(peer_id)
            .or_insert_with(client_multicast_limit)
            .allow(1, now)
    }

    fn allow_client_unknown_unicast(&mut self, peer_id: u32, now: Instant) -> bool {
        self.client_unknown_unicast_limits
            .entry(peer_id)
            .or_insert_with(client_unknown_unicast_limit)
            .allow(1, now)
    }

    fn allow_client_total_bandwidth(&mut self, peer_id: u32, bytes: u64, now: Instant) -> bool {
        self.client_total_bandwidth_limits
            .entry(peer_id)
            .or_insert_with(client_total_bandwidth_limit)
            .allow(bytes, now)
    }

    fn peer(&self, peer_id: u32) -> Option<&PeerInfo> {
        self.gateway
            .as_ref()
            .filter(|peer| peer.info.peer_id() == peer_id)
            .map(|peer| &peer.info)
            .or_else(|| self.clients.get(&peer_id).map(|peer| &peer.info))
    }

    fn mark_peer_seen(&mut self, peer_id: u32, seen_at: Instant) {
        if let Some(gateway) = self
            .gateway
            .as_mut()
            .filter(|peer| peer.info.peer_id() == peer_id)
        {
            gateway.last_seen = seen_at;
            return;
        }

        if let Some(client) = self.clients.get_mut(&peer_id) {
            client.last_seen = seen_at;
        }
    }

    fn all_peer_ids_except(&self, ingress_peer_id: u32) -> Vec<u32> {
        let mut peer_ids =
            Vec::with_capacity(self.clients.len() + usize::from(self.gateway.is_some()));

        if let Some(gateway) = &self.gateway
            && gateway.info.peer_id() != ingress_peer_id
        {
            peer_ids.push(gateway.info.peer_id());
        }

        peer_ids.extend(
            self.clients
                .keys()
                .copied()
                .filter(|peer_id| *peer_id != ingress_peer_id),
        );
        peer_ids.sort_unstable();
        peer_ids
    }

    fn gateway_peer_id_except(&self, ingress_peer_id: u32) -> Vec<u32> {
        self.gateway
            .as_ref()
            .filter(|gateway| gateway.info.peer_id() != ingress_peer_id)
            .map_or_else(Vec::new, |gateway| vec![gateway.info.peer_id()])
    }
}

#[derive(Debug, Clone)]
struct TokenBucket {
    tokens: u64,
    capacity: u64,
    refill_per_second: u64,
    last_refill: Instant,
}

impl TokenBucket {
    fn new(capacity: u64, refill_per_second: u64) -> Self {
        assert!(capacity > 0, "rate-limit capacity must be nonzero");
        assert!(
            refill_per_second > 0,
            "rate-limit refill rate must be nonzero"
        );

        Self {
            tokens: capacity,
            capacity,
            refill_per_second,
            last_refill: Instant::now(),
        }
    }

    fn allow(&mut self, cost: u64, now: Instant) -> bool {
        self.refill(now);
        if self.tokens < cost {
            false
        } else {
            self.tokens -= cost;
            true
        }
    }

    fn refill(&mut self, now: Instant) {
        let elapsed_secs = now.saturating_duration_since(self.last_refill).as_secs();
        if elapsed_secs == 0 {
            return;
        }

        let refill = elapsed_secs.saturating_mul(self.refill_per_second);
        self.tokens = self.tokens.saturating_add(refill).min(self.capacity);
        self.last_refill = now;
    }
}

fn client_multicast_limit() -> TokenBucket {
    TokenBucket::new(
        u64::from(CLIENT_MULTICAST_BURST_FRAMES),
        u64::from(CLIENT_MULTICAST_REFILL_FRAMES_PER_SECOND),
    )
}

fn client_unknown_unicast_limit() -> TokenBucket {
    TokenBucket::new(
        u64::from(CLIENT_UNKNOWN_UNICAST_BURST_FRAMES),
        u64::from(CLIENT_UNKNOWN_UNICAST_REFILL_FRAMES_PER_SECOND),
    )
}

fn client_total_bandwidth_limit() -> TokenBucket {
    TokenBucket::new(
        CLIENT_TOTAL_BANDWIDTH_BURST_BYTES,
        CLIENT_TOTAL_BANDWIDTH_REFILL_BYTES_PER_SECOND,
    )
}

fn reject_control_error(error: ControlError) -> Reject {
    let reason = match error {
        ControlError::UnsupportedVersion { .. } => RejectReason::UnsupportedVersion,
        ControlError::InvalidClientMac { .. } => RejectReason::InvalidMac,
        ControlError::Mtu(_) => RejectReason::MtuTooSmall,
        ControlError::EmptyRoomCode
        | ControlError::RoomCodeTooLong { .. }
        | ControlError::InvalidRoomCodeByte { .. }
        | ControlError::MissingClientMac
        | ControlError::UnexpectedGatewayMac
        | ControlError::InvalidPeerId
        | ControlError::EffectiveMtuTooSmall { .. } => RejectReason::MalformedHello,
    };

    Reject::new(reason, error.to_string())
}

#[cfg(test)]
mod tests {
    use std::time::{Duration, Instant};

    use super::*;
    use lanparty_proto::MAX_STANDARD_ETHERNET_FRAME_LEN;

    fn room() -> RoomCode {
        RoomCode::new("ABCD").unwrap()
    }

    fn mac(last: u8) -> MacAddr {
        MacAddr::new([0x02, 0, 0, 0, 0, last])
    }

    fn client_hello(last_mac_octet: u8) -> EndpointHello {
        EndpointHello::client(room(), mac(last_mac_octet), 1400).unwrap()
    }

    fn gateway_hello() -> EndpointHello {
        EndpointHello::gateway(room(), 1400).unwrap()
    }

    fn ethernet(destination: MacAddr, source: MacAddr) -> Vec<u8> {
        ethernet_with_payload(destination, source, ETHERTYPE_IPV4, &[])
    }

    fn ethernet_with_payload(
        destination: MacAddr,
        source: MacAddr,
        ethertype_or_len: u16,
        payload: &[u8],
    ) -> Vec<u8> {
        let mut frame = Vec::new();
        frame.extend_from_slice(&destination.octets());
        frame.extend_from_slice(&source.octets());
        frame.extend_from_slice(&ethertype_or_len.to_be_bytes());
        frame.extend_from_slice(payload);
        frame
    }

    fn ipv4_udp_payload(source_port: u16, destination_port: u16) -> Vec<u8> {
        let mut packet = vec![0; 28];
        packet[0] = 0x45;
        packet[9] = IP_PROTOCOL_UDP;
        packet[20..22].copy_from_slice(&source_port.to_be_bytes());
        packet[22..24].copy_from_slice(&destination_port.to_be_bytes());
        packet
    }

    fn udp_payload(source_port: u16, destination_port: u16) -> Vec<u8> {
        let mut packet = vec![0; 8];
        packet[0..2].copy_from_slice(&source_port.to_be_bytes());
        packet[2..4].copy_from_slice(&destination_port.to_be_bytes());
        packet
    }

    fn ipv6_udp_payload(source_port: u16, destination_port: u16) -> Vec<u8> {
        ipv6_payload(IP_PROTOCOL_UDP, &udp_payload(source_port, destination_port))
    }

    fn ipv6_udp_after_destination_options_payload(
        source_port: u16,
        destination_port: u16,
    ) -> Vec<u8> {
        ipv6_payload(
            IPV6_NEXT_HEADER_DESTINATION_OPTIONS,
            &ipv6_extension_payload(IP_PROTOCOL_UDP, &udp_payload(source_port, destination_port)),
        )
    }

    fn ipv6_router_advertisement_payload() -> Vec<u8> {
        ipv6_payload(IPV6_NEXT_HEADER_ICMPV6, &[ICMPV6_ROUTER_ADVERTISEMENT])
    }

    fn ipv6_router_advertisement_after_destination_options_payload() -> Vec<u8> {
        ipv6_payload(
            IPV6_NEXT_HEADER_DESTINATION_OPTIONS,
            &ipv6_extension_payload(IPV6_NEXT_HEADER_ICMPV6, &[ICMPV6_ROUTER_ADVERTISEMENT]),
        )
    }

    fn ipv6_echo_request_after_destination_options_payload() -> Vec<u8> {
        ipv6_payload(
            IPV6_NEXT_HEADER_DESTINATION_OPTIONS,
            &ipv6_extension_payload(IPV6_NEXT_HEADER_ICMPV6, &[128]),
        )
    }

    fn ipv6_fragment_payload(next_header: u8, fragment_payload: &[u8]) -> Vec<u8> {
        let mut packet = Vec::with_capacity(8 + fragment_payload.len());
        packet.push(next_header);
        packet.push(0);
        packet.extend_from_slice(&0_u16.to_be_bytes());
        packet.extend_from_slice(&1_u32.to_be_bytes());
        packet.extend_from_slice(fragment_payload);
        packet
    }

    fn ipv6_payload(next_header: u8, upper_payload: &[u8]) -> Vec<u8> {
        let mut packet = vec![0; 40];
        packet[0] = 0x60;
        packet[6] = next_header;
        packet.extend_from_slice(upper_payload);
        packet
    }

    fn ipv6_extension_payload(next_header: u8, upper_payload: &[u8]) -> Vec<u8> {
        let mut packet = Vec::with_capacity(8 + upper_payload.len());
        packet.push(next_header);
        packet.push(0);
        packet.extend_from_slice(&[0; 6]);
        packet.extend_from_slice(upper_payload);
        packet
    }

    fn physical_mac() -> MacAddr {
        MacAddr::new([0x00, 1, 2, 3, 4, 5])
    }

    fn assert_dropped(decision: &ForwardingDecision, reason: DropReason) {
        assert_eq!(decision.action(), FrameAction::Dropped);
        assert_eq!(decision.drop_reason(), Some(reason));
        assert!(decision.targets().is_empty());
    }

    fn assert_filtered(decision: &ForwardingDecision, reason: DropReason) {
        assert_eq!(decision.action(), FrameAction::Filtered);
        assert_eq!(decision.drop_reason(), Some(reason));
        assert!(decision.targets().is_empty());
    }

    fn assert_rate_limited(decision: &ForwardingDecision) {
        assert_eq!(decision.action(), FrameAction::RateLimited);
        assert_eq!(decision.drop_reason(), Some(DropReason::RateLimit));
        assert!(decision.targets().is_empty());
    }

    #[test]
    fn accepts_gateway_and_client_into_room() {
        let mut registry = RoomRegistry::default();

        let gateway = registry.join(gateway_hello()).unwrap();
        let client = registry.join(client_hello(1)).unwrap();
        let snapshot = registry.snapshot(&room()).unwrap();

        assert_eq!(registry.room_count(), 1);
        assert_eq!(gateway.peer().role(), Role::Gateway);
        assert_eq!(gateway.welcome().peer_id(), 1);
        assert!(gateway.welcome().gateway_connected());
        assert_eq!(gateway.welcome().gateway_peer_id(), Some(1));
        assert_eq!(client.peer().role(), Role::Client);
        assert_eq!(client.welcome().peer_id(), 2);
        assert!(client.welcome().gateway_connected());
        assert_eq!(client.welcome().gateway_peer_id(), Some(1));
        assert_eq!(snapshot.gateway().unwrap().peer_id(), 1);
        assert_eq!(snapshot.clients().len(), 1);
        assert_eq!(snapshot.effective_tap_mtu(), 1200);
        assert!(snapshot.last_seen(gateway.peer().peer_id()).is_some());
        assert!(snapshot.last_seen(client.peer().peer_id()).is_some());
    }

    #[test]
    fn reports_missing_gateway_to_client_joining_first() {
        let mut registry = RoomRegistry::default();

        let client = registry.join(client_hello(1)).unwrap();

        assert!(!client.welcome().gateway_connected());
        assert_eq!(client.welcome().gateway_peer_id(), None);
    }

    #[test]
    fn rejects_second_gateway() {
        let mut registry = RoomRegistry::default();
        registry.join(gateway_hello()).unwrap();

        let reject = registry.join(gateway_hello()).unwrap_err();

        assert_eq!(reject.reason(), &RejectReason::GatewayAlreadyConnected);
    }

    #[test]
    fn rejects_duplicate_client_mac() {
        let mut registry = RoomRegistry::default();
        registry.join(client_hello(1)).unwrap();

        let reject = registry.join(client_hello(1)).unwrap_err();

        assert_eq!(reject.reason(), &RejectReason::DuplicateMac);
    }

    #[test]
    fn enforces_client_limit() {
        let mut registry = RoomRegistry::new(1);
        registry.join(client_hello(1)).unwrap();

        let reject = registry.join(client_hello(2)).unwrap_err();

        assert_eq!(reject.reason(), &RejectReason::RoomFull);
    }

    #[test]
    fn keeps_room_mtu_stable_after_first_peer() {
        let mut registry = RoomRegistry::default();
        let first = EndpointHello::client(room(), mac(1), 1024).unwrap();
        registry.join(first).unwrap();

        let reject = registry
            .join(EndpointHello::gateway(room(), 900).unwrap())
            .unwrap_err();

        assert_eq!(reject.reason(), &RejectReason::MtuTooSmall);
        assert_eq!(registry.snapshot(&room()).unwrap().effective_tap_mtu(), 972);
    }

    #[test]
    fn second_peer_uses_existing_lower_room_mtu() {
        let mut registry = RoomRegistry::default();
        let first = EndpointHello::client(room(), mac(1), 1024).unwrap();
        registry.join(first).unwrap();

        let gateway = registry.join(gateway_hello()).unwrap();

        assert_eq!(gateway.welcome().effective_tap_mtu(), 972);
    }

    #[test]
    fn removes_client_from_room_indexes() {
        let mut registry = RoomRegistry::default();
        let client = registry.join(client_hello(1)).unwrap();
        registry.join(client_hello(2)).unwrap();

        let result = registry.leave(&room(), client.peer().peer_id()).unwrap();
        let snapshot = registry.snapshot(&room()).unwrap();

        assert_eq!(result.peer(), client.peer());
        assert!(!result.room_removed());
        assert_eq!(snapshot.clients().len(), 1);
        assert_eq!(snapshot.last_seen(client.peer().peer_id()), None);
        assert!(registry.join(client_hello(1)).is_ok());
    }

    #[test]
    fn removes_empty_room_after_last_peer_leaves() {
        let mut registry = RoomRegistry::default();
        let client = registry.join(client_hello(1)).unwrap();

        let result = registry.leave(&room(), client.peer().peer_id()).unwrap();

        assert_eq!(result.peer(), client.peer());
        assert!(result.room_removed());
        assert_eq!(registry.room_count(), 0);
        assert!(registry.snapshot(&room()).is_none());
    }

    #[test]
    fn removes_gateway_without_removing_room_when_clients_remain() {
        let mut registry = RoomRegistry::default();
        let gateway = registry.join(gateway_hello()).unwrap();
        registry.join(client_hello(1)).unwrap();

        let result = registry.leave(&room(), gateway.peer().peer_id()).unwrap();
        let snapshot = registry.snapshot(&room()).unwrap();

        assert_eq!(result.peer(), gateway.peer());
        assert!(!result.room_removed());
        assert!(snapshot.gateway().is_none());
        assert_eq!(snapshot.last_seen(gateway.peer().peer_id()), None);
        assert_eq!(snapshot.clients().len(), 1);
        assert!(registry.join(gateway_hello()).is_ok());
    }

    #[test]
    fn reports_unknown_peer_on_leave() {
        let mut registry = RoomRegistry::default();
        registry.join(client_hello(1)).unwrap();

        let error = registry.leave(&room(), 99).unwrap_err();

        assert_eq!(
            error,
            ForwardingError::UnknownIngressPeer {
                room: room(),
                peer_id: 99,
            }
        );
    }

    #[test]
    fn forwards_unknown_client_unicast_to_gateway() {
        let mut registry = RoomRegistry::default();
        let gateway = registry.join(gateway_hello()).unwrap();
        let client_one = registry.join(client_hello(1)).unwrap();
        let client_two = registry.join(client_hello(2)).unwrap();
        let frame = ethernet(MacAddr::new([0x00, 1, 2, 3, 4, 5]), mac(1));

        let decision = registry
            .forward_ethernet(&room(), client_one.peer().peer_id(), &frame)
            .unwrap();

        assert_eq!(decision.action(), FrameAction::Forwarded);
        assert_eq!(decision.targets(), &[gateway.peer().peer_id()]);
        assert!(!decision.targets().contains(&client_two.peer().peer_id()));
        assert_eq!(decision.drop_reason(), None);
    }

    #[test]
    fn drops_gateway_unicast_to_unknown_remote_mac() {
        let mut registry = RoomRegistry::default();
        let gateway = registry.join(gateway_hello()).unwrap();
        registry.join(client_hello(1)).unwrap();
        registry.join(client_hello(2)).unwrap();
        let frame = ethernet(physical_mac(), MacAddr::new([0x00, 1, 2, 3, 4, 5]));

        let decision = registry
            .forward_ethernet(&room(), gateway.peer().peer_id(), &frame)
            .unwrap();

        assert_dropped(&decision, DropReason::UnknownDestination);
    }

    #[test]
    fn forwards_gateway_unicast_to_matching_client() {
        let mut registry = RoomRegistry::default();
        let gateway = registry.join(gateway_hello()).unwrap();
        let client_one = registry.join(client_hello(1)).unwrap();
        let client_two = registry.join(client_hello(2)).unwrap();
        let frame = ethernet(mac(2), MacAddr::new([0x00, 1, 2, 3, 4, 5]));

        let decision = registry
            .forward_ethernet(&room(), gateway.peer().peer_id(), &frame)
            .unwrap();

        assert_eq!(decision.action(), FrameAction::Forwarded);
        assert_eq!(decision.targets(), &[client_two.peer().peer_id()]);
        assert!(!decision.targets().contains(&client_one.peer().peer_id()));
    }

    #[test]
    fn floods_broadcast_without_reflecting_ingress() {
        let mut registry = RoomRegistry::default();
        let gateway = registry.join(gateway_hello()).unwrap();
        let client_one = registry.join(client_hello(1)).unwrap();
        let client_two = registry.join(client_hello(2)).unwrap();
        let frame = ethernet(MacAddr::BROADCAST, mac(1));

        let decision = registry
            .forward_ethernet(&room(), client_one.peer().peer_id(), &frame)
            .unwrap();

        assert_eq!(decision.action(), FrameAction::Forwarded);
        assert_eq!(
            decision.targets(),
            &[gateway.peer().peer_id(), client_two.peer().peer_id()]
        );
    }

    #[test]
    fn refreshes_peer_last_seen_after_valid_frames() {
        let mut registry = RoomRegistry::default();
        let gateway = registry.join(gateway_hello()).unwrap();
        let client = registry.join(client_hello(1)).unwrap();
        let client_seen_at = Instant::now() + Duration::from_secs(5);
        let gateway_seen_at = client_seen_at + Duration::from_secs(1);
        let client_frame = ethernet(MacAddr::BROADCAST, mac(1));
        let gateway_frame = ethernet(mac(1), physical_mac());

        registry
            .forward_ethernet_at(
                &room(),
                client.peer().peer_id(),
                &client_frame,
                client_seen_at,
            )
            .unwrap();
        registry
            .forward_ethernet_at(
                &room(),
                gateway.peer().peer_id(),
                &gateway_frame,
                gateway_seen_at,
            )
            .unwrap();

        let snapshot = registry.snapshot(&room()).unwrap();
        assert_eq!(
            snapshot.last_seen(client.peer().peer_id()),
            Some(client_seen_at)
        );
        assert_eq!(
            snapshot.last_seen(gateway.peer().peer_id()),
            Some(gateway_seen_at)
        );
    }

    #[test]
    fn keeps_last_seen_unchanged_for_unauthorized_client_source() {
        let mut registry = RoomRegistry::default();
        let client = registry.join(client_hello(1)).unwrap();
        let before = registry
            .snapshot(&room())
            .unwrap()
            .last_seen(client.peer().peer_id())
            .unwrap();
        let frame = ethernet(MacAddr::BROADCAST, mac(2));

        let decision = registry
            .forward_ethernet_at(
                &room(),
                client.peer().peer_id(),
                &frame,
                before + Duration::from_secs(5),
            )
            .unwrap();
        assert_filtered(&decision, DropReason::UnauthorizedSourceMac);

        assert_eq!(
            registry
                .snapshot(&room())
                .unwrap()
                .last_seen(client.peer().peer_id()),
            Some(before)
        );
    }

    #[test]
    fn rate_limits_client_broadcast_after_burst() {
        let mut registry = RoomRegistry::default();
        let gateway = registry.join(gateway_hello()).unwrap();
        let client_one = registry.join(client_hello(1)).unwrap();
        let client_two = registry.join(client_hello(2)).unwrap();
        let frame = ethernet(MacAddr::BROADCAST, mac(1));
        let now = Instant::now();

        for _ in 0..CLIENT_MULTICAST_BURST_FRAMES {
            let decision = registry
                .forward_ethernet_at(&room(), client_one.peer().peer_id(), &frame, now)
                .unwrap();
            assert_eq!(decision.action(), FrameAction::Forwarded);
            assert_eq!(
                decision.targets(),
                &[gateway.peer().peer_id(), client_two.peer().peer_id()]
            );
        }

        let decision = registry
            .forward_ethernet_at(&room(), client_one.peer().peer_id(), &frame, now)
            .unwrap();
        assert_rate_limited(&decision);

        let decision = registry
            .forward_ethernet_at(
                &room(),
                client_one.peer().peer_id(),
                &frame,
                now + Duration::from_secs(1),
            )
            .unwrap();
        assert_eq!(decision.action(), FrameAction::Forwarded);
    }

    #[test]
    fn rate_limits_client_unknown_unicast_after_burst() {
        let mut registry = RoomRegistry::default();
        let gateway = registry.join(gateway_hello()).unwrap();
        let client_one = registry.join(client_hello(1)).unwrap();
        let client_two = registry.join(client_hello(2)).unwrap();
        let unknown_unicast = ethernet(physical_mac(), mac(1));
        let known_unicast = ethernet(mac(2), mac(1));
        let now = Instant::now();

        for _ in 0..CLIENT_UNKNOWN_UNICAST_BURST_FRAMES {
            let decision = registry
                .forward_ethernet_at(&room(), client_one.peer().peer_id(), &unknown_unicast, now)
                .unwrap();
            assert_eq!(decision.action(), FrameAction::Forwarded);
            assert_eq!(decision.targets(), &[gateway.peer().peer_id()]);
        }

        let decision = registry
            .forward_ethernet_at(&room(), client_one.peer().peer_id(), &unknown_unicast, now)
            .unwrap();
        assert_rate_limited(&decision);

        let decision = registry
            .forward_ethernet_at(&room(), client_one.peer().peer_id(), &known_unicast, now)
            .unwrap();
        assert_eq!(decision.action(), FrameAction::Forwarded);
        assert_eq!(decision.targets(), &[client_two.peer().peer_id()]);

        let decision = registry
            .forward_ethernet_at(
                &room(),
                client_one.peer().peer_id(),
                &unknown_unicast,
                now + Duration::from_secs(1),
            )
            .unwrap();
        assert_eq!(decision.action(), FrameAction::Forwarded);
    }

    #[test]
    fn rate_limits_client_total_bandwidth_after_burst() {
        let mut registry = RoomRegistry::default();
        let client_one = registry.join(client_hello(1)).unwrap();
        let client_two = registry.join(client_hello(2)).unwrap();
        let payload = vec![0; usize::from(client_one.welcome().effective_tap_mtu())];
        let frame = ethernet_with_payload(mac(2), mac(1), ETHERTYPE_IPV4, &payload);
        let frame_len = frame.len() as u64;
        let burst_frames = CLIENT_TOTAL_BANDWIDTH_BURST_BYTES / frame_len;
        let now = Instant::now();

        assert!(burst_frames > 0);

        for _ in 0..burst_frames {
            let decision = registry
                .forward_ethernet_at(&room(), client_one.peer().peer_id(), &frame, now)
                .unwrap();
            assert_eq!(decision.action(), FrameAction::Forwarded);
            assert_eq!(decision.targets(), &[client_two.peer().peer_id()]);
        }

        let decision = registry
            .forward_ethernet_at(&room(), client_one.peer().peer_id(), &frame, now)
            .unwrap();
        assert_rate_limited(&decision);

        let decision = registry
            .forward_ethernet_at(
                &room(),
                client_two.peer().peer_id(),
                &ethernet(mac(1), mac(2)),
                now,
            )
            .unwrap();
        assert_eq!(decision.action(), FrameAction::Forwarded);
        assert_eq!(decision.targets(), &[client_one.peer().peer_id()]);

        let decision = registry
            .forward_ethernet_at(
                &room(),
                client_one.peer().peer_id(),
                &frame,
                now + Duration::from_secs(1),
            )
            .unwrap();
        assert_eq!(decision.action(), FrameAction::Forwarded);
    }

    #[test]
    fn filters_client_frames_with_forged_source_mac() {
        let mut registry = RoomRegistry::default();
        let client = registry.join(client_hello(1)).unwrap();
        let frame = ethernet(MacAddr::BROADCAST, mac(2));

        let decision = registry
            .forward_ethernet(&room(), client.peer().peer_id(), &frame)
            .unwrap();

        assert_filtered(&decision, DropReason::UnauthorizedSourceMac);
    }

    #[test]
    fn filters_invalid_source_macs_from_clients_and_gateway() {
        let mut registry = RoomRegistry::default();
        let gateway = registry.join(gateway_hello()).unwrap();
        let client = registry.join(client_hello(1)).unwrap();
        let client_frame = ethernet(MacAddr::BROADCAST, MacAddr::BROADCAST);
        let gateway_frame = ethernet(mac(1), MacAddr::ZERO);

        let client_decision = registry
            .forward_ethernet(&room(), client.peer().peer_id(), &client_frame)
            .unwrap();
        let gateway_decision = registry
            .forward_ethernet(&room(), gateway.peer().peer_id(), &gateway_frame)
            .unwrap();

        assert_filtered(&client_decision, DropReason::InvalidSourceMac);
        assert_filtered(&gateway_decision, DropReason::InvalidSourceMac);
    }

    #[test]
    fn filters_jumbo_frames() {
        let mut registry = RoomRegistry::default();
        registry.join(gateway_hello()).unwrap();
        let client = registry.join(client_hello(1)).unwrap();
        let mut frame = ethernet(MacAddr::BROADCAST, mac(1));
        frame.resize(MAX_STANDARD_ETHERNET_FRAME_LEN + 1, 0);

        let decision = registry
            .forward_ethernet(&room(), client.peer().peer_id(), &frame)
            .unwrap();

        assert_filtered(&decision, DropReason::JumboFrame);
    }

    #[test]
    fn drops_frames_above_effective_tap_mtu() {
        let mut registry = RoomRegistry::default();
        let gateway = registry.join(gateway_hello()).unwrap();
        let client = registry.join(client_hello(1)).unwrap();
        let oversized_payload = vec![0; usize::from(client.welcome().effective_tap_mtu()) + 1];
        let client_frame = ethernet_with_payload(
            MacAddr::BROADCAST,
            mac(1),
            ETHERTYPE_IPV4,
            &oversized_payload,
        );
        let gateway_frame =
            ethernet_with_payload(mac(1), physical_mac(), ETHERTYPE_IPV4, &oversized_payload);

        let client_decision = registry
            .forward_ethernet(&room(), client.peer().peer_id(), &client_frame)
            .unwrap();
        let gateway_decision = registry
            .forward_ethernet(&room(), gateway.peer().peer_id(), &gateway_frame)
            .unwrap();

        assert_dropped(&client_decision, DropReason::TapMtuExceeded);
        assert_dropped(&gateway_decision, DropReason::TapMtuExceeded);
    }

    #[test]
    fn filters_l2_control_plane_frames_from_clients_and_gateway() {
        let mut registry = RoomRegistry::default();
        let gateway = registry.join(gateway_hello()).unwrap();
        let client = registry.join(client_hello(1)).unwrap();
        let stp_destination = MacAddr::new([0x01, 0x80, 0xc2, 0, 0, 0]);
        let client_frame = ethernet_with_payload(stp_destination, mac(1), 0x0026, &[]);
        let gateway_frame =
            ethernet_with_payload(stp_destination, physical_mac(), ETHERTYPE_LLDP, &[]);

        let client_decision = registry
            .forward_ethernet(&room(), client.peer().peer_id(), &client_frame)
            .unwrap();
        let gateway_decision = registry
            .forward_ethernet(&room(), gateway.peer().peer_id(), &gateway_frame)
            .unwrap();

        assert_filtered(&client_decision, DropReason::ControlPlaneEtherType);
        assert_filtered(&gateway_decision, DropReason::ControlPlaneEtherType);
    }

    #[test]
    fn filters_remote_vlan_tagged_frames_but_allows_lan_tags() {
        let mut registry = RoomRegistry::default();
        let gateway = registry.join(gateway_hello()).unwrap();
        let client = registry.join(client_hello(1)).unwrap();
        let payload = [0, 42, 0x08, 0x00, 1, 2, 3, 4];
        let client_frame =
            ethernet_with_payload(MacAddr::BROADCAST, mac(1), ETHERTYPE_8021Q, &payload);
        let gateway_frame = ethernet_with_payload(
            MacAddr::BROADCAST,
            physical_mac(),
            ETHERTYPE_8021AD,
            &payload,
        );

        let client_decision = registry
            .forward_ethernet(&room(), client.peer().peer_id(), &client_frame)
            .unwrap();
        let gateway_decision = registry
            .forward_ethernet(&room(), gateway.peer().peer_id(), &gateway_frame)
            .unwrap();

        assert_filtered(&client_decision, DropReason::VlanTaggedFrame);
        assert_eq!(gateway_decision.action(), FrameAction::Forwarded);
        assert_eq!(gateway_decision.targets(), &[client.peer().peer_id()]);
    }

    #[test]
    fn filters_remote_dhcp_server_replies_but_allows_lan_replies() {
        let mut registry = RoomRegistry::default();
        let gateway = registry.join(gateway_hello()).unwrap();
        let client = registry.join(client_hello(1)).unwrap();
        let payload = ipv4_udp_payload(DHCPV4_SERVER_PORT, DHCPV4_CLIENT_PORT);
        let client_frame =
            ethernet_with_payload(MacAddr::BROADCAST, mac(1), ETHERTYPE_IPV4, &payload);
        let gateway_frame =
            ethernet_with_payload(MacAddr::BROADCAST, physical_mac(), ETHERTYPE_IPV4, &payload);

        let client_decision = registry
            .forward_ethernet(&room(), client.peer().peer_id(), &client_frame)
            .unwrap();
        let gateway_decision = registry
            .forward_ethernet(&room(), gateway.peer().peer_id(), &gateway_frame)
            .unwrap();

        assert_filtered(&client_decision, DropReason::DhcpServerReply);
        assert_eq!(gateway_decision.action(), FrameAction::Forwarded);
        assert_eq!(gateway_decision.targets(), &[client.peer().peer_id()]);
    }

    #[test]
    fn filters_remote_dhcpv6_server_replies_but_allows_lan_replies() {
        let mut registry = RoomRegistry::default();
        let gateway = registry.join(gateway_hello()).unwrap();
        let client = registry.join(client_hello(1)).unwrap();
        let destination = MacAddr::new([0x33, 0x33, 0, 1, 0, 2]);
        let payload =
            ipv6_udp_after_destination_options_payload(DHCPV6_SERVER_PORT, DHCPV6_CLIENT_PORT);
        let client_frame = ethernet_with_payload(destination, mac(1), ETHERTYPE_IPV6, &payload);
        let gateway_frame =
            ethernet_with_payload(destination, physical_mac(), ETHERTYPE_IPV6, &payload);

        let client_decision = registry
            .forward_ethernet(&room(), client.peer().peer_id(), &client_frame)
            .unwrap();
        let gateway_decision = registry
            .forward_ethernet(&room(), gateway.peer().peer_id(), &gateway_frame)
            .unwrap();

        assert_filtered(&client_decision, DropReason::DhcpServerReply);
        assert_eq!(gateway_decision.action(), FrameAction::Forwarded);
        assert_eq!(gateway_decision.targets(), &[client.peer().peer_id()]);
    }

    #[test]
    fn allows_remote_dhcpv4_client_requests() {
        let mut registry = RoomRegistry::default();
        let gateway = registry.join(gateway_hello()).unwrap();
        let client = registry.join(client_hello(1)).unwrap();
        let payload = ipv4_udp_payload(DHCPV4_CLIENT_PORT, DHCPV4_SERVER_PORT);
        let frame = ethernet_with_payload(MacAddr::BROADCAST, mac(1), ETHERTYPE_IPV4, &payload);

        let decision = registry
            .forward_ethernet(&room(), client.peer().peer_id(), &frame)
            .unwrap();

        assert_eq!(decision.action(), FrameAction::Forwarded);
        assert_eq!(decision.targets(), &[gateway.peer().peer_id()]);
    }

    #[test]
    fn allows_remote_dhcpv6_client_requests() {
        let mut registry = RoomRegistry::default();
        let gateway = registry.join(gateway_hello()).unwrap();
        let client = registry.join(client_hello(1)).unwrap();
        let destination = MacAddr::new([0x33, 0x33, 0, 1, 0, 2]);
        let payload = ipv6_udp_payload(DHCPV6_CLIENT_PORT, DHCPV6_SERVER_PORT);
        let frame = ethernet_with_payload(destination, mac(1), ETHERTYPE_IPV6, &payload);

        let decision = registry
            .forward_ethernet(&room(), client.peer().peer_id(), &frame)
            .unwrap();

        assert_eq!(decision.action(), FrameAction::Forwarded);
        assert_eq!(decision.targets(), &[gateway.peer().peer_id()]);
    }

    #[test]
    fn filters_remote_ipv6_fragments_but_allows_lan_fragments() {
        let mut registry = RoomRegistry::default();
        let gateway = registry.join(gateway_hello()).unwrap();
        let client = registry.join(client_hello(1)).unwrap();
        let destination = MacAddr::new([0x33, 0x33, 0, 0, 0, 1]);
        let payload = ipv6_payload(
            IPV6_NEXT_HEADER_FRAGMENT,
            &ipv6_fragment_payload(IP_PROTOCOL_UDP, b"partial upper payload"),
        );
        let client_frame = ethernet_with_payload(destination, mac(1), ETHERTYPE_IPV6, &payload);
        let gateway_frame =
            ethernet_with_payload(destination, physical_mac(), ETHERTYPE_IPV6, &payload);

        let client_decision = registry
            .forward_ethernet(&room(), client.peer().peer_id(), &client_frame)
            .unwrap();
        let gateway_decision = registry
            .forward_ethernet(&room(), gateway.peer().peer_id(), &gateway_frame)
            .unwrap();

        assert_filtered(&client_decision, DropReason::Ipv6Fragment);
        assert_eq!(gateway_decision.action(), FrameAction::Forwarded);
        assert_eq!(gateway_decision.targets(), &[client.peer().peer_id()]);
    }

    #[test]
    fn filters_remote_ipv6_router_advertisements() {
        let mut registry = RoomRegistry::default();
        registry.join(gateway_hello()).unwrap();
        let client = registry.join(client_hello(1)).unwrap();
        let destination = MacAddr::new([0x33, 0x33, 0, 0, 0, 1]);
        let frame = ethernet_with_payload(
            destination,
            mac(1),
            ETHERTYPE_IPV6,
            &ipv6_router_advertisement_payload(),
        );

        let decision = registry
            .forward_ethernet(&room(), client.peer().peer_id(), &frame)
            .unwrap();

        assert_filtered(&decision, DropReason::Ipv6RouterAdvertisement);
    }

    #[test]
    fn filters_remote_ipv6_router_advertisements_after_extension_headers() {
        let mut registry = RoomRegistry::default();
        registry.join(gateway_hello()).unwrap();
        let client = registry.join(client_hello(1)).unwrap();
        let destination = MacAddr::new([0x33, 0x33, 0, 0, 0, 1]);
        let frame = ethernet_with_payload(
            destination,
            mac(1),
            ETHERTYPE_IPV6,
            &ipv6_router_advertisement_after_destination_options_payload(),
        );

        let decision = registry
            .forward_ethernet(&room(), client.peer().peer_id(), &frame)
            .unwrap();

        assert_filtered(&decision, DropReason::Ipv6RouterAdvertisement);
    }

    #[test]
    fn allows_remote_icmpv6_that_is_not_router_advertisement() {
        let mut registry = RoomRegistry::default();
        let gateway = registry.join(gateway_hello()).unwrap();
        let client = registry.join(client_hello(1)).unwrap();
        let destination = MacAddr::new([0x33, 0x33, 0, 0, 0, 1]);
        let frame = ethernet_with_payload(
            destination,
            mac(1),
            ETHERTYPE_IPV6,
            &ipv6_echo_request_after_destination_options_payload(),
        );

        let decision = registry
            .forward_ethernet(&room(), client.peer().peer_id(), &frame)
            .unwrap();

        assert_eq!(decision.action(), FrameAction::Forwarded);
        assert_eq!(decision.targets(), &[gateway.peer().peer_id()]);
    }

    #[test]
    fn drops_malformed_frames() {
        let mut registry = RoomRegistry::default();
        let client = registry.join(client_hello(1)).unwrap();

        let decision = registry
            .forward_ethernet(&room(), client.peer().peer_id(), &[0; 4])
            .unwrap();

        assert_eq!(decision.action(), FrameAction::Dropped);
        assert_eq!(decision.drop_reason(), Some(DropReason::Malformed));
    }

    #[test]
    fn reports_unknown_ingress_peer() {
        let mut registry = RoomRegistry::default();
        registry.join(client_hello(1)).unwrap();
        let frame = ethernet(MacAddr::BROADCAST, mac(1));

        let error = registry.forward_ethernet(&room(), 99, &frame).unwrap_err();

        assert_eq!(
            error,
            ForwardingError::UnknownIngressPeer {
                room: room(),
                peer_id: 99,
            }
        );
    }
}