Skip to main content

ant_quic/
node.rs

1// Copyright 2024 Saorsa Labs Ltd.
2//
3// This Saorsa Network Software is licensed under the General Public License (GPL), version 3.
4// Please see the file LICENSE-GPL, or visit <http://www.gnu.org/licenses/> for the full text.
5//
6// Full details available at https://saorsalabs.com/licenses
7
8//! Zero-configuration P2P node
9//!
10//! This module provides [`Node`] - the simple API for creating P2P nodes
11//! that work out of the box with zero configuration. Every node automatically:
12//!
13//! - Uses 100% post-quantum cryptography (ML-KEM-768)
14//! - Works behind any NAT via native QUIC hole punching
15//! - Offers relay/bootstrap/coordinator capability hints by default
16//! - Exposes a practical status snapshot via [`NodeStatus`]
17//!
18//! # Zero Configuration
19//!
20//! ```rust,ignore
21//! use ant_quic::Node;
22//!
23//! #[tokio::main]
24//! async fn main() -> anyhow::Result<()> {
25//!     // Create a node - that's it!
26//!     let node = Node::new().await?;
27//!
28//!     println!("I am: {:?}", node.peer_id());
29//!     println!("Listening on: {:?}", node.local_addr());
30//!
31//!     // Check status
32//!     let status = node.status().await;
33//!     println!("NAT behavior hint: {}", status.nat_type);
34//!     println!("Can receive direct: {}", status.can_receive_direct);
35//!     println!("Acting as relay: {}", status.is_relaying);
36//!
37//!     // Connect to a peer
38//!     let conn = node.connect_addr("quic.saorsalabs.com:9000".parse()?).await?;
39//!
40//!     // Accept connections
41//!     let incoming = node.accept().await;
42//!
43//!     Ok(())
44//! }
45//! ```
46
47use std::net::SocketAddr;
48use std::sync::Arc;
49use std::time::{Duration, Instant};
50
51use crate::bootstrap_cache::PeerCapabilities;
52use crate::crypto::pqc::types::{MlDsaPublicKey, MlDsaSecretKey};
53use tokio::sync::broadcast;
54use tracing::info;
55
56use crate::host_identity::HostIdentity;
57use crate::nat_traversal_api::PeerId;
58use crate::node_config::NodeConfig;
59use crate::node_event::NodeEvent;
60use crate::node_status::{NatType, NodeStatus};
61use crate::p2p_endpoint::{
62    AckDiagnosticsSnapshot, ConnectionHealth, ConnectionTransportStats,
63    DataChannelDiagnosticsSnapshot, EndpointError, P2pEndpoint, P2pEvent, PeerConnection,
64    PeerLifecycleEvent,
65};
66use crate::reachability::{DIRECT_REACHABILITY_TTL, socket_addr_scope};
67use crate::unified_config::P2pConfig;
68use crate::unified_config::load_or_generate_endpoint_keypair;
69
70/// Error type for Node operations
71#[derive(Debug, thiserror::Error)]
72pub enum NodeError {
73    /// Failed to create node
74    #[error("Failed to create node: {0}")]
75    Creation(String),
76
77    /// Connection error
78    #[error("Connection error: {0}")]
79    Connection(String),
80
81    /// Endpoint error
82    #[error("Endpoint error: {0}")]
83    Endpoint(#[from] EndpointError),
84
85    /// Shutting down
86    #[error("Node is shutting down")]
87    ShuttingDown,
88}
89
90/// Zero-configuration P2P node
91///
92/// This is the primary API for ant-quic. Create a node with zero configuration
93/// and it will automatically handle NAT traversal, post-quantum cryptography,
94/// and peer discovery.
95///
96/// # Symmetric P2P
97///
98/// All nodes are equal - every node can:
99/// - Connect to other nodes
100/// - Accept incoming connections
101/// - Act as coordinator for NAT traversal
102/// - Act as relay for peers behind restrictive NATs
103///
104/// # Post-Quantum Security
105///
106/// v0.2: Every connection uses pure post-quantum cryptography:
107/// - Key Exchange: ML-KEM-768 (FIPS 203)
108/// - Authentication: ML-DSA-65 (FIPS 204)
109/// - Ed25519 is used ONLY for the 32-byte PeerId compact identifier
110///
111/// There is no classical crypto fallback - security is quantum-resistant by default.
112///
113/// # Example
114///
115/// ```rust,ignore
116/// use ant_quic::Node;
117///
118/// // Zero configuration
119/// let node = Node::new().await?;
120///
121/// // Or with known peers
122/// let node = Node::with_peers(vec!["quic.saorsalabs.com:9000".parse()?]).await?;
123///
124/// // Or with persistent identity
125/// let keypair = load_keypair()?;
126/// let node = Node::with_keypair(keypair).await?;
127/// ```
128pub struct Node {
129    /// Inner P2pEndpoint
130    inner: Arc<P2pEndpoint>,
131
132    /// Start time for uptime calculation
133    start_time: Instant,
134
135    /// Event broadcaster for unified events
136    event_tx: broadcast::Sender<NodeEvent>,
137}
138
139fn node_config_to_p2p_config(config: NodeConfig) -> Result<P2pConfig, NodeError> {
140    let mut p2p_config = P2pConfig::default();
141
142    // Build transport registry first (before any partial moves)
143    p2p_config.transport_registry = config.build_transport_registry();
144
145    if let Some(bind_addr) = config.bind_addr {
146        p2p_config.bind_addr = Some(bind_addr.into());
147    }
148
149    p2p_config.known_peers = config.known_peers.into_iter().map(Into::into).collect();
150    p2p_config.keypair = config.keypair;
151
152    if let Some(capacity) = config.data_channel_capacity {
153        p2p_config.data_channel_capacity = capacity;
154    }
155    if let Some(streams) = config.max_concurrent_uni_streams {
156        p2p_config.max_concurrent_uni_streams = streams;
157    }
158    if let Some(max_message_size) = config.max_message_size {
159        if max_message_size == 0 {
160            return Err(NodeError::Creation(
161                "max_message_size must be at least 1".to_string(),
162            ));
163        }
164        p2p_config.max_message_size = max_message_size;
165    }
166    // Reviewer P2 #2: pipe NodeConfig::port_mapping_enabled into the
167    // underlying P2pConfig's NAT port-mapping toggle so app-level
168    // opt-out (e.g. x0x daemon config / CLI flag) actually disables
169    // the UPnP discovery task.
170    if let Some(enabled) = config.port_mapping_enabled {
171        p2p_config.nat.port_mapping.enabled = enabled;
172    }
173
174    // Issue #206: pipe NodeConfig's mDNS plane-isolation knobs into the
175    // underlying P2pConfig discovery policy so embedders using the simple
176    // Node API can keep co-located daemons on different logical planes
177    // (e.g. prod + testnet) from discovering and auto-connecting to each
178    // other via the default `ant-quic` mDNS service.
179    if let Some(enabled) = config.mdns_enabled {
180        let mut mdns = p2p_config.discovery.mdns.unwrap_or_default();
181        mdns.enabled = enabled;
182        p2p_config.discovery.mdns = Some(mdns);
183    }
184    if let Some(namespace) = config.mdns_namespace {
185        let mut mdns = p2p_config.discovery.mdns.unwrap_or_default();
186        mdns.namespace = Some(namespace);
187        p2p_config.discovery.mdns = Some(mdns);
188    }
189
190    if let Some(cache_config) = config.bootstrap_cache {
191        p2p_config.bootstrap_cache = cache_config;
192    }
193
194    Ok(p2p_config)
195}
196
197impl std::fmt::Debug for Node {
198    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
199        f.debug_struct("Node")
200            .field("peer_id", &self.peer_id())
201            .field("local_addr", &self.local_addr())
202            .finish_non_exhaustive()
203    }
204}
205
206impl Node {
207    // === Creation ===
208
209    /// Create a node with automatic configuration
210    ///
211    /// This is the recommended way to create a node. It will:
212    /// - Bind to a random port on all interfaces (0.0.0.0:0)
213    /// - Generate a fresh Ed25519 keypair
214    /// - Enable all NAT traversal capabilities
215    /// - Use 100% post-quantum cryptography
216    ///
217    /// # Example
218    ///
219    /// ```rust,ignore
220    /// let node = Node::new().await?;
221    /// ```
222    pub async fn new() -> Result<Self, NodeError> {
223        Self::with_config(NodeConfig::default()).await
224    }
225
226    /// Create a node with a specific bind address
227    ///
228    /// Use this when you need a specific port for firewall rules or port forwarding.
229    ///
230    /// # Example
231    ///
232    /// ```rust,ignore
233    /// let node = Node::bind("0.0.0.0:9000".parse()?).await?;
234    /// ```
235    pub async fn bind(addr: SocketAddr) -> Result<Self, NodeError> {
236        Self::with_config(NodeConfig::with_bind_addr(addr)).await
237    }
238
239    /// Create a node with known peers
240    ///
241    /// Use this when you have a list of known peers to connect to initially.
242    /// These can be any nodes in the network - they'll help with NAT traversal.
243    ///
244    /// # Example
245    ///
246    /// ```rust,ignore
247    /// let node = Node::with_peers(vec![
248    ///     "quic.saorsalabs.com:9000".parse()?,
249    ///     "peer2.example.com:9000".parse()?,
250    /// ]).await?;
251    /// ```
252    pub async fn with_peers(peers: Vec<SocketAddr>) -> Result<Self, NodeError> {
253        Self::with_config(NodeConfig::with_known_peers(peers)).await
254    }
255
256    /// Create a node with an existing keypair
257    ///
258    /// Use this for persistent identity across restarts. The peer ID
259    /// is derived from the public key, so using the same keypair
260    /// gives you the same peer ID.
261    ///
262    /// # Example
263    ///
264    /// ```rust,ignore
265    /// let (public_key, secret_key) = load_keypair_from_file("~/.ant-quic/identity.key")?;
266    /// let node = Node::with_keypair(public_key, secret_key).await?;
267    /// ```
268    pub async fn with_keypair(
269        public_key: MlDsaPublicKey,
270        secret_key: MlDsaSecretKey,
271    ) -> Result<Self, NodeError> {
272        Self::with_config(NodeConfig::with_keypair(public_key, secret_key)).await
273    }
274
275    /// Create a node with a HostIdentity for persistent encrypted identity
276    ///
277    /// This is the recommended way to create a node with persistent identity.
278    /// The keypair is encrypted at rest using a key derived from the HostIdentity.
279    ///
280    /// # Arguments
281    ///
282    /// * `host` - The HostIdentity for key derivation
283    /// * `network_id` - Network identifier for per-network keypair isolation
284    /// * `storage_dir` - Directory to store the encrypted keypair
285    ///
286    /// # Example
287    ///
288    /// ```rust,ignore
289    /// use ant_quic::{Node, HostIdentity};
290    ///
291    /// let host = HostIdentity::generate();
292    /// let node = Node::with_host_identity(
293    ///     &host,
294    ///     b"my-network",
295    ///     "/var/lib/ant-quic",
296    /// ).await?;
297    /// ```
298    pub async fn with_host_identity(
299        host: &HostIdentity,
300        network_id: &[u8],
301        storage_dir: impl AsRef<std::path::Path>,
302    ) -> Result<Self, NodeError> {
303        let (public_key, secret_key) =
304            load_or_generate_endpoint_keypair(host, network_id, storage_dir.as_ref()).map_err(
305                |e| NodeError::Creation(format!("Failed to load/generate keypair: {e}")),
306            )?;
307
308        Self::with_keypair(public_key, secret_key).await
309    }
310
311    /// Create a node with full configuration
312    ///
313    /// For power users who need specific settings. Most applications
314    /// should use `Node::new()` or one of the convenience methods.
315    ///
316    /// # Example
317    ///
318    /// ```rust,ignore
319    /// let config = NodeConfig::builder()
320    ///     .bind_addr("0.0.0.0:9000".parse()?)
321    ///     .known_peer("quic.saorsalabs.com:9000".parse()?)
322    ///     .keypair(load_keypair()?)
323    ///     .build();
324    ///
325    /// let node = Node::with_config(config).await?;
326    /// ```
327    pub async fn with_config(config: NodeConfig) -> Result<Self, NodeError> {
328        let p2p_config = node_config_to_p2p_config(config)?;
329
330        // Create event channel
331        let (event_tx, _) = broadcast::channel(256);
332
333        // Create P2pEndpoint
334        let endpoint = P2pEndpoint::new(p2p_config)
335            .await
336            .map_err(NodeError::Endpoint)?;
337
338        info!("Node created with peer ID: {:?}", endpoint.peer_id());
339
340        let inner = Arc::new(endpoint);
341
342        // Spawn event bridge task to forward P2pEvent -> NodeEvent
343        Self::spawn_event_bridge(Arc::clone(&inner), event_tx.clone());
344
345        Ok(Self {
346            inner,
347            start_time: Instant::now(),
348            event_tx,
349        })
350    }
351
352    /// Spawn a background task to bridge P2pEvents to NodeEvents
353    fn spawn_event_bridge(endpoint: Arc<P2pEndpoint>, event_tx: broadcast::Sender<NodeEvent>) {
354        let mut p2p_events = endpoint.subscribe();
355
356        tokio::spawn(async move {
357            loop {
358                match p2p_events.recv().await {
359                    Ok(p2p_event) => {
360                        if let Some(node_event) = Self::convert_event(p2p_event) {
361                            // Ignore send errors - means no subscribers
362                            let _ = event_tx.send(node_event);
363                        }
364                    }
365                    Err(broadcast::error::RecvError::Closed) => {
366                        // Channel closed, endpoint shutting down
367                        break;
368                    }
369                    Err(broadcast::error::RecvError::Lagged(n)) => {
370                        // Subscriber lagged behind, log and continue
371                        tracing::warn!("Event bridge lagged by {} events", n);
372                    }
373                }
374            }
375        });
376    }
377
378    /// Convert a P2pEvent to a NodeEvent
379    ///
380    /// Uses the From trait implementation for DisconnectReason conversion.
381    fn convert_event(p2p_event: P2pEvent) -> Option<NodeEvent> {
382        match p2p_event {
383            P2pEvent::PeerConnected {
384                peer_id,
385                addr,
386                side: _,
387                traversal_method,
388            } => Some(NodeEvent::PeerConnected {
389                peer_id,
390                addr,
391                method: traversal_method,
392                direct: traversal_method.is_direct(),
393            }),
394            P2pEvent::PeerDisconnected { peer_id, reason } => Some(NodeEvent::PeerDisconnected {
395                peer_id,
396                reason: reason.into(), // Use From trait
397            }),
398            P2pEvent::ExternalAddressDiscovered { addr } => {
399                Some(NodeEvent::ExternalAddressDiscovered { addr })
400            }
401            P2pEvent::PortMappingEstablished { external_addr } => {
402                Some(NodeEvent::PortMappingEstablished { external_addr })
403            }
404            P2pEvent::PortMappingRenewed { external_addr } => {
405                Some(NodeEvent::PortMappingRenewed { external_addr })
406            }
407            P2pEvent::PortMappingAddressChanged {
408                previous_addr,
409                external_addr,
410            } => Some(NodeEvent::PortMappingAddressChanged {
411                previous_addr,
412                external_addr,
413            }),
414            P2pEvent::PortMappingFailed { error } => Some(NodeEvent::PortMappingFailed { error }),
415            P2pEvent::PortMappingRemoved { external_addr } => {
416                Some(NodeEvent::PortMappingRemoved { external_addr })
417            }
418            P2pEvent::DirectPathStatus { peer_id, status } => {
419                Some(NodeEvent::DirectPathStatus { peer_id, status })
420            }
421            P2pEvent::DataReceived { peer_id, bytes } => Some(NodeEvent::DataReceived {
422                peer_id,
423                stream_id: 0, // P2pEvent doesn't track stream IDs
424                bytes,
425            }),
426            P2pEvent::ConstrainedDataReceived {
427                remote_addr,
428                connection_id,
429                data,
430            } => {
431                // For constrained data, derive a synthetic peer ID from the transport address
432                let synthetic_peer_id = {
433                    use std::collections::hash_map::DefaultHasher;
434                    use std::hash::{Hash, Hasher};
435                    let synthetic_addr = remote_addr.to_synthetic_socket_addr();
436                    let mut hasher = DefaultHasher::new();
437                    synthetic_addr.hash(&mut hasher);
438                    let hash = hasher.finish();
439                    let mut peer_id_bytes = [0u8; 32];
440                    peer_id_bytes[..8].copy_from_slice(&hash.to_le_bytes());
441                    PeerId(peer_id_bytes)
442                };
443                Some(NodeEvent::DataReceived {
444                    peer_id: synthetic_peer_id,
445                    stream_id: connection_id as u64,
446                    bytes: data.len(),
447                })
448            }
449            P2pEvent::MdnsServiceAdvertised {
450                service,
451                namespace,
452                instance_fullname,
453            } => Some(NodeEvent::MdnsServiceAdvertised {
454                service,
455                namespace,
456                instance_fullname,
457            }),
458            P2pEvent::MdnsPeerDiscovered { peer } => Some(NodeEvent::MdnsPeerDiscovered { peer }),
459            P2pEvent::MdnsPeerUpdated { peer } => Some(NodeEvent::MdnsPeerUpdated { peer }),
460            P2pEvent::MdnsPeerRemoved { peer } => Some(NodeEvent::MdnsPeerRemoved { peer }),
461            P2pEvent::MdnsPeerEligible { peer } => Some(NodeEvent::MdnsPeerEligible { peer }),
462            P2pEvent::MdnsPeerIneligible { peer, reason } => {
463                Some(NodeEvent::MdnsPeerIneligible { peer, reason })
464            }
465            P2pEvent::MdnsPeerApprovalRequired { peer, reason } => {
466                Some(NodeEvent::MdnsPeerApprovalRequired { peer, reason })
467            }
468            P2pEvent::MdnsAutoConnectAttempted { peer, addresses } => {
469                Some(NodeEvent::MdnsAutoConnectAttempted { peer, addresses })
470            }
471            P2pEvent::MdnsAutoConnectSucceeded {
472                peer,
473                authenticated_peer_id,
474                remote_addr,
475            } => Some(NodeEvent::MdnsAutoConnectSucceeded {
476                peer,
477                authenticated_peer_id,
478                remote_addr,
479            }),
480            P2pEvent::MdnsAutoConnectFailed {
481                peer,
482                addresses,
483                error,
484            } => Some(NodeEvent::MdnsAutoConnectFailed {
485                peer,
486                addresses,
487                error,
488            }),
489            // Events without direct NodeEvent equivalents are ignored
490            P2pEvent::NatTraversalProgress { .. }
491            | P2pEvent::BootstrapStatus { .. }
492            | P2pEvent::PeerAuthenticated { .. }
493            | P2pEvent::PeerAddressUpdated { .. }
494            | P2pEvent::RelayEstablished { .. } => None,
495        }
496    }
497
498    // === Identity ===
499
500    /// Get this node's peer ID
501    ///
502    /// The peer ID is derived from the Ed25519 public key and is
503    /// the unique identifier for this node on the network.
504    pub fn peer_id(&self) -> PeerId {
505        self.inner.peer_id()
506    }
507
508    /// Get the local bind address
509    ///
510    /// Returns `None` if the endpoint hasn't bound yet.
511    pub fn local_addr(&self) -> Option<SocketAddr> {
512        self.inner.local_addr()
513    }
514
515    /// Get the observed external address
516    ///
517    /// This is the address as seen by other peers on the network.
518    /// Returns `None` if no external address has been discovered yet.
519    pub fn external_addr(&self) -> Option<SocketAddr> {
520        self.inner.external_addr()
521    }
522
523    /// Return the latest best-effort direct-path status for a peer, when known.
524    pub fn direct_path_status(&self, peer_id: PeerId) -> Option<crate::DirectPathStatus> {
525        self.inner.direct_path_status(peer_id)
526    }
527
528    /// Get the ML-DSA-65 public key bytes (1952 bytes)
529    pub fn public_key_bytes(&self) -> &[u8] {
530        self.inner.public_key_bytes()
531    }
532
533    /// Get access to the underlying P2pEndpoint for advanced operations.
534    pub fn inner_endpoint(&self) -> &Arc<P2pEndpoint> {
535        &self.inner
536    }
537
538    /// The node's bootstrap peer cache.
539    ///
540    /// This is the single cache instance the endpoint uses for
541    /// quality-scored reconnection, coordinator selection and bootstrap
542    /// tokens (configured via [`NodeConfig::bootstrap_cache`]). Embedders
543    /// should enrich and
544    /// query this shared instance rather than opening a second cache on
545    /// the same directory. The endpoint runs cache maintenance itself —
546    /// do not call `start_maintenance` on this handle.
547    pub fn bootstrap_cache(&self) -> Arc<crate::BootstrapCache> {
548        Arc::clone(&self.inner.bootstrap_cache)
549    }
550
551    /// Get the transport registry for this node
552    ///
553    /// The transport registry contains all registered transport providers (UDP, BLE, etc.)
554    /// that this node can use for connectivity.
555    pub fn transport_registry(&self) -> &crate::transport::TransportRegistry {
556        self.inner.transport_registry()
557    }
558
559    // === Connections ===
560
561    /// Connect to a peer by address.
562    ///
563    /// Thin facade over [`P2pEndpoint::connect_addr`], which uses the unified
564    /// outbound connectivity orchestrator.
565    pub async fn connect_addr(&self, addr: SocketAddr) -> Result<PeerConnection, NodeError> {
566        self.inner
567            .connect_addr(addr)
568            .await
569            .map_err(NodeError::Endpoint)
570    }
571
572    /// Connect to a peer by durable peer ID.
573    ///
574    /// Thin facade over the unified peer-oriented [`P2pEndpoint`] connect path.
575    /// Strategy selection remains internal to the endpoint.
576    pub async fn connect_peer(&self, peer_id: PeerId) -> Result<PeerConnection, NodeError> {
577        self.inner
578            .connect_peer(peer_id)
579            .await
580            .map_err(NodeError::Endpoint)
581    }
582
583    /// Connect to a peer by durable peer ID.
584    ///
585    /// Compatibility-oriented alias retained for older callers. Prefer
586    /// [`Self::connect_peer`] as the canonical peer-oriented public surface.
587    #[deprecated(note = "use connect_peer(peer_id) for the canonical peer-oriented API")]
588    pub async fn connect(&self, peer_id: PeerId) -> Result<PeerConnection, NodeError> {
589        self.connect_peer(peer_id).await
590    }
591
592    /// Connect to a peer by durable peer ID plus explicit address hints.
593    ///
594    /// Use this when the caller has candidate addresses for the peer and wants
595    /// the transport to combine those hints with peer-authenticated fallback
596    /// orchestration.
597    pub async fn connect_peer_with_addrs(
598        &self,
599        peer_id: PeerId,
600        addrs: Vec<SocketAddr>,
601    ) -> Result<PeerConnection, NodeError> {
602        self.inner
603            .connect_peer_with_addrs(peer_id, addrs)
604            .await
605            .map_err(NodeError::Endpoint)
606    }
607
608    /// Merge externally discovered peer hints into the node's transport view.
609    ///
610    /// This is the advanced discovery bridge for callers that learn peer
611    /// addresses or assist-role capability hints from higher layers.
612    pub async fn upsert_peer_hints(
613        &self,
614        peer_id: PeerId,
615        addrs: Vec<SocketAddr>,
616        capabilities: Option<PeerCapabilities>,
617    ) {
618        self.inner
619            .upsert_peer_hints(peer_id, addrs, capabilities)
620            .await;
621    }
622
623    /// Accept an incoming connection
624    ///
625    /// Waits for and accepts the next incoming connection.
626    /// Returns `None` if the node is shutting down.
627    ///
628    /// # Example
629    ///
630    /// ```rust,ignore
631    /// while let Some(conn) = node.accept().await {
632    ///     println!("Accepted connection from: {:?}", conn.peer_id);
633    ///     // Handle connection...
634    /// }
635    /// ```
636    pub async fn accept(&self) -> Option<PeerConnection> {
637        self.inner.accept().await
638    }
639
640    /// Add a known peer dynamically.
641    ///
642    /// Thin facade over [`P2pEndpoint::add_known_peer`]. Known peers help with
643    /// initial connectivity, discovery, and NAT traversal coordination.
644    pub async fn add_peer(&self, addr: SocketAddr) {
645        self.inner.add_known_peer(addr).await;
646    }
647
648    /// Connect to all known peers
649    ///
650    /// Returns the number of successful connections.
651    pub async fn connect_known_peers(&self) -> Result<usize, NodeError> {
652        self.inner
653            .connect_known_peers()
654            .await
655            .map_err(NodeError::Endpoint)
656    }
657
658    /// Disconnect from a peer
659    pub async fn disconnect(&self, peer_id: &PeerId) -> Result<(), NodeError> {
660        self.inner
661            .disconnect(peer_id)
662            .await
663            .map_err(NodeError::Endpoint)
664    }
665
666    /// Get list of connected peers
667    pub async fn connected_peers(&self) -> Vec<PeerConnection> {
668        self.inner.connected_peers().await
669    }
670
671    /// Check if connected to a peer
672    pub async fn is_connected(&self, peer_id: &PeerId) -> bool {
673        self.inner.is_connected(peer_id).await
674    }
675
676    /// Get a best-effort connection health snapshot for a peer.
677    pub async fn connection_health(&self, peer_id: &PeerId) -> ConnectionHealth {
678        self.inner.connection_health(peer_id).await
679    }
680
681    /// Get qlog-style transport path telemetry for a connected peer.
682    pub async fn connection_transport_stats(
683        &self,
684        peer_id: &PeerId,
685    ) -> Option<ConnectionTransportStats> {
686        self.inner.connection_transport_stats(peer_id).await
687    }
688
689    /// Subscribe to lifecycle events for a specific peer.
690    pub fn subscribe_peer_events(
691        &self,
692        peer_id: &PeerId,
693    ) -> broadcast::Receiver<PeerLifecycleEvent> {
694        self.inner.subscribe_peer_events(peer_id)
695    }
696
697    /// Subscribe to lifecycle events for all peers.
698    pub fn subscribe_all_peer_events(&self) -> broadcast::Receiver<(PeerId, PeerLifecycleEvent)> {
699        self.inner.subscribe_all_peer_events()
700    }
701
702    // === Messaging ===
703
704    /// Send data to a peer
705    pub async fn send(&self, peer_id: &PeerId, data: &[u8]) -> Result<(), NodeError> {
706        self.inner
707            .send(peer_id, data)
708            .await
709            .map_err(NodeError::Endpoint)
710    }
711
712    /// Send data and wait until the remote receive pipeline accepts it.
713    pub async fn send_with_receive_ack(
714        &self,
715        peer_id: &PeerId,
716        data: &[u8],
717        timeout: Duration,
718    ) -> Result<(), NodeError> {
719        self.inner
720            .send_with_receive_ack(peer_id, data, timeout)
721            .await
722            .map_err(NodeError::Endpoint)
723    }
724
725    /// Same as [`send_with_receive_ack`] but the caller supplies the ACK-v2
726    /// request id. Repeated calls with the same `(peer_id, request_id, data)`
727    /// are duplicate-safe at the receiver — the second arrival is replayed
728    /// from the receiver-side ACK dedupe cache and the payload is not
729    /// redelivered to `recv()`. Intended for application-level request
730    /// hedging (x0x X0X-0066).
731    pub async fn send_with_receive_ack_with_request_id(
732        &self,
733        peer_id: &PeerId,
734        request_id: [u8; 16],
735        data: &[u8],
736        timeout: Duration,
737    ) -> Result<(), NodeError> {
738        self.inner
739            .send_with_receive_ack_with_request_id(peer_id, request_id, data, timeout)
740            .await
741            .map_err(NodeError::Endpoint)
742    }
743
744    /// Actively probe peer liveness and measure round-trip time.
745    ///
746    /// Sends a lightweight probe envelope and waits for the peer's reader task
747    /// to acknowledge it. Returns the measured round-trip duration on success.
748    /// Probe traffic is invisible to [`Self::recv`] — it does not emit
749    /// `DataReceived` events or deliver payloads.
750    pub async fn probe_peer(
751        &self,
752        peer_id: &PeerId,
753        timeout: Duration,
754    ) -> Result<Duration, NodeError> {
755        self.inner
756            .probe_peer(peer_id, timeout)
757            .await
758            .map_err(NodeError::Endpoint)
759    }
760
761    /// Snapshot stage-by-stage ACK-v2 latency and outcome diagnostics.
762    pub fn ack_diagnostics(&self) -> AckDiagnosticsSnapshot {
763        self.inner.ack_diagnostics()
764    }
765
766    /// Snapshot `data_tx` channel saturation diagnostics (X0X-0039).
767    ///
768    /// Surfaces depth, capacity, and cumulative high-water-count for the
769    /// shared `mpsc::Sender` fed by every per-connection reader task.
770    /// Consumed by `x0x` `/diagnostics/connectivity` to detect mesh-burst
771    /// back-pressure.
772    pub fn data_channel_diagnostics(&self) -> DataChannelDiagnosticsSnapshot {
773        self.inner.data_channel_diagnostics()
774    }
775
776    /// Snapshot GSO bundle send diagnostics (X0X-0043).
777    ///
778    /// Returns cumulative counts of multi-segment GSO bundles submitted to
779    /// the kernel send path and of bundles reported as partial / failed.
780    /// Consumed by `x0x` `/diagnostics/connectivity` to test the Quinn
781    /// issue #2627 GSO-tail-drop hypothesis as an alternative root cause
782    /// for X0X-0030 idle-rot send timeouts. See
783    /// [`crate::diagnostics::gso`] for the full discussion.
784    pub fn gso_diagnostics(&self) -> crate::GsoDiagnosticsSnapshot {
785        self.inner.gso_diagnostics()
786    }
787
788    /// Receive data from any peer
789    pub async fn recv(&self) -> Result<(PeerId, Vec<u8>), NodeError> {
790        self.inner.recv().await.map_err(NodeError::Endpoint)
791    }
792
793    // === Application byte-streams ============================================
794    //
795    // Bidirectional QUIC byte-streams to/from a connected peer. These are the
796    // stream primitive x0x's tailnet forwarding consumes. See
797    // `docs/design/node-app-bidi-streams.md` for the separation guarantee.
798
799    /// Open a bidirectional **application** byte-stream to a connected peer.
800    ///
801    /// Returns a `(send, recv)` pair. `send` (`[`HighLevelSendStream`]`)
802    /// implements [`tokio::io::AsyncWrite`] and `recv` (`[`HighLevelRecvStream`]`)
803    /// implements [`tokio::io::AsyncRead`], so callers can bridge them directly
804    /// to a local TCP/SOCKS socket with `tokio::io::copy` / `copy_bidirectional`.
805    ///
806    /// The stream shares the peer's existing authenticated QUIC connection
807    /// (direct or relayed) and inherits its ML-DSA-65 peer identity. It is
808    /// demultiplexed from ant-quic's internal ACK-v2 / relay / message traffic
809    /// by a reserved stream prefix, so it never interferes with [`Self::send`]
810    /// / [`Self::recv`] and the peer's [`Self::accept_bi`] never sees internal
811    /// streams. Byte-level backpressure is QUIC-native.
812    ///
813    /// # Example
814    ///
815    /// ```rust,ignore
816    /// let (mut send, mut recv) = node.open_bi(&peer_id).await?;
817    /// use tokio::io::AsyncWriteExt;
818    /// send.write_all(b"hello").await?;
819    /// // Always finish explicitly. A send stream dropped without `finish()` is
820    /// // reset, so the peer observes a stream error instead of a short read that
821    /// // looks like a complete message.
822    /// send.finish()?;
823    /// ```
824    ///
825    /// # Errors
826    ///
827    /// - [`NodeError::ShuttingDown`] if the node is shutting down.
828    /// - [`NodeError::Endpoint`] with [`EndpointError::PeerNotFound`] if no live
829    ///   QUIC connection exists for this peer.
830    pub async fn open_bi(
831        &self,
832        peer_id: &PeerId,
833    ) -> Result<(crate::HighLevelSendStream, crate::HighLevelRecvStream), NodeError> {
834        self.inner
835            .open_bi(peer_id)
836            .await
837            .map_err(NodeError::Endpoint)
838    }
839
840    /// Accept the next inbound **application** bidirectional byte-stream from
841    /// any peer.
842    ///
843    /// Yields `(peer_id, send, recv)`. Only application-opened streams are
844    /// surfaced here — ant-quic's internal transport streams (ACK-v2, MASQUE
845    /// relay, message datagrams) are demultiplexed earlier in the reader task
846    /// and can **never** be returned by this method. This is the core
847    /// separation invariant; see the regression test
848    /// `accept_bi_never_yields_internal_stream`.
849    ///
850    /// Like [`Self::open_bi`], the streams inherit the connection's ML-DSA-65
851    /// peer auth and use QUIC-native backpressure.
852    ///
853    /// # Errors
854    ///
855    /// Returns [`NodeError::ShuttingDown`] once shutdown has begun and the
856    /// internal queue is drained.
857    pub async fn accept_bi(
858        &self,
859    ) -> Result<
860        (
861            PeerId,
862            crate::HighLevelSendStream,
863            crate::HighLevelRecvStream,
864        ),
865        NodeError,
866    > {
867        self.inner.accept_bi().await.map_err(NodeError::Endpoint)
868    }
869
870    // === Observability ===
871
872    /// Get a snapshot of the node's current status
873    ///
874    /// This provides a practical snapshot of the node's state,
875    /// including a best-effort NAT behavior hint, connectivity,
876    /// relay/coordinator hints, and performance.
877    ///
878    /// # Example
879    ///
880    /// ```rust,ignore
881    /// let status = node.status().await;
882    /// println!("NAT behavior hint: {}", status.nat_type);
883    /// println!("Connected peers: {}", status.connected_peers);
884    /// println!("Acting as relay: {}", status.is_relaying);
885    /// ```
886    pub async fn status(&self) -> NodeStatus {
887        let stats = self.inner.stats().await;
888        let connected_peers = self.inner.connected_peers().await;
889
890        // Derive a best-effort NAT behavior hint from native connectivity
891        // outcomes only. This is observational telemetry, not authoritative
892        // NAT classification.
893        let nat_type = self.detect_nat_type(&stats);
894
895        // Address knowledge and reachability are separate concepts.
896        // A global address is not proof of direct reachability.
897        let local_addr = self.local_addr();
898        let external_addr = self.external_addr();
899
900        // Collect ALL external addresses (both IPv4 and IPv6) from all
901        // connections and QUIC paths. This is critical for dual-stack nodes
902        // where different peers report different address families.
903        let mut external_addrs = self.inner.all_external_addrs();
904        // Ensure the primary external address is included (backward compat)
905        if let Some(addr) = external_addr {
906            if !external_addrs.contains(&addr) {
907                external_addrs.insert(0, addr);
908            }
909        }
910
911        // Calculate hole punch success rate
912        let hole_punch_success_rate = if stats.nat_traversal_attempts > 0 {
913            stats.nat_traversal_successes as f64 / stats.nat_traversal_attempts as f64
914        } else {
915            0.0
916        };
917
918        let has_global_address = external_addrs
919            .iter()
920            .copied()
921            .chain(local_addr)
922            .any(|addr| {
923                socket_addr_scope(addr)
924                    .is_some_and(|scope| scope == crate::ReachabilityScope::Global)
925            });
926        let port_mapping = self.inner.port_mapping_snapshot();
927        let mdns = self.inner.mdns_snapshot();
928
929        // A node is directly reachable only after fresh, peer-verified direct
930        // inbound evidence. Scope is freshness-aware too, so an old global
931        // observation cannot keep inflating current reachability.
932        let fresh_scope = [
933            (
934                crate::ReachabilityScope::Global,
935                stats.last_direct_global_at,
936            ),
937            (
938                crate::ReachabilityScope::LocalNetwork,
939                stats.last_direct_local_at,
940            ),
941            (
942                crate::ReachabilityScope::Loopback,
943                stats.last_direct_loopback_at,
944            ),
945        ]
946        .into_iter()
947        .find_map(|(scope, seen)| {
948            seen.filter(|instant| instant.elapsed() <= DIRECT_REACHABILITY_TTL)
949                .map(|_| scope)
950        });
951        let can_receive_direct =
952            stats.active_direct_incoming_connections > 0 || fresh_scope.is_some();
953        let direct_reachability_scope = fresh_scope;
954
955        // Relay/coordinator activity is still best-effort, but we can surface
956        // a conservative runtime snapshot from existing NAT/relay state instead
957        // of hard-coded false/zero placeholders.
958        let runtime_assist = self.inner.runtime_assist_snapshot().await;
959        let relay_service_enabled = self.inner.relay_service_enabled();
960        let coordinator_service_enabled = self.inner.coordinator_service_enabled();
961        let bootstrap_service_enabled = self.inner.bootstrap_service_enabled();
962        let is_relaying = runtime_assist.active_relay_sessions > 0;
963        let relay_sessions = runtime_assist.active_relay_sessions;
964        let relay_bytes_forwarded = runtime_assist.relay_bytes_forwarded;
965        let is_coordinating = runtime_assist.successful_coordinations > 0;
966        let coordination_sessions =
967            usize::try_from(runtime_assist.successful_coordinations).unwrap_or(usize::MAX);
968
969        // Calculate average RTT from connected peers
970        let mut total_rtt = Duration::ZERO;
971        let mut rtt_count = 0u32;
972        for peer in &connected_peers {
973            if let Some(metrics) = self.inner.connection_metrics(&peer.peer_id).await {
974                if let Some(rtt) = metrics.rtt {
975                    total_rtt += rtt;
976                    rtt_count += 1;
977                }
978            }
979        }
980        let avg_rtt = if rtt_count > 0 {
981            total_rtt / rtt_count
982        } else {
983            Duration::ZERO
984        };
985
986        NodeStatus {
987            peer_id: self.peer_id(),
988            local_addr: local_addr.unwrap_or_else(|| {
989                "0.0.0.0:0".parse().unwrap_or_else(|_| {
990                    SocketAddr::new(std::net::IpAddr::V4(std::net::Ipv4Addr::UNSPECIFIED), 0)
991                })
992            }),
993            external_addrs,
994            nat_type,
995            can_receive_direct,
996            direct_reachability_scope,
997            has_global_address,
998            port_mapping_active: port_mapping.active,
999            port_mapping_addr: port_mapping.external_addr,
1000            mdns_browsing: mdns.browsing,
1001            mdns_advertising: mdns.advertising,
1002            mdns_discovered_peers: mdns.discovered_peers.len(),
1003            relay_service_enabled,
1004            coordinator_service_enabled,
1005            bootstrap_service_enabled,
1006            connected_peers: connected_peers.len(),
1007            active_connections: stats.active_connections,
1008            pending_connections: 0, // Not tracked yet
1009            direct_connections: stats.direct_connections,
1010            relayed_connections: stats.relayed_connections,
1011            hole_punch_success_rate,
1012            is_relaying,
1013            relay_sessions,
1014            relay_bytes_forwarded,
1015            is_coordinating,
1016            coordination_sessions,
1017            avg_rtt,
1018            uptime: self.start_time.elapsed(),
1019        }
1020    }
1021
1022    /// Subscribe to node events
1023    ///
1024    /// Returns a receiver for all significant node events including
1025    /// connections, disconnections, NAT detection, and relay activity.
1026    ///
1027    /// # Example
1028    ///
1029    /// ```rust,ignore
1030    /// let mut events = node.subscribe();
1031    /// tokio::spawn(async move {
1032    ///     while let Ok(event) = events.recv().await {
1033    ///         match event {
1034    ///             NodeEvent::PeerConnected { peer_id, .. } => {
1035    ///                 println!("Connected: {:?}", peer_id);
1036    ///             }
1037    ///             _ => {}
1038    ///         }
1039    ///     }
1040    /// });
1041    /// ```
1042    pub fn subscribe(&self) -> broadcast::Receiver<NodeEvent> {
1043        self.event_tx.subscribe()
1044    }
1045
1046    /// Subscribe to raw P2pEvents (for advanced use)
1047    ///
1048    /// This provides access to the underlying P2pEndpoint events.
1049    /// Most applications should use `subscribe()` for NodeEvents.
1050    pub fn subscribe_raw(&self) -> broadcast::Receiver<P2pEvent> {
1051        self.inner.subscribe()
1052    }
1053
1054    // === Shutdown ===
1055
1056    /// Gracefully shut down the node
1057    ///
1058    /// This closes all connections and releases resources.
1059    pub async fn shutdown(self) {
1060        self.inner.shutdown().await;
1061    }
1062
1063    /// Check if the node is still running
1064    pub fn is_running(&self) -> bool {
1065        self.inner.is_running()
1066    }
1067
1068    // === Private Helpers ===
1069
1070    /// Derive a coarse NAT behavior hint from native QUIC connection outcomes.
1071    ///
1072    /// This does not classify NAT mapping/filtering behavior in the RFC 4787 /
1073    /// RFC 5780 sense.
1074    fn detect_nat_type(&self, stats: &crate::p2p_endpoint::EndpointStats) -> NatType {
1075        // This remains a soft debug hint only. Do not treat it as direct
1076        // reachability evidence.
1077        if stats.direct_connections > 0 && stats.relayed_connections == 0 {
1078            return NatType::FullCone;
1079        }
1080
1081        if stats.direct_connections > 0 && stats.relayed_connections > 0 {
1082            return NatType::PortRestricted;
1083        }
1084
1085        if stats.relayed_connections > stats.direct_connections {
1086            return NatType::Symmetric;
1087        }
1088
1089        NatType::Unknown
1090    }
1091}
1092
1093// Enable cloning through Arc
1094impl Clone for Node {
1095    fn clone(&self) -> Self {
1096        Self {
1097            inner: Arc::clone(&self.inner),
1098            start_time: self.start_time,
1099            event_tx: self.event_tx.clone(),
1100        }
1101    }
1102}
1103
1104#[cfg(test)]
1105mod tests {
1106    use super::*;
1107    use crate::derive_peer_id_from_public_key;
1108
1109    #[tokio::test]
1110    async fn test_node_new_default() {
1111        let node = Node::new().await;
1112        assert!(node.is_ok(), "Node::new() should succeed: {:?}", node.err());
1113
1114        let node = node.unwrap();
1115        assert!(node.is_running());
1116
1117        // Peer ID should be valid (non-zero)
1118        let peer_id = node.peer_id();
1119        assert_ne!(peer_id.0, [0u8; 32]);
1120
1121        node.shutdown().await;
1122    }
1123
1124    #[tokio::test]
1125    async fn test_node_bind() {
1126        let addr: SocketAddr = "127.0.0.1:0".parse().unwrap();
1127        let node = Node::bind(addr).await;
1128        assert!(node.is_ok(), "Node::bind() should succeed");
1129
1130        let node = node.unwrap();
1131        assert!(node.local_addr().is_some());
1132
1133        node.shutdown().await;
1134    }
1135
1136    #[tokio::test]
1137    async fn test_node_with_peers() {
1138        let peers = vec!["127.0.0.1:9000".parse().unwrap()];
1139        let node = Node::with_peers(peers).await;
1140        assert!(node.is_ok(), "Node::with_peers() should succeed");
1141
1142        node.unwrap().shutdown().await;
1143    }
1144
1145    #[tokio::test]
1146    async fn test_node_with_config() {
1147        let addr: SocketAddr = "127.0.0.1:0".parse().unwrap();
1148        let config = NodeConfig::builder().bind_addr(addr).build();
1149
1150        let node = Node::with_config(config).await;
1151        assert!(node.is_ok(), "Node::with_config() should succeed");
1152
1153        node.unwrap().shutdown().await;
1154    }
1155
1156    #[test]
1157    fn test_node_config_max_message_size_propagates_to_p2p_config() {
1158        let config = NodeConfig::builder()
1159            .max_message_size(10 * 1024 * 1024)
1160            .build();
1161        let p2p_config = node_config_to_p2p_config(config).unwrap();
1162        assert_eq!(p2p_config.max_message_size, 10 * 1024 * 1024);
1163    }
1164
1165    #[test]
1166    fn test_node_config_rejects_zero_max_message_size() {
1167        let config = NodeConfig::builder().max_message_size(0).build();
1168        let err = node_config_to_p2p_config(config).unwrap_err();
1169        assert!(err.to_string().contains("max_message_size"));
1170    }
1171
1172    /// Issue #206: NodeConfig's mDNS knobs must reach the P2pConfig
1173    /// discovery policy, otherwise `Node` embedders have no way to
1174    /// isolate co-located planes that share the default `ant-quic`
1175    /// mDNS service (namespace filter only fires once one is set).
1176    #[test]
1177    fn test_node_config_mdns_namespace_propagates_to_p2p_config() {
1178        let config = NodeConfig::builder().mdns_namespace("testnet").build();
1179        let p2p_config = node_config_to_p2p_config(config).unwrap();
1180        let mdns = p2p_config.discovery.mdns.expect("mDNS config should exist");
1181        assert_eq!(mdns.namespace.as_deref(), Some("testnet"));
1182        // Untouched knobs keep the ant-quic defaults.
1183        assert!(mdns.enabled);
1184    }
1185
1186    #[test]
1187    fn test_node_config_mdns_enabled_propagates_to_p2p_config() {
1188        let config = NodeConfig::builder().mdns_enabled(false).build();
1189        let p2p_config = node_config_to_p2p_config(config).unwrap();
1190        let mdns = p2p_config.discovery.mdns.expect("mDNS config should exist");
1191        assert!(!mdns.enabled);
1192    }
1193
1194    #[test]
1195    fn test_node_config_default_keeps_default_mdns_policy() {
1196        let p2p_config = node_config_to_p2p_config(NodeConfig::default()).unwrap();
1197        let mdns = p2p_config.discovery.mdns.expect("mDNS config should exist");
1198        assert!(mdns.enabled);
1199        assert_eq!(mdns.namespace, None);
1200    }
1201
1202    #[tokio::test]
1203    async fn test_node_status() {
1204        let node = Node::new().await.unwrap();
1205        let status = node.status().await;
1206
1207        // Check status fields are populated
1208        assert_ne!(status.peer_id.0, [0u8; 32]);
1209        assert_eq!(status.connected_peers, 0); // No connections yet
1210        assert!(!status.port_mapping_active);
1211        assert_eq!(status.port_mapping_addr, None);
1212        assert!(status.relay_service_enabled);
1213        assert!(status.coordinator_service_enabled);
1214        assert!(status.bootstrap_service_enabled);
1215        assert!(!status.is_relaying);
1216        assert!(!status.is_coordinating);
1217
1218        node.shutdown().await;
1219    }
1220
1221    #[tokio::test]
1222    async fn test_node_subscribe() {
1223        let node = Node::new().await.unwrap();
1224        let _events = node.subscribe();
1225
1226        // Just verify subscription works
1227        node.shutdown().await;
1228    }
1229
1230    #[tokio::test]
1231    async fn test_node_is_clone() {
1232        let node1 = Node::new().await.unwrap();
1233        let node2 = node1.clone();
1234
1235        // Both should have same peer ID
1236        assert_eq!(node1.peer_id(), node2.peer_id());
1237
1238        node1.shutdown().await;
1239        // node2 still references the same Arc, so shutdown already happened
1240    }
1241
1242    #[tokio::test]
1243    async fn test_node_debug() {
1244        let node = Node::new().await.unwrap();
1245        let debug_str = format!("{:?}", node);
1246        assert!(debug_str.contains("Node"));
1247        assert!(debug_str.contains("peer_id"));
1248
1249        node.shutdown().await;
1250    }
1251
1252    #[tokio::test]
1253    async fn test_node_identity() {
1254        use crate::crypto::raw_public_keys::key_utils::derive_peer_id_from_key_bytes;
1255
1256        let node = Node::new().await.unwrap();
1257
1258        // Verify identity methods
1259        let peer_id = node.peer_id();
1260        let public_key = node.public_key_bytes();
1261
1262        // Peer ID should be derived from public key (ML-DSA-65)
1263        let derived = derive_peer_id_from_key_bytes(public_key).unwrap();
1264        assert_eq!(peer_id, derived);
1265
1266        node.shutdown().await;
1267    }
1268
1269    #[tokio::test]
1270    async fn test_connected_peers_empty() {
1271        let node = Node::new().await.unwrap();
1272        let peers = node.connected_peers().await;
1273        assert!(peers.is_empty());
1274
1275        node.shutdown().await;
1276    }
1277
1278    // Full peer establishment remains exercised in the default-feature matrix.
1279    // The stripped no-default-features lib configuration is a portability/
1280    // compile-surface check and does not guarantee loopback connection success.
1281    #[cfg(all(feature = "platform-verifier", feature = "network-discovery"))]
1282    #[tokio::test]
1283    async fn test_connect_peer_with_addrs_uses_explicit_hint() {
1284        let listener = Node::bind("127.0.0.1:0".parse().unwrap()).await.unwrap();
1285        let dialer = Node::bind("127.0.0.1:0".parse().unwrap()).await.unwrap();
1286
1287        let listener_addr = listener.local_addr().expect("listener addr");
1288        let listener_addr = if listener_addr.ip().is_unspecified() {
1289            SocketAddr::new(
1290                std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST),
1291                listener_addr.port(),
1292            )
1293        } else {
1294            listener_addr
1295        };
1296        let peer_conn = tokio::time::timeout(
1297            Duration::from_secs(30),
1298            dialer.connect_peer_with_addrs(listener.peer_id(), vec![listener_addr]),
1299        )
1300        .await
1301        .expect("connect should not time out")
1302        .expect("dialer should connect using explicit address hint");
1303        assert_eq!(peer_conn.peer_id, listener.peer_id());
1304
1305        let accepted = tokio::time::timeout(std::time::Duration::from_secs(5), listener.accept())
1306            .await
1307            .expect("accept should complete")
1308            .expect("listener should accept");
1309        assert_eq!(accepted.peer_id, dialer.peer_id());
1310
1311        dialer.shutdown().await;
1312        listener.shutdown().await;
1313    }
1314
1315    #[cfg(all(feature = "platform-verifier", feature = "network-discovery"))]
1316    #[tokio::test]
1317    async fn test_connect_peer_uses_upserted_peer_hints() {
1318        let listener = Node::bind("127.0.0.1:0".parse().unwrap()).await.unwrap();
1319        let dialer = Node::bind("127.0.0.1:0".parse().unwrap()).await.unwrap();
1320
1321        let listener_addr = listener.local_addr().expect("listener addr");
1322        let listener_addr = if listener_addr.ip().is_unspecified() {
1323            SocketAddr::new(
1324                std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST),
1325                listener_addr.port(),
1326            )
1327        } else {
1328            listener_addr
1329        };
1330
1331        dialer
1332            .upsert_peer_hints(listener.peer_id(), vec![listener_addr], None)
1333            .await;
1334
1335        let peer_conn = tokio::time::timeout(
1336            Duration::from_secs(30),
1337            dialer.connect_peer(listener.peer_id()),
1338        )
1339        .await
1340        .expect("connect should not time out")
1341        .expect("dialer should connect using upserted peer hints");
1342        assert_eq!(peer_conn.peer_id, listener.peer_id());
1343
1344        let accepted = tokio::time::timeout(std::time::Duration::from_secs(5), listener.accept())
1345            .await
1346            .expect("accept should complete")
1347            .expect("listener should accept");
1348        assert_eq!(accepted.peer_id, dialer.peer_id());
1349
1350        dialer.shutdown().await;
1351        listener.shutdown().await;
1352    }
1353
1354    #[tokio::test]
1355    async fn test_node_error_types() {
1356        // Test error conversions
1357        let err = NodeError::Creation("test".to_string());
1358        assert!(err.to_string().contains("test"));
1359
1360        let err = NodeError::Connection("connection failed".to_string());
1361        assert!(err.to_string().contains("connection"));
1362
1363        let err = NodeError::ShuttingDown;
1364        assert!(err.to_string().contains("shutting down"));
1365    }
1366
1367    #[tokio::test]
1368    async fn test_node_with_keypair_persistence() {
1369        use crate::crypto::raw_public_keys::key_utils::generate_ml_dsa_keypair;
1370
1371        // Generate an ML-DSA-65 keypair
1372        let (public_key, secret_key) = generate_ml_dsa_keypair().unwrap();
1373        let expected_peer_id = derive_peer_id_from_public_key(&public_key);
1374        let expected_public_key_bytes = public_key.as_bytes().to_vec();
1375
1376        // Create node with the keypair
1377        let node = Node::with_keypair(public_key, secret_key).await.unwrap();
1378
1379        // Verify the node uses the same identity
1380        assert_eq!(node.peer_id(), expected_peer_id);
1381        assert_eq!(node.public_key_bytes(), expected_public_key_bytes);
1382
1383        node.shutdown().await;
1384    }
1385
1386    #[tokio::test]
1387    async fn test_node_keypair_via_config() {
1388        use crate::crypto::raw_public_keys::key_utils::generate_ml_dsa_keypair;
1389
1390        // Generate an ML-DSA-65 keypair
1391        let (public_key, secret_key) = generate_ml_dsa_keypair().unwrap();
1392        let expected_peer_id = derive_peer_id_from_public_key(&public_key);
1393        let expected_public_key_bytes = public_key.as_bytes().to_vec();
1394
1395        // Create node via config with keypair
1396        let config = NodeConfig::with_keypair(public_key, secret_key);
1397        let node = Node::with_config(config).await.unwrap();
1398
1399        // Verify the node uses the same identity
1400        assert_eq!(node.peer_id(), expected_peer_id);
1401        assert_eq!(node.public_key_bytes(), expected_public_key_bytes);
1402
1403        node.shutdown().await;
1404    }
1405
1406    #[tokio::test]
1407    async fn test_node_event_bridge_exists() {
1408        let node = Node::new().await.unwrap();
1409
1410        // Subscribe to events - this should work
1411        let mut events = node.subscribe();
1412
1413        // The event channel should be connected (won't receive anything yet,
1414        // but the bridge task should be running)
1415        // We can't easily test event reception without connections,
1416        // but we verify the infrastructure is in place
1417        assert!(events.try_recv().is_err()); // No events yet
1418
1419        node.shutdown().await;
1420    }
1421
1422    #[tokio::test]
1423    async fn test_node_with_host_identity() {
1424        use crate::host_identity::HostIdentity;
1425
1426        // Create a temporary directory for storage
1427        let temp_dir =
1428            std::env::temp_dir().join(format!("ant-quic-test-node-{}", std::process::id()));
1429        let _ = std::fs::create_dir_all(&temp_dir);
1430
1431        // Generate a HostIdentity
1432        let host = HostIdentity::generate();
1433        let network_id = b"test-network";
1434
1435        // Create first node with host identity
1436        let node1 = Node::with_host_identity(&host, network_id, &temp_dir)
1437            .await
1438            .unwrap();
1439        let peer_id_1 = node1.peer_id();
1440        let public_key_1 = node1.public_key_bytes().to_vec();
1441
1442        // Verify the node is running
1443        assert!(node1.is_running());
1444
1445        // Shutdown and cleanup
1446        node1.shutdown().await;
1447
1448        // Create second node with same host identity - should have same identity
1449        let node2 = Node::with_host_identity(&host, network_id, &temp_dir)
1450            .await
1451            .unwrap();
1452        let peer_id_2 = node2.peer_id();
1453        let public_key_2 = node2.public_key_bytes().to_vec();
1454
1455        // Verify both nodes have the same identity
1456        assert_eq!(peer_id_1, peer_id_2);
1457        assert_eq!(public_key_1, public_key_2);
1458
1459        node2.shutdown().await;
1460
1461        // Cleanup temp directory
1462        let _ = std::fs::remove_dir_all(&temp_dir);
1463    }
1464
1465    #[tokio::test]
1466    async fn test_node_host_identity_per_network_isolation() {
1467        use crate::host_identity::HostIdentity;
1468
1469        // Create a temporary directory for storage
1470        let temp_dir =
1471            std::env::temp_dir().join(format!("ant-quic-test-isolation-{}", std::process::id()));
1472        let _ = std::fs::create_dir_all(&temp_dir);
1473
1474        // Generate a HostIdentity
1475        let host = HostIdentity::generate();
1476
1477        // Create nodes with different network IDs
1478        let node1 = Node::with_host_identity(&host, b"network-1", &temp_dir)
1479            .await
1480            .unwrap();
1481        let peer_id_1 = node1.peer_id();
1482
1483        let node2 = Node::with_host_identity(&host, b"network-2", &temp_dir)
1484            .await
1485            .unwrap();
1486        let peer_id_2 = node2.peer_id();
1487
1488        // Different networks should have different identities (privacy)
1489        assert_ne!(peer_id_1, peer_id_2);
1490
1491        node1.shutdown().await;
1492        node2.shutdown().await;
1493
1494        // Cleanup temp directory
1495        let _ = std::fs::remove_dir_all(&temp_dir);
1496    }
1497}