impl FipsPrivateMeshRuntime {
    pub(crate) async fn recv_mesh_event(&self) -> Result<Option<FipsPrivateMeshEvent>> {
        loop {
            let Some(message) = self.endpoint.recv().await else {
                return Ok(None);
            };

            if let Some(event) = self.endpoint_message_to_mesh_event(message, None).await? {
                return Ok(Some(event));
            }
        }
    }

    #[cfg(any(target_os = "linux", target_os = "macos", target_os = "windows"))]
    pub(crate) async fn recv_mesh_event_batch(
        &self,
        limit: usize,
    ) -> Result<Option<Vec<FipsPrivateMeshEvent>>> {
        let limit = limit.clamp(1, FIPS_MESH_EVENT_DRAIN_LIMIT);
        let mut messages = Vec::with_capacity(limit);
        let mut events = Vec::with_capacity(limit);
        if self
            .recv_mesh_event_batch_into(&mut messages, &mut events, limit)
            .await?
            .is_none()
        {
            return Ok(None);
        }
        Ok(Some(events))
    }

    #[cfg(any(target_os = "linux", target_os = "macos", target_os = "windows"))]
    pub(crate) async fn recv_mesh_event_batch_into(
        &self,
        messages: &mut Vec<FipsEndpointMessage>,
        events: &mut Vec<FipsPrivateMeshEvent>,
        limit: usize,
    ) -> Result<Option<usize>> {
        let limit = limit.clamp(1, FIPS_MESH_EVENT_DRAIN_LIMIT);
        events.clear();
        loop {
            let Some(_) = self.endpoint.recv_batch_into(messages, limit).await else {
                return Ok(None);
            };
            let now = Some(unix_timestamp());
            events.reserve(messages.len());
            for message in messages.drain(..) {
                if let Some(event) = self.endpoint_message_to_mesh_event(message, now).await? {
                    events.push(event);
                }
            }
            if !events.is_empty() {
                return Ok(Some(events.len()));
            }
        }
    }

    #[cfg(any(target_os = "linux", target_os = "macos"))]
    fn recv_mesh_event_batch_blocking_into(
        &self,
        messages: &mut Vec<FipsEndpointMessage>,
        events: &mut Vec<FipsPrivateMeshEvent>,
        limit: usize,
        stop: &AtomicBool,
    ) -> Result<Option<usize>> {
        let limit = limit.clamp(1, FIPS_MESH_EVENT_DRAIN_LIMIT);
        events.clear();
        loop {
            if stop.load(Ordering::Acquire) {
                return Ok(None);
            }

            let Some(_) = self.endpoint.blocking_recv_batch_into(messages, limit) else {
                return Ok(None);
            };
            let now = Some(unix_timestamp());
            events.reserve(messages.len());
            for message in messages.drain(..) {
                if stop.load(Ordering::Acquire) {
                    return Ok(None);
                }
                if let Some(event) = self.endpoint_message_to_mesh_event_blocking(message, now)? {
                    events.push(event);
                }
            }
            if !events.is_empty() {
                return Ok(Some(events.len()));
            }
        }
    }

    #[cfg(any(target_os = "linux", target_os = "macos"))]
    fn recv_mesh_event_batch_blocking_for_each(
        &self,
        limit: usize,
        stop: &AtomicBool,
        mut handle_event: impl FnMut(FipsPrivateMeshEvent) -> bool,
    ) -> Result<Option<usize>> {
        let limit = limit.clamp(1, FIPS_MESH_EVENT_DRAIN_LIMIT);
        loop {
            if stop.load(Ordering::Acquire) {
                return Ok(None);
            }

            let now = Some(unix_timestamp());
            let mut emitted = 0usize;
            let mut pending_error: Option<anyhow::Error> = None;
            let received = self.endpoint.blocking_recv_batch_for_each(limit, |message| {
                if stop.load(Ordering::Acquire) {
                    return false;
                }
                let outcome = match self.endpoint_message_to_mesh_event_outcome(message, now) {
                    Ok(outcome) => outcome,
                    Err(error) => {
                        pending_error = Some(error);
                        return false;
                    }
                };
                if let Some(reply) = outcome.reply
                    && let Err(error) = self.endpoint.blocking_send_to_peer(reply.peer, reply.data)
                {
                    pending_error = Some(error.into());
                    return false;
                }
                if let Some(event) = outcome.event {
                    emitted = emitted.saturating_add(1);
                    return handle_event(event);
                }
                true
            });

            if let Some(error) = pending_error {
                return Err(error);
            }
            if stop.load(Ordering::Acquire) {
                return Ok(None);
            }
            if received.is_none() {
                return Ok(None);
            }
            if emitted > 0 {
                return Ok(Some(emitted));
            }
        }
    }

    #[cfg(any(target_os = "linux", target_os = "macos"))]
    fn wake_blocking_mesh_recv(&self) {
        let npub = self.endpoint.npub().to_string();
        let _ = self.endpoint.blocking_send(npub, Vec::new());
    }

    #[cfg(any(target_os = "linux", target_os = "macos", target_os = "windows"))]
    pub(crate) async fn try_recv_mesh_event(&self) -> Result<Option<FipsPrivateMeshEvent>> {
        self.try_recv_mesh_event_with_timestamp(None).await
    }

    #[cfg(any(target_os = "linux", target_os = "macos", target_os = "windows"))]
    async fn try_recv_mesh_event_with_timestamp(
        &self,
        now: Option<u64>,
    ) -> Result<Option<FipsPrivateMeshEvent>> {
        loop {
            let Some(message) = self.endpoint.try_recv() else {
                return Ok(None);
            };

            if let Some(event) = self.endpoint_message_to_mesh_event(message, now).await? {
                return Ok(Some(event));
            }
        }
    }

    async fn endpoint_message_to_mesh_event(
        &self,
        message: FipsEndpointMessage,
        now: Option<u64>,
    ) -> Result<Option<FipsPrivateMeshEvent>> {
        let outcome = self.endpoint_message_to_mesh_event_outcome(message, now)?;
        if let Some(reply) = outcome.reply
            && let Err(error) = self.endpoint.send_to_peer(reply.peer, reply.data).await
        {
            eprintln!("fips: failed to reply to peer ping: {error}");
        }
        Ok(outcome.event)
    }

    fn endpoint_message_to_mesh_event_blocking(
        &self,
        message: FipsEndpointMessage,
        now: Option<u64>,
    ) -> Result<Option<FipsPrivateMeshEvent>> {
        let outcome = self.endpoint_message_to_mesh_event_outcome(message, now)?;
        if let Some(reply) = outcome.reply
            && let Err(error) = self.endpoint.blocking_send_to_peer(reply.peer, reply.data)
        {
            eprintln!("fips: failed to reply to peer ping: {error}");
        }
        Ok(outcome.event)
    }

    fn endpoint_message_to_mesh_event_outcome(
        &self,
        message: FipsEndpointMessage,
        now: Option<u64>,
    ) -> Result<FipsEndpointMessageOutcome> {
        if let Some(frame) = self.decode_endpoint_control_frame(&message)? {
            let source_pubkey = {
                let mesh = self.mesh.load();
                control_frame_source_pubkey(&mesh, message.source_peer, &frame)
            };
            let Some(source_pubkey) = source_pubkey else {
                return Ok(FipsEndpointMessageOutcome::none());
            };
            let now = now.unwrap_or_else(unix_timestamp);
            self.note_control_rx(&source_pubkey, message.data.len(), now)?;
            match frame {
                FipsControlFrame::Ping {
                    network_id,
                    sent_at,
                } => {
                    let reply = FipsControlFrame::Pong {
                        network_id,
                        sent_at,
                        replied_at: now,
                    };
                    let encoded = encode_fips_control_frame(&reply)?;
                    return Ok(FipsEndpointMessageOutcome::event_with_reply(
                        FipsPrivateMeshEvent::Presence {
                            participant_pubkey: source_pubkey,
                            last_seen_at: now,
                        },
                        message.source_peer,
                        encoded,
                    ));
                }
                FipsControlFrame::Pong { sent_at, .. } => {
                    self.note_pong(&source_pubkey, sent_at)?;
                    return Ok(FipsEndpointMessageOutcome::event(
                        FipsPrivateMeshEvent::Presence {
                            participant_pubkey: source_pubkey,
                            last_seen_at: now,
                        },
                    ));
                }
                FipsControlFrame::JoinRequest {
                    requested_at,
                    request,
                } => {
                    return Ok(FipsEndpointMessageOutcome::event(
                        FipsPrivateMeshEvent::JoinRequest {
                            sender_pubkey: source_pubkey,
                            requested_at,
                            request,
                        },
                    ));
                }
                FipsControlFrame::Roster {
                    network_id,
                    roster,
                    signed_roster,
                } => {
                    return Ok(FipsEndpointMessageOutcome::event(
                        FipsPrivateMeshEvent::Roster {
                            sender_pubkey: source_pubkey,
                            network_id,
                            roster,
                            signed_roster,
                        },
                    ));
                }
                FipsControlFrame::Capabilities {
                    network_id,
                    capabilities,
                } => {
                    self.record_peer_capabilities(&source_pubkey, &capabilities, now)?;
                    return Ok(FipsEndpointMessageOutcome::event(
                        FipsPrivateMeshEvent::Capabilities {
                            sender_pubkey: source_pubkey,
                            network_id,
                            capabilities,
                        },
                    ));
                }
                FipsControlFrame::Fragment { .. } => {
                    return Ok(FipsEndpointMessageOutcome::none());
                }
            }
        }

        let data_len = message.data.len();
        let source_node_addr = *message.source_peer.node_addr();
        let mesh = self.mesh.load();
        if let Some(packet) = mesh.receive_endpoint_data_owned_with_source_node_addr(
            source_node_addr.as_bytes(),
            message.data,
        ) {
            let now = now.unwrap_or_else(unix_timestamp);
            self.note_data_rx(
                packet.source_pubkey,
                packet.source_pubkey_bytes,
                data_len,
                now,
            )?;
            return Ok(FipsEndpointMessageOutcome::event(
                FipsPrivateMeshEvent::Packet(packet.bytes),
            ));
        }

        Ok(FipsEndpointMessageOutcome::none())
    }

    fn decode_endpoint_control_frame(
        &self,
        message: &FipsEndpointMessage,
    ) -> Result<Option<FipsControlFrame>> {
        let Some(frame) = decode_fips_control_frame(&message.data)? else {
            return Ok(None);
        };
        let FipsControlFrame::Fragment {
            id,
            index,
            total,
            data,
        } = frame
        else {
            return Ok(Some(frame));
        };

        let source_key = *message.source_peer.node_addr().as_bytes();
        let Some(reassembled) = self
            .control_fragments
            .lock()
            .map_err(|_| anyhow!("FIPS control fragment buffer lock poisoned"))?
            .push(source_key, id, index, total, data, unix_timestamp())?
        else {
            return Ok(None);
        };
        decode_fips_control_frame(&reassembled)
    }

    #[cfg(test)]
    pub(crate) async fn recv_tunnel_packet(&self) -> Result<Option<Vec<u8>>> {
        loop {
            match self.recv_mesh_event().await? {
                Some(FipsPrivateMeshEvent::Packet(packet)) => return Ok(Some(packet)),
                Some(_) => {}
                None => return Ok(None),
            }
        }
    }

}
