//! 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;

use lanparty_ctrl::{
    ControlError, EndpointHello, PeerInfo, Reject, RejectReason, Role, RoomCode, ServerWelcome,
};
use lanparty_obs::{DropReason, FrameAction};
use lanparty_proto::{ETHERNET_HEADER_LEN, EthernetFrame, MacAddr, recommended_tap_mtu};
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 ETHERTYPE_IPV4: u16 = 0x0800;
const ETHERTYPE_EAPOL: u16 = 0x888e;
const ETHERTYPE_SLOW_PROTOCOLS: u16 = 0x8809;
const ETHERTYPE_LLDP: u16 = 0x88cc;
const ETHERTYPE_IPV6: u16 = 0x86dd;
const IPV4_PROTOCOL_UDP: u8 = 17;
const IPV6_NEXT_HEADER_ICMPV6: u8 = 58;
const DHCP_SERVER_PORT: u16 = 67;
const DHCP_CLIENT_PORT: u16 = 68;
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>,
}

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

#[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]
    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(
        &self,
        room: &RoomCode,
        ingress_peer_id: u32,
        frame_bytes: &[u8],
    ) -> Result<ForwardingDecision, ForwardingError> {
        self.rooms
            .get(room)
            .ok_or_else(|| ForwardingError::UnknownRoom(room.clone()))?
            .forward_ethernet(room, ingress_peer_id, frame_bytes)
    }

    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<PeerInfo>,
    clients: HashMap<u32, PeerInfo>,
    clients_by_mac: HashMap<MacAddr, u32>,
}

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

    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 welcome =
            ServerWelcome::new(self.room_id, peer_id, effective_tap_mtu).map_err(|error| {
                Reject::new(
                    RejectReason::InternalError,
                    format!("welcome failed: {error}"),
                )
            })?;

        match peer.role() {
            Role::Gateway => {
                self.gateway = Some(peer.clone());
            }
            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(), peer.clone());
            }
        }

        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().cloned().collect();
        clients.sort_by_key(PeerInfo::peer_id);

        RoomSnapshot {
            room_id: self.room_id,
            effective_tap_mtu: self.effective_tap_mtu.unwrap_or_default(),
            gateway: self.gateway.clone(),
            clients,
        }
    }

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

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

        Ok(peer)
    }

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

    fn forward_ethernet(
        &self,
        room_code: &RoomCode,
        ingress_peer_id: u32,
        frame_bytes: &[u8],
    ) -> 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 frame = match EthernetFrame::parse(frame_bytes) {
            Ok(frame) => frame,
            Err(_) => return Ok(ForwardingDecision::dropped(DropReason::Malformed)),
        };

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

        if let Some(drop_reason) = safety_drop_reason(ingress.role(), frame) {
            return Ok(ForwardingDecision::dropped(drop_reason));
        }

        let targets = if frame.destination().is_multicast() {
            self.all_peer_ids_except(ingress_peer_id)
        } else if let Some(client_peer_id) = self.clients_by_mac.get(&frame.destination()) {
            if *client_peer_id == ingress_peer_id {
                Vec::new()
            } else {
                vec![*client_peer_id]
            }
        } else {
            self.all_peer_ids_except(ingress_peer_id)
        };

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

        Ok(ForwardingDecision::forwarded(targets))
    }

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

    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.peer_id() != ingress_peer_id
        {
            peer_ids.push(gateway.peer_id());
        }

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

fn safety_drop_reason(ingress_role: Role, frame: EthernetFrame<'_>) -> Option<DropReason> {
    if frame.is_jumbo() {
        return Some(DropReason::JumboFrame);
    }

    if is_control_plane_frame(frame) {
        return Some(DropReason::ControlPlaneEtherType);
    }

    if ingress_role == Role::Client && is_dhcp_server_reply(frame) {
        return Some(DropReason::DhcpServerReply);
    }

    if ingress_role == Role::Client && is_ipv6_router_advertisement(frame) {
        return Some(DropReason::Ipv6RouterAdvertisement);
    }

    None
}

fn is_control_plane_frame(frame: EthernetFrame<'_>) -> bool {
    matches!(
        frame.ethertype_or_len(),
        ETHERTYPE_EAPOL | ETHERTYPE_SLOW_PROTOCOLS | ETHERTYPE_LLDP
    ) || is_link_local_control_destination(frame.destination())
}

fn is_link_local_control_destination(mac: MacAddr) -> bool {
    let [a, b, c, d, e, f] = mac.octets();

    [a, b, c, d, e] == [0x01, 0x80, 0xc2, 0x00, 0x00] && f <= 0x0f
}

fn is_dhcp_server_reply(frame: EthernetFrame<'_>) -> bool {
    let bytes = frame.bytes();
    if frame.ethertype_or_len() != ETHERTYPE_IPV4 || bytes.len() < ETHERNET_HEADER_LEN + 20 {
        return false;
    }

    let ipv4 = &bytes[ETHERNET_HEADER_LEN..];
    let version = ipv4[0] >> 4;
    let header_len = usize::from(ipv4[0] & 0x0f) * 4;
    if version != 4
        || header_len < 20
        || ipv4.len() < header_len + 8
        || ipv4[9] != IPV4_PROTOCOL_UDP
    {
        return false;
    }

    let udp = &ipv4[header_len..];
    let source_port = u16::from_be_bytes([udp[0], udp[1]]);
    let destination_port = u16::from_be_bytes([udp[2], udp[3]]);

    source_port == DHCP_SERVER_PORT && destination_port == DHCP_CLIENT_PORT
}

fn is_ipv6_router_advertisement(frame: EthernetFrame<'_>) -> bool {
    let bytes = frame.bytes();
    if frame.ethertype_or_len() != ETHERTYPE_IPV6 || bytes.len() < ETHERNET_HEADER_LEN + 40 + 1 {
        return false;
    }

    let ipv6 = &bytes[ETHERNET_HEADER_LEN..];
    let version = ipv6[0] >> 4;
    let next_header = ipv6[6];

    version == 6
        && next_header == IPV6_NEXT_HEADER_ICMPV6
        && ipv6[40] == ICMPV6_ROUTER_ADVERTISEMENT
}

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 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] = IPV4_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 ipv6_router_advertisement_payload() -> Vec<u8> {
        let mut packet = vec![0; 41];
        packet[0] = 0x60;
        packet[6] = IPV6_NEXT_HEADER_ICMPV6;
        packet[40] = ICMPV6_ROUTER_ADVERTISEMENT;
        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());
    }

    #[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_eq!(client.peer().role(), Role::Client);
        assert_eq!(client.welcome().peer_id(), 2);
        assert_eq!(snapshot.gateway().unwrap().peer_id(), 1);
        assert_eq!(snapshot.clients().len(), 1);
        assert_eq!(snapshot.effective_tap_mtu(), 1200);
    }

    #[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!(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.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 = registry.join(client_hello(1)).unwrap();
        let frame = ethernet(MacAddr::new([0x00, 1, 2, 3, 4, 5]), mac(1));

        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()]);
        assert_eq!(decision.drop_reason(), None);
    }

    #[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 drops_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_eq!(decision.action(), FrameAction::Dropped);
        assert_eq!(
            decision.drop_reason(),
            Some(DropReason::UnauthorizedSourceMac)
        );
        assert!(decision.targets().is_empty());
    }

    #[test]
    fn drops_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_dropped(&decision, DropReason::JumboFrame);
    }

    #[test]
    fn drops_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_dropped(&client_decision, DropReason::ControlPlaneEtherType);
        assert_dropped(&gateway_decision, DropReason::ControlPlaneEtherType);
    }

    #[test]
    fn drops_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(DHCP_SERVER_PORT, DHCP_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_dropped(&client_decision, DropReason::DhcpServerReply);
        assert_eq!(gateway_decision.action(), FrameAction::Forwarded);
        assert_eq!(gateway_decision.targets(), &[client.peer().peer_id()]);
    }

    #[test]
    fn drops_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_dropped(&decision, DropReason::Ipv6RouterAdvertisement);
    }

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