offline_protocol_transport/

internet.rs

//! Internet transport (TCP/WebSocket relay) queue engine.
//!
//! No sockets are opened here. The platform side owns the actual TCP or
//! WebSocket connection to the relay server; the Rust side manages the
//! send/receive queues, metrics, heartbeat bookkeeping, and the
//! confirmation loop.
//!
//! The bridge contract: the platform reports connectivity via
//! [`InternetTransport::on_status_changed`] (nothing else makes this
//! transport `Available`), polls [`InternetTransport::get_next_message`]
//! for outbound wire bytes — unlike the other transports there is no
//! messages-available callback — reports outcomes via
//! [`InternetTransport::confirm_sent`] /
//! [`InternetTransport::report_send_failure`], and injects inbound bytes
//! via [`InternetTransport::on_data_received`].

use crate::constants::{
    INTERNET_DEFAULT_SERVER_ADDRESS, INTERNET_PENDING_CONFIRMATION_TIMEOUT_SECS,
};
use crate::{Result, Transport, TransportMetrics, TransportStatus, TransportType};
use offline_protocol_core::{Message, MutexExt};
use std::collections::{HashMap, VecDeque};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use crate::common::recalculate_delivery_ratios;

/// Internet transport configuration.
///
/// Connection lifecycle (timeouts, reconnection, keepalive) is owned by the
/// platform bridge that manages the actual socket — this config carries only
/// what the Rust side itself consumes.
#[derive(Debug, Clone)]
pub struct InternetConfig {
    /// Server address (WebSocket URL or TCP address)
    pub server_address: String,
}

impl Default for InternetConfig {
    fn default() -> Self {
        Self {
            server_address: INTERNET_DEFAULT_SERVER_ADDRESS.to_string(),
        }
    }
}

/// Internet transport implementation.
///
/// Queue engine for a platform-managed TCP/WebSocket connection to a
/// central relay server. Useful for hybrid online/offline scenarios.
///
/// ## Lock ordering
///
/// When acquiring more than one lock in a single scope, follow this order to
/// prevent deadlocks:
///
/// 1. `status`
/// 2. `pending_confirmation`
/// 3. `send_queue`
/// 4. `metrics`
/// 5. `receive_queue`
///
/// Not every method needs all of these — the ordering only matters when
/// multiple locks are held simultaneously.
pub struct InternetTransport {
    /// Local device ID
    device_id: String,
    /// Configuration
    config: InternetConfig,
    /// Transport status
    status: Arc<Mutex<TransportStatus>>,
    /// Received message queue
    receive_queue: Arc<Mutex<VecDeque<Message>>>,
    /// Send queue — messages waiting to be polled by the platform via `get_next_message()`
    send_queue: Arc<Mutex<VecDeque<Message>>>,
    /// Messages dequeued by the platform but not yet confirmed as sent.
    /// Key: message_id string, Value: when the message was dequeued.
    ///
    /// Bounded by: (a) `drain_expired_pending()` which runs on every
    /// `get_next_message()` poll, and (b) `fail_all_pending()` which runs
    /// on transport disconnect. If the platform stops polling *and* the
    /// transport status stays `Available`, entries accumulate up to the
    /// number of messages enqueued during that window. Under normal
    /// operation (platform polls at ≥1 Hz) the map stays small.
    pending_confirmation: Arc<Mutex<HashMap<String, Instant>>>,
    /// Transport metrics
    metrics: Arc<Mutex<TransportMetrics>>,
}

impl InternetTransport {
    /// Creates a new internet transport.
    pub fn new(device_id: impl Into<String>) -> Self {
        Self::with_config(device_id, InternetConfig::default())
    }

    /// Creates a new internet transport with custom configuration.
    pub fn with_config(device_id: impl Into<String>, config: InternetConfig) -> Self {
        Self {
            device_id: device_id.into(),
            config,
            status: Arc::new(Mutex::new(TransportStatus::Unavailable)),
            receive_queue: Arc::new(Mutex::new(VecDeque::new())),
            send_queue: Arc::new(Mutex::new(VecDeque::new())),
            pending_confirmation: Arc::new(Mutex::new(HashMap::new())),
            metrics: Arc::new(Mutex::new(TransportMetrics::default())),
        }
    }

    /// Gets the local device ID.
    pub fn device_id(&self) -> &str {
        &self.device_id
    }

    /// Gets the configuration.
    pub fn config(&self) -> &InternetConfig {
        &self.config
    }

    /// Called when a message is received from the server.
    pub fn on_message_received(&self, message: Message) {
        crate::common::on_message_received(&self.receive_queue, message);
    }

    /// Like [`on_message_received`](Self::on_message_received), but attaches a
    /// transport-verified `peer_id` to the message.
    ///
    /// # Parameters
    ///
    /// * `peer_id` — The **user-level identifier** (i.e., the remote peer's
    ///   `UserId` string) that the relay server has authenticated for this
    ///   connection (e.g. via WebSocket session token). This is **not** the raw
    ///   transport address (IP, session token, etc.). The protocol layer
    ///   strict-matches this value against `message.sender` for control
    ///   frames claiming direct origin (`hop_count == 0`); mesh-relayed
    ///   frames (`hop_count > 0`) are exempt because there the authenticated
    ///   uploader is the carrier, not the origin.
    pub fn on_message_received_from(&self, message: Message, peer_id: String) {
        crate::common::on_message_received_from(&self.receive_queue, message, peer_id);
    }

    /// Serializes a message to JSON bytes.
    pub fn serialize_message(&self, message: &Message) -> Result<Vec<u8>> {
        crate::common::serialize_message(message)
    }

    /// Drains messages that have been pending confirmation longer than the timeout,
    /// recording each as a failure.
    fn drain_expired_pending(&self) {
        let timeout = Duration::from_secs(INTERNET_PENDING_CONFIRMATION_TIMEOUT_SECS);
        let now = Instant::now();
        let mut expired_count: u32 = 0;

        {
            let mut pending = self.pending_confirmation.lock_or_recover();
            pending.retain(|_id, enqueued_at| {
                if now.duration_since(*enqueued_at) >= timeout {
                    expired_count += 1;
                    false
                } else {
                    true
                }
            });
        }

        if expired_count > 0 {
            tracing::warn!(
                expired_count = expired_count,
                "Pending confirmations expired, recorded as failures"
            );
            self.record_bulk_failures(expired_count);
        }
    }

    /// Fails all pending confirmations immediately.
    ///
    /// Called when the transport disconnects or the transport is stopped so
    /// that metrics reflect the loss right away rather than waiting for the
    /// per-message expiry timeout.
    fn fail_all_pending(&self) {
        let count = {
            let mut pending = self.pending_confirmation.lock_or_recover();
            let count = u32::try_from(pending.len()).unwrap_or(u32::MAX);
            pending.clear();
            count
        };

        if count > 0 {
            tracing::warn!(
                count = count,
                "Failing all pending confirmations due to transport disconnect"
            );
            self.record_bulk_failures(count);
        }
    }

    /// Records `count` delivery failures in transport metrics.
    fn record_bulk_failures(&self, count: u32) {
        let mut metrics = self.metrics.lock_or_recover();
        metrics.failure_count = metrics.failure_count.saturating_add(count);
        recalculate_delivery_ratios(&mut metrics);
    }

    /// Returns the count of messages currently awaiting confirmation from the platform.
    pub fn pending_confirmation_count(&self) -> usize {
        self.pending_confirmation.lock_or_recover().len()
    }

    /// Updates transport metrics while preserving confirmation-loop delivery counts.
    ///
    /// The confirmation loop (`confirm_sent` / `report_send_failure`) owns
    /// `success_count`, `failure_count`, `delivery_ratio`, and `drop_rate`.
    /// This method only copies externally-managed fields from `incoming`,
    /// so adding a new confirmation-loop field doesn't require updating this
    /// function — it's preserved by default.
    pub fn update_metrics(&self, incoming: TransportMetrics) {
        let mut current = self.metrics.lock_or_recover();
        current.rssi = incoming.rssi;
        current.latency_ms = incoming.latency_ms;
        current.bandwidth_bps = incoming.bandwidth_bps;
        current.congestion = incoming.congestion;
        current.queue_depth = incoming.queue_depth;
        current.battery_level = incoming.battery_level;
        current.is_charging = incoming.is_charging;
        current.relay_connection_count = incoming.relay_connection_count;
        current.is_active_relay = incoming.is_active_relay;
        current.average_hop_count = incoming.average_hop_count;
        current.energy_cost = incoming.energy_cost;
    }
}

impl Transport for InternetTransport {
    fn as_any(&self) -> &dyn std::any::Any {
        self
    }

    fn transport_type(&self) -> TransportType {
        TransportType::Internet
    }

    fn status(&self) -> TransportStatus {
        *self.status.lock_or_recover()
    }

    fn metrics(&self) -> TransportMetrics {
        self.metrics.lock_or_recover().clone()
    }

    fn send(&self, message: &Message) -> Result<()> {
        // Check status
        if self.status() != TransportStatus::Available {
            return Err(crate::Error::TransportNotAvailable(
                "Internet transport is not available".to_string(),
            ));
        }

        // Add to send queue
        let mut queue = self.send_queue.lock_or_recover();
        queue.push_back(message.clone());

        // Update metrics
        let mut metrics = self.metrics.lock_or_recover();
        metrics.queue_depth = queue.len();
        metrics.congestion = ((metrics.queue_depth as f32) / 25.0).clamp(0.0, 1.0);

        Ok(())
    }

    fn receive(&self) -> Result<Option<Message>> {
        let mut queue = self.receive_queue.lock_or_recover();
        Ok(queue.pop_front())
    }

    fn start(&self) -> Result<()> {
        // Status will be updated by platform implementation via on_status_changed()
        // Platform code should establish connection and call on_status_changed()
        Ok(())
    }

    fn stop(&self) -> Result<()> {
        self.fail_all_pending();
        *self.status.lock_or_recover() = TransportStatus::Disconnected;
        Ok(())
    }

    /// Called when connection status changes.
    ///
    /// EDGE CASE HANDLING:
    /// - Messages in send_queue are preserved during disconnection
    /// - They will be sent when transport becomes available again
    fn on_status_changed(&self, status: TransportStatus) {
        let previous_status = {
            let mut guard = self.status.lock_or_recover();
            let prev = *guard;
            *guard = status;
            prev
        };

        if status == TransportStatus::Available {
            let queue_len = self.send_queue.lock_or_recover().len();

            if queue_len > 0 {
                tracing::info!(
                    pending_messages = queue_len,
                    "Internet transport available, {} messages pending in queue",
                    queue_len
                );
            }
        } else if previous_status == TransportStatus::Available
            && status != TransportStatus::Available
        {
            // Fail all pending confirmations — the connection is gone so the
            // platform can no longer report outcomes for in-flight messages.
            self.fail_all_pending();

            let queue_len = self.send_queue.lock_or_recover().len();
            if queue_len > 0 {
                tracing::warn!(
                    pending_messages = queue_len,
                    new_status = ?status,
                    "Internet transport disconnected with {} messages in queue (will retry)",
                    queue_len
                );
            }
        }
    }

    /// Called when raw data is received (platform callback).
    ///
    /// This deserializes the message and queues it.
    fn on_data_received(&self, data: Vec<u8>) -> Result<()> {
        crate::common::on_data_received(&self.receive_queue, data)
    }

    /// Like [`Transport::on_data_received`], but attaches a
    /// transport-verified `peer_id` to the deserialized message.
    ///
    /// This is the production ingest path for relay traffic: the FFI layer
    /// routes every inbound relay frame with a non-empty authenticated
    /// sender through here.
    ///
    /// # Parameters
    ///
    /// * `peer_id` — The **user-level identifier** (i.e., the remote peer's
    ///   `UserId` string) that the relay server has authenticated for this
    ///   connection. This is **not** the raw transport address (IP, session
    ///   token, etc.). The protocol layer strict-matches this value against
    ///   `message.sender` for control frames claiming direct origin
    ///   (`hop_count == 0`); mesh-relayed frames (`hop_count > 0`) are
    ///   exempt because there the authenticated uploader is the carrier,
    ///   not the origin.
    fn on_data_received_from(&self, data: Vec<u8>, peer_id: String) -> Result<()> {
        crate::common::on_data_received_from(&self.receive_queue, data, peer_id)
    }

    /// Gets the next message to send (for platform implementation).
    ///
    /// Returns the message_id and serialized bytes, or None if no messages to send.
    /// The message enters the pending-confirmation state until the platform calls
    /// `confirm_sent()` or `report_send_failure()`.
    ///
    /// Also drains any pending confirmations that have exceeded the timeout,
    /// recording them as failures to keep DORS metrics accurate.
    fn get_next_message(&self) -> Result<Option<(String, Vec<u8>)>> {
        // Expire stale pending confirmations before processing the next message.
        // This is the only call site — it runs each time the platform polls, which
        // is frequent enough to bound the pending map and keep metrics fresh.
        self.drain_expired_pending();

        let (message, queue_depth) = {
            let mut queue = self.send_queue.lock_or_recover();
            match queue.pop_front() {
                Some(m) => {
                    let depth = queue.len();
                    (m, depth)
                }
                None => return Ok(None),
            }
        };

        let message_id = message.id.as_str().to_string();

        // Serialize before inserting into pending_confirmation so a
        // serialization failure does not orphan an entry that the platform
        // can never confirm or fail.
        let data = self.serialize_message(&message)?;

        {
            let mut pending = self.pending_confirmation.lock_or_recover();
            pending.insert(message_id.clone(), Instant::now());
        }

        {
            let mut metrics = self.metrics.lock_or_recover();
            metrics.queue_depth = queue_depth;
        }

        Ok(Some((message_id, data)))
    }

    /// Platform confirms that a message was successfully sent over the wire (e.g., WebSocket).
    ///
    /// This updates transport metrics to reflect real delivery outcomes,
    /// enabling DORS to make accurate routing decisions.
    fn confirm_sent(&self, message_id: &str) {
        let was_pending = {
            let mut pending = self.pending_confirmation.lock_or_recover();
            pending.remove(message_id).is_some()
        };

        if was_pending {
            let mut metrics = self.metrics.lock_or_recover();
            metrics.success_count = metrics.success_count.saturating_add(1);
            recalculate_delivery_ratios(&mut metrics);
            tracing::trace!(
                message_id = message_id,
                success_count = metrics.success_count,
                "confirm_sent: recorded success"
            );
        } else {
            tracing::debug!(
                message_id = message_id,
                "confirm_sent: unknown or already-resolved message"
            );
        }
    }

    /// Platform reports that a message failed to send (e.g., WebSocket error).
    fn report_send_failure(&self, message_id: &str) {
        let was_pending = {
            let mut pending = self.pending_confirmation.lock_or_recover();
            pending.remove(message_id).is_some()
        };

        if was_pending {
            let mut metrics = self.metrics.lock_or_recover();
            metrics.failure_count = metrics.failure_count.saturating_add(1);
            recalculate_delivery_ratios(&mut metrics);
            tracing::trace!(
                message_id = message_id,
                failure_count = metrics.failure_count,
                "report_send_failure: recorded failure"
            );
        } else {
            tracing::debug!(
                message_id = message_id,
                "report_send_failure: unknown or already-resolved message"
            );
        }
    }
}

/// Internet transport builder for configuration.
pub struct InternetTransportBuilder {
    device_id: String,
    config: InternetConfig,
}

impl InternetTransportBuilder {
    /// Creates a new builder.
    pub fn new(device_id: impl Into<String>) -> Self {
        Self {
            device_id: device_id.into(),
            config: InternetConfig::default(),
        }
    }

    /// Sets the server address.
    pub fn server_address(mut self, address: impl Into<String>) -> Self {
        self.config.server_address = address.into();
        self
    }

    /// Builds the transport.
    pub fn build(self) -> InternetTransport {
        InternetTransport::with_config(self.device_id, self.config)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::constants::DEFAULT_MAX_MESSAGE_SIZE;
    use offline_protocol_core::{AppId, UserId, WireCodec};

    fn create_test_message() -> Message {
        Message::new(
            UserId::new("alice").unwrap(),
            UserId::new("bob").unwrap(),
            AppId::new("test").unwrap(),
            "Test message",
        )
    }

    /// The internet transport is the relay path, and its platform bridge treats
    /// the payload as UTF-8 text. It must therefore always emit JSON — even when
    /// the message carries a `BinaryV1` stamp from per-peer negotiation (only the
    /// mesh transports honor that stamp). This locks the "internet relay stays
    /// JSON" invariant against a refactor that swaps `serialize_message` for the
    /// codec-aware `serialize_message_with`.
    #[test]
    fn serialize_stays_json_even_when_message_stamped_binary() {
        let transport = InternetTransport::new("test-device");
        let mut message = create_test_message();
        message.set_wire_codec(WireCodec::BinaryV1);

        let data = transport.serialize_message(&message).unwrap();

        assert_eq!(
            data.first(),
            Some(&b'{'),
            "internet transport must emit JSON regardless of the wire-codec stamp"
        );
        // And the emitted bytes must still round-trip through the JSON decoder.
        assert_eq!(transport.deserialize_message(&data).unwrap().id, message.id);
    }

    #[test]
    fn test_internet_transport_creation() {
        let transport = InternetTransport::new("test-device");
        assert_eq!(transport.device_id(), "test-device");
        assert_eq!(transport.transport_type(), TransportType::Internet);
        assert_eq!(transport.status(), TransportStatus::Unavailable);
    }

    #[test]
    fn test_builder() {
        let transport = InternetTransportBuilder::new("test-device")
            .server_address("ws://example.com:8080")
            .build();

        assert_eq!(transport.config().server_address, "ws://example.com:8080");
    }

    #[test]
    fn test_send_receive() {
        let transport = InternetTransport::new("test-device");

        // Mark as available
        transport.on_status_changed(TransportStatus::Available);

        // Send message
        let message = create_test_message();
        let message_id = message.id.as_str().to_string();
        assert!(transport.send(&message).is_ok());

        // Should have message in queue; get_next_message returns (id, bytes)
        let next = transport.get_next_message().unwrap();
        assert!(next.is_some());
        let (returned_id, data) = next.unwrap();
        assert_eq!(returned_id, message_id);
        assert!(!data.is_empty());

        // Message should now be in pending confirmation
        assert_eq!(transport.pending_confirmation_count(), 1);

        // Confirm sent — should update success metrics
        transport.confirm_sent(&returned_id);
        assert_eq!(transport.pending_confirmation_count(), 0);
        let metrics = transport.metrics();
        assert_eq!(metrics.success_count, 1);
    }

    #[test]
    fn test_send_failure_tracking() {
        let transport = InternetTransport::new("test-device");
        transport.on_status_changed(TransportStatus::Available);

        let message = create_test_message();
        assert!(transport.send(&message).is_ok());

        let (msg_id, _data) = transport.get_next_message().unwrap().unwrap();
        assert_eq!(transport.pending_confirmation_count(), 1);

        // Report failure
        transport.report_send_failure(&msg_id);
        assert_eq!(transport.pending_confirmation_count(), 0);
        let metrics = transport.metrics();
        assert_eq!(metrics.failure_count, 1);
        assert_eq!(metrics.success_count, 0);
    }

    #[test]
    fn test_pending_expiry_on_drain() {
        let transport = InternetTransport::new("test-device");
        transport.on_status_changed(TransportStatus::Available);

        let message = create_test_message();
        assert!(transport.send(&message).is_ok());

        let (msg_id, _data) = transport.get_next_message().unwrap().unwrap();
        assert_eq!(transport.pending_confirmation_count(), 1);

        // Backdate the pending entry so it appears expired.
        {
            let mut pending = transport.pending_confirmation.lock().unwrap();
            let expired_time = Instant::now()
                - Duration::from_secs(INTERNET_PENDING_CONFIRMATION_TIMEOUT_SECS + 1);
            pending.insert(msg_id.clone(), expired_time);
        }

        // Calling get_next_message (even with an empty queue) triggers drain.
        let next = transport.get_next_message().unwrap();
        assert!(next.is_none());

        // The expired entry should have been drained and counted as a failure.
        assert_eq!(transport.pending_confirmation_count(), 0);
        let metrics = transport.metrics();
        assert_eq!(metrics.failure_count, 1);
        assert_eq!(metrics.success_count, 0);
    }

    #[test]
    fn test_fail_all_pending_on_disconnect() {
        let transport = InternetTransport::new("test-device");
        transport.on_status_changed(TransportStatus::Available);

        // Enqueue and dequeue two messages to put them in pending state.
        let msg1 = create_test_message();
        let msg2 = create_test_message();
        assert!(transport.send(&msg1).is_ok());
        assert!(transport.send(&msg2).is_ok());

        let _ = transport.get_next_message().unwrap().unwrap();
        let _ = transport.get_next_message().unwrap().unwrap();
        assert_eq!(transport.pending_confirmation_count(), 2);

        // Simulate disconnect — should fail all pending.
        transport.on_status_changed(TransportStatus::Disconnected);
        assert_eq!(transport.pending_confirmation_count(), 0);
        let metrics = transport.metrics();
        assert_eq!(metrics.failure_count, 2);
        assert_eq!(metrics.success_count, 0);
    }

    #[test]
    fn test_serialization() {
        let transport = InternetTransport::new("test-device");
        let message = create_test_message();

        // Serialize
        let data = transport.serialize_message(&message).unwrap();
        assert!(!data.is_empty());

        // Deserialize
        let deserialized = transport.deserialize_message(&data).unwrap();
        assert_eq!(deserialized.id, message.id);
    }

    #[test]
    fn test_stop_fails_pending() {
        let transport = InternetTransport::new("test-device");
        transport.on_status_changed(TransportStatus::Available);

        let msg = create_test_message();
        assert!(transport.send(&msg).is_ok());
        let _ = transport.get_next_message().unwrap().unwrap();
        assert_eq!(transport.pending_confirmation_count(), 1);

        transport.stop().unwrap();
        assert_eq!(transport.pending_confirmation_count(), 0);
        assert_eq!(transport.metrics().failure_count, 1);
    }

    /// Exhaustive destructuring guard for `update_metrics`.
    ///
    /// When a field is added to `TransportMetrics`, this test will fail to
    /// compile — forcing the developer to decide whether
    /// `InternetTransport::update_metrics` should copy the new field (platform-
    /// reported) or leave it alone (confirmation-loop-managed).
    #[test]
    fn update_metrics_covers_all_fields() {
        let m = TransportMetrics::default();
        // Confirmation-loop-managed fields (preserved by update_metrics):
        //   success_count, failure_count, delivery_ratio, drop_rate
        // Platform-reported fields (copied by update_metrics):
        //   rssi, latency_ms, bandwidth_bps, congestion, queue_depth,
        //   battery_level, is_charging, relay_connection_count,
        //   is_active_relay, average_hop_count, energy_cost
        let TransportMetrics {
            rssi: _,
            latency_ms: _,
            bandwidth_bps: _,
            congestion: _,
            queue_depth: _,
            success_count: _,
            failure_count: _,
            battery_level: _,
            is_charging: _,
            relay_connection_count: _,
            is_active_relay: _,
            delivery_ratio: _,
            drop_rate: _,
            average_hop_count: _,
            energy_cost: _,
        } = m;
    }

    #[test]
    fn test_send_when_unavailable_fails() {
        let transport = InternetTransport::new("test-device");
        let message = create_test_message();
        let result = transport.send(&message);
        assert!(result.is_err());
        assert!(matches!(
            result.unwrap_err(),
            crate::Error::TransportNotAvailable(_)
        ));
    }

    #[test]
    fn test_receive_when_empty_returns_none() {
        let transport = InternetTransport::new("test-device");
        let received = transport.receive().unwrap();
        assert!(received.is_none());
    }

    #[test]
    fn test_on_data_received_invalid_json_drops_ok() {
        let transport = InternetTransport::new("test-device");
        let result = transport.on_data_received(b"not valid json".to_vec());
        assert!(result.is_ok());
        assert!(transport.receive().unwrap().is_none());
    }

    #[test]
    fn test_on_data_received_valid_json_queues_message() {
        let transport = InternetTransport::new("test-device");
        let message = create_test_message();
        let data = transport.serialize_message(&message).unwrap();
        transport.on_data_received(data).unwrap();
        let received = transport.receive().unwrap();
        assert!(received.is_some());
        assert_eq!(received.unwrap().id, message.id);
    }

    #[test]
    fn test_update_metrics_preserves_confirmation_counts() {
        let transport = InternetTransport::new("test-device");
        transport.on_status_changed(TransportStatus::Available);
        let msg = create_test_message();
        transport.send(&msg).unwrap();
        let (id, _) = transport.get_next_message().unwrap().unwrap();
        transport.confirm_sent(&id);
        assert_eq!(transport.metrics().success_count, 1);

        let mut incoming = TransportMetrics::default();
        incoming.rssi = Some(-70);
        incoming.latency_ms = Some(100);
        transport.update_metrics(incoming);

        let m = transport.metrics();
        assert_eq!(m.rssi, Some(-70));
        assert_eq!(m.latency_ms, Some(100));
        assert_eq!(m.success_count, 1);
    }

    #[test]
    fn test_confirm_sent_unknown_id_no_panic() {
        let transport = InternetTransport::new("test-device");
        transport.confirm_sent("nonexistent-id");
        assert_eq!(transport.metrics().success_count, 0);
    }

    #[test]
    fn test_report_send_failure_unknown_id_no_panic() {
        let transport = InternetTransport::new("test-device");
        transport.report_send_failure("nonexistent-id");
        assert_eq!(transport.metrics().failure_count, 0);
    }

    #[test]
    fn test_start_does_not_require_connection() {
        let transport = InternetTransport::new("test-device");
        assert!(transport.start().is_ok());
        assert_eq!(transport.status(), TransportStatus::Unavailable);
    }

    #[test]
    fn test_on_data_received_rejects_oversized_payload() {
        let transport = InternetTransport::new("test-device");
        let oversized = vec![0u8; DEFAULT_MAX_MESSAGE_SIZE + 1];
        let result = transport.on_data_received(oversized);
        assert!(result.is_err());
        assert!(matches!(
            result.unwrap_err(),
            crate::Error::MessageTooLarge(_, _)
        ));
    }

    #[test]
    fn test_on_data_received_from_rejects_oversized_payload() {
        let transport = InternetTransport::new("test-device");
        let oversized = vec![0u8; DEFAULT_MAX_MESSAGE_SIZE + 1];
        let result = transport.on_data_received_from(oversized, "peer1".to_string());
        assert!(result.is_err());
        assert!(matches!(
            result.unwrap_err(),
            crate::Error::MessageTooLarge(_, _)
        ));
    }
}