firecloud_net/

node.rs

//! FireCloud P2P Node

use crate::{
    FireCloudBehaviour, FireCloudEvent, HealthMonitor, NetError, NetResult, 
    TransferRequest, TransferResponse,
    MessageRequest, MessageResponse,
    relay_manager::{RelayManager, RelayManagerConfig},
};
use futures::StreamExt;
use libp2p::{
    dcutr,
    gossipsub::{self, IdentTopic},
    identify,
    identity::Keypair,
    kad,
    mdns,
    multiaddr::Protocol,
    ping,
    request_response::{self, OutboundRequestId, ResponseChannel},
    swarm::{SwarmEvent, dial_opts::DialOpts},
    Multiaddr, PeerId, Swarm,
};
use std::{collections::{HashSet, HashMap}, time::Duration};
use tokio::sync::mpsc;
use tracing::{debug, info, warn};

/// Node configuration
#[derive(Debug, Clone)]
pub struct NodeConfig {
    /// Port to listen on (0 for random)
    pub port: u16,
    /// Enable mDNS local discovery
    pub enable_mdns: bool,
    /// Bootstrap peers for Kademlia
    pub bootstrap_peers: Vec<Multiaddr>,
    /// Bootstrap relay addresses for NAT traversal
    pub bootstrap_relays: Vec<Multiaddr>,
}

impl Default for NodeConfig {
    fn default() -> Self {
        Self {
            port: 0,
            enable_mdns: true,
            bootstrap_peers: Vec::new(),
            bootstrap_relays: Vec::new(),
        }
    }
}

/// Events emitted by the node
#[derive(Debug)]
pub enum NodeEvent {
    /// A new peer was discovered
    PeerDiscovered(PeerId),
    /// A peer disconnected
    PeerDisconnected(PeerId),
    /// Received a message on a topic
    Message {
        source: PeerId,
        topic: String,
        data: Vec<u8>,
    },
    /// Node is listening on an address
    Listening(Multiaddr),
    /// Received a transfer request from a peer (needs response via respond_transfer)
    TransferRequest {
        peer: PeerId,
        request: TransferRequest,
        channel: ResponseChannel<TransferResponse>,
    },
    /// Received a response to our transfer request
    TransferResponse {
        peer: PeerId,
        request_id: OutboundRequestId,
        response: TransferResponse,
    },
    /// Transfer request failed
    TransferFailed {
        peer: PeerId,
        request_id: OutboundRequestId,
        error: String,
    },
    /// DHT: Found providers for a key
    ProvidersFound {
        key: String,
        providers: Vec<PeerId>,
    },
    /// DHT: Successfully started providing
    ProvideStarted {
        key: String,
    },
    /// DHT: Record found in DHT
    RecordFound {
        key: String,
        value: Vec<u8>,
    },
    /// DHT: Record stored successfully
    RecordStored {
        key: String,
    },
    /// DHT: Query failed
    DhtQueryFailed {
        key: String,
        error: String,
    },
    
    // ========================================================================
    // Messaging Events (Phase 1.5)
    // ========================================================================
    
    /// Received a message request from a peer (needs response via respond_message)
    MessageRequest {
        peer: PeerId,
        request: MessageRequest,
        channel: ResponseChannel<MessageResponse>,
    },
    /// Received a response to our message request
    MessageResponse {
        peer: PeerId,
        request_id: OutboundRequestId,
        response: MessageResponse,
    },
    /// Message request failed
    MessageFailed {
        peer: PeerId,
        request_id: OutboundRequestId,
        error: String,
    },
}

/// A FireCloud P2P node
pub struct FireCloudNode {
    swarm: Swarm<FireCloudBehaviour>,
    local_peer_id: PeerId,
    known_peers: HashSet<PeerId>,
    /// Peers discovered via mDNS (local network)
    local_peers: HashSet<PeerId>,
    /// Latest measured round-trip times per peer (from ping events)
    peer_latencies: HashMap<PeerId, Duration>,
    /// Health monitoring for all peers
    health_monitor: HealthMonitor,
    /// Relay connection manager for NAT traversal
    relay_manager: RelayManager,
    event_tx: mpsc::Sender<NodeEvent>,
    event_rx: mpsc::Receiver<NodeEvent>,
}

impl FireCloudNode {
    /// Create a new FireCloud node
    pub async fn new(config: NodeConfig) -> NetResult<Self> {
        // Generate identity
        let local_key = Keypair::generate_ed25519();
        let local_peer_id = PeerId::from(local_key.public());

        info!("Local peer ID: {}", local_peer_id);

        // Build swarm with QUIC transport + relay client
        // The .with_relay_client() method handles transport integration automatically
        let swarm = libp2p::SwarmBuilder::with_existing_identity(local_key.clone())
            .with_tokio()
            .with_quic()
            .with_relay_client(
                libp2p::noise::Config::new,
                libp2p::yamux::Config::default,
            )
            .map_err(|e| NetError::Other(e.to_string()))?
            .with_behaviour(|keypair, relay_client| {
                FireCloudBehaviour::new_with_relay(local_peer_id, keypair, relay_client)
                    .expect("Failed to create FireCloudBehaviour")
            })
            .map_err(|e| NetError::Other(e.to_string()))?
            .with_swarm_config(|c| c.with_idle_connection_timeout(Duration::from_secs(60)))
            .build();

        let (event_tx, event_rx) = mpsc::channel(100);

        // Initialize relay manager with bootstrap relays
        let relay_config = RelayManagerConfig {
            bootstrap_relays: config.bootstrap_relays.clone(),
            min_active_relays: 2,
            max_relay_connections: 5,
            reconnect_interval: Duration::from_secs(30),
            connection_timeout: Duration::from_secs(10),
        };
        let relay_manager = RelayManager::new(relay_config);

        let mut node = Self {
            swarm,
            local_peer_id,
            known_peers: HashSet::new(),
            local_peers: HashSet::new(),
            peer_latencies: HashMap::new(),
            health_monitor: HealthMonitor::new(),
            relay_manager,
            event_tx,
            event_rx,
        };

        // Start listening
        let listen_addr: Multiaddr = format!("/ip4/0.0.0.0/udp/{}/quic-v1", config.port)
            .parse()
            .map_err(|e| NetError::Other(format!("Invalid address: {}", e)))?;

        node.swarm
            .listen_on(listen_addr)
            .map_err(|e| NetError::Transport(e.to_string()))?;

        // Bootstrap Kademlia if peers provided
        for addr in config.bootstrap_peers {
            if let Some(peer_id) = extract_peer_id(&addr) {
                node.swarm.behaviour_mut().kademlia.add_address(&peer_id, addr);
            }
        }

        // Connect to bootstrap relays for NAT traversal
        node.connect_to_bootstrap_relays();

        Ok(node)
    }

    /// Connect to bootstrap relay nodes
    fn connect_to_bootstrap_relays(&mut self) {
        let relays_to_connect = self.relay_manager.relays_to_connect();
        
        for relay_addr in relays_to_connect {
            info!("Connecting to bootstrap relay: {}", relay_addr);
            
            // Mark as connecting in relay manager
            self.relay_manager.mark_connecting(&relay_addr);
            
            // Dial the relay
            match self.swarm.dial(relay_addr.clone()) {
                Ok(_) => {
                    debug!("Dialing relay at {}", relay_addr);
                }
                Err(e) => {
                    warn!("Failed to dial relay {}: {}", relay_addr, e);
                    self.relay_manager.mark_failed(&relay_addr, format!("Dial failed: {}", e));
                }
            }
        }
        
        // Log relay stats
        let stats = self.relay_manager.stats();
        info!(
            "Relay connections: {} total, {} connected, {} listening, {} failed",
            stats.total_relays, stats.connected_relays, stats.listening_relays, stats.failed_relays
        );
    }

    /// Get the latest measured RTT for a peer, if available
    pub fn peer_latency(&self, peer: &PeerId) -> Option<Duration> {
        self.peer_latencies.get(peer).cloned()
    }

    /// Select the peer with the lowest recorded latency from the provided list
    /// Returns None if no latencies are known for the provided peers
    pub fn select_lowest_latency_peer(&self, candidates: &[PeerId]) -> Option<PeerId> {
        candidates
            .iter()
            .filter_map(|p| self.peer_latencies.get(p).map(|d| (p, d)))
            .min_by_key(|(_, d)| *d)
            .map(|(p, _)| *p)
    }

    /// Check if a peer was discovered via mDNS (i.e., is on the local network)
    pub fn is_local_peer(&self, peer: &PeerId) -> bool {
        self.local_peers.contains(peer)
    }

    /// Get list of local (mDNS-discovered) peers
    pub fn local_peers(&self) -> &HashSet<PeerId> {
        &self.local_peers
    }

    /// Choose the best peer from candidates for chunk retrieval.
    /// Priority:
    /// 1. Local peers (mDNS-discovered) with lowest latency
    /// 2. Any peer with lowest latency
    /// 3. First candidate if no latency info available
    pub fn choose_best_peer(&self, candidates: &[PeerId]) -> Option<PeerId> {
        if candidates.is_empty() {
            return None;
        }

        // Filter to local peers first
        let local_candidates: Vec<PeerId> = candidates
            .iter()
            .filter(|p| self.local_peers.contains(p))
            .cloned()
            .collect();

        // Try to pick lowest-latency local peer
        if !local_candidates.is_empty() {
            if let Some(best) = self.select_lowest_latency_peer(&local_candidates) {
                return Some(best);
            }
            // If no latency info, just return first local peer
            return Some(local_candidates[0]);
        }

        // No local peers; pick lowest-latency from all candidates
        if let Some(best) = self.select_lowest_latency_peer(candidates) {
            return Some(best);
        }

        // Fallback: return first candidate
        Some(candidates[0])
    }

    /// Get local peer ID
    pub fn local_peer_id(&self) -> PeerId {
        self.local_peer_id
    }

    /// Get known peers
    pub fn known_peers(&self) -> &HashSet<PeerId> {
        &self.known_peers
    }

    /// Get the number of connected peers
    pub fn connected_peers_count(&self) -> usize {
        self.swarm.connected_peers().count()
    }

    /// Get all connected peer IDs
    pub fn connected_peers(&self) -> Vec<PeerId> {
        self.swarm.connected_peers().cloned().collect()
    }

    /// Get Kademlia routing table size (approximate)
    pub fn kademlia_peers_count(&self) -> usize {
        // Return known peers count as approximation
        // The actual Kademlia kbuckets iteration requires mutable access
        self.known_peers.len()
    }

    /// Add a bootstrap peer to Kademlia
    pub fn add_bootstrap_peer(&mut self, peer_id: &PeerId, addr: Multiaddr) {
        self.swarm.behaviour_mut().kademlia.add_address(peer_id, addr);
        info!("Added bootstrap peer: {}", peer_id);
    }

    /// Trigger Kademlia bootstrap process
    pub fn bootstrap(&mut self) -> NetResult<kad::QueryId> {
        self.swarm
            .behaviour_mut()
            .kademlia
            .bootstrap()
            .map_err(|e| NetError::Protocol(format!("Bootstrap failed: {:?}", e)))
    }

    /// Subscribe to a GossipSub topic
    pub fn subscribe(&mut self, topic: &str) -> NetResult<()> {
        let topic = IdentTopic::new(topic);
        self.swarm
            .behaviour_mut()
            .gossipsub
            .subscribe(&topic)
            .map_err(|e| NetError::Protocol(format!("Subscribe failed: {}", e)))?;
        Ok(())
    }

    /// Publish a message to a topic
    pub fn publish(&mut self, topic: &str, data: Vec<u8>) -> NetResult<()> {
        let topic = IdentTopic::new(topic);
        self.swarm
            .behaviour_mut()
            .gossipsub
            .publish(topic, data)
            .map_err(|e| NetError::Protocol(format!("Publish failed: {}", e)))?;
        Ok(())
    }

    /// Dial a peer by address
    pub fn dial(&mut self, addr: Multiaddr) -> NetResult<()> {
        self.swarm
            .dial(DialOpts::unknown_peer_id().address(addr).build())
            .map_err(|e| NetError::Dial(e.to_string()))?;
        Ok(())
    }

    /// Get reference to health monitor
    pub fn health_monitor(&self) -> &HealthMonitor {
        &self.health_monitor
    }

    /// Get best peer for transfer (uses health monitor)
    pub fn get_best_peer_for_transfer(&self) -> Option<PeerId> {
        self.health_monitor.get_best_peer()
    }

    /// Get healthy peers sorted by health score
    pub fn get_healthy_peers(&self) -> Vec<PeerId> {
        self.health_monitor.get_healthy_peers()
    }

    /// Cleanup offline peers from health monitor
    pub fn cleanup_offline_peers(&mut self) {
        self.health_monitor.cleanup_offline_peers();
    }

    /// Send a transfer request to a peer
    pub fn send_transfer_request(&mut self, peer: &PeerId, request: TransferRequest) -> OutboundRequestId {
        self.swarm.behaviour_mut().transfer.send_request(peer, request)
    }

    /// Send a response to a transfer request
    pub fn respond_transfer(&mut self, channel: ResponseChannel<TransferResponse>, response: TransferResponse) -> NetResult<()> {
        self.swarm.behaviour_mut().transfer.send_response(channel, response)
            .map_err(|_| NetError::Protocol("Failed to send response".to_string()))
    }

    /// Request a chunk from a peer
    pub fn request_chunk(&mut self, peer: &PeerId, hash: String) -> OutboundRequestId {
        self.send_transfer_request(peer, TransferRequest::GetChunk { hash })
    }

    // ========================================================================
    // Messaging API (Phase 1.5)
    // ========================================================================

    /// Send a message request to a peer
    pub fn send_message_request(&mut self, peer: &PeerId, request: MessageRequest) -> OutboundRequestId {
        self.swarm.behaviour_mut().messaging.send_request(peer, request)
    }

    /// Send a response to a message request
    pub fn respond_message(&mut self, channel: ResponseChannel<MessageResponse>, response: MessageResponse) -> NetResult<()> {
        self.swarm.behaviour_mut().messaging.send_response(channel, response)
            .map_err(|_| NetError::Protocol("Failed to send message response".to_string()))
    }

    /// Send a friend request to a peer
    pub fn send_friend_request(&mut self, peer: &PeerId, name: Option<String>) -> OutboundRequestId {
        self.send_message_request(peer, MessageRequest::FriendRequest { name })
    }

    /// Accept a friend request from a peer
    pub fn send_friend_accept(&mut self, peer: &PeerId) -> OutboundRequestId {
        self.send_message_request(peer, MessageRequest::FriendAccept)
    }

    /// Send a direct message to a friend (content already padded to 1KB!)
    pub fn send_direct_message(
        &mut self,
        peer: &PeerId,
        content: Vec<u8>,
        message_id: String,
        timestamp: i64,
    ) -> OutboundRequestId {
        self.send_message_request(
            peer,
            MessageRequest::DirectMessage {
                content,
                message_id,
                timestamp,
            },
        )
    }

    /// Ping a peer to check if they're online
    pub fn ping_peer(&mut self, peer: &PeerId) -> OutboundRequestId {
        self.send_message_request(peer, MessageRequest::Ping)
    }

    /// Store a chunk on a peer
    pub fn store_chunk_on_peer(&mut self, peer: &PeerId, hash: String, data: Vec<u8>, original_size: u64) -> OutboundRequestId {
        self.send_transfer_request(peer, TransferRequest::StoreChunk { hash, data, original_size })
    }

    /// Check if a peer has a chunk
    pub fn check_chunk(&mut self, peer: &PeerId, hash: String) -> OutboundRequestId {
        self.send_transfer_request(peer, TransferRequest::HasChunk { hash })
    }

    // ==================== DHT (Kademlia) Methods ====================

    /// Announce this node as a provider for a file (by file_id)
    /// Other nodes can then find us via get_providers
    pub fn announce_file(&mut self, file_id: &str) -> kad::QueryId {
        let key = kad::RecordKey::new(&file_id.as_bytes());
        self.swarm.behaviour_mut().kademlia.start_providing(key)
            .expect("Failed to start providing")
    }

    /// Stop announcing as a provider for a file
    pub fn stop_announcing_file(&mut self, file_id: &str) {
        let key = kad::RecordKey::new(&file_id.as_bytes());
        self.swarm.behaviour_mut().kademlia.stop_providing(&key);
    }

    /// Find providers for a file (by file_id)
    /// Results will come via NodeEvent::ProvidersFound
    pub fn find_file_providers(&mut self, file_id: &str) -> kad::QueryId {
        let key = kad::RecordKey::new(&file_id.as_bytes());
        self.swarm.behaviour_mut().kademlia.get_providers(key)
    }

    /// Store a record in the DHT (for manifest metadata)
    /// Results will come via NodeEvent::RecordStored
    pub fn put_dht_record(&mut self, key: &str, value: Vec<u8>) -> kad::QueryId {
        let record = kad::Record {
            key: kad::RecordKey::new(&key.as_bytes()),
            value,
            publisher: Some(self.local_peer_id),
            expires: None,
        };
        self.swarm.behaviour_mut().kademlia.put_record(record, kad::Quorum::One)
            .expect("Failed to put record")
    }

    /// Get a record from the DHT
    /// Results will come via NodeEvent::RecordFound
    pub fn get_dht_record(&mut self, key: &str) -> kad::QueryId {
        let key = kad::RecordKey::new(&key.as_bytes());
        self.swarm.behaviour_mut().kademlia.get_record(key)
    }

    /// Bootstrap Kademlia DHT (find closest peers to self)
    pub fn bootstrap_dht(&mut self) -> Result<kad::QueryId, kad::NoKnownPeers> {
        self.swarm.behaviour_mut().kademlia.bootstrap()
    }

    // ==================== End DHT Methods ====================

    /// Run the node event loop
    pub async fn run(&mut self) {
        loop {
            tokio::select! {
                event = self.swarm.select_next_some() => {
                    self.handle_swarm_event(event).await;
                }
            }
        }
    }

    /// Poll for the next event (non-blocking version for integration)
    pub async fn poll_event(&mut self) -> Option<NodeEvent> {
        tokio::select! {
            event = self.swarm.select_next_some() => {
                self.handle_swarm_event(event).await;
                self.event_rx.try_recv().ok()
            }
            event = self.event_rx.recv() => {
                event
            }
        }
    }

    async fn handle_swarm_event(&mut self, event: SwarmEvent<FireCloudEvent>) {
        match event {
            SwarmEvent::NewListenAddr { address, .. } => {
                info!("Listening on {}/p2p/{}", address, self.local_peer_id);
                let _ = self.event_tx.send(NodeEvent::Listening(address)).await;
            }

            SwarmEvent::Behaviour(FireCloudEvent::Mdns(mdns::Event::Discovered(peers))) => {
                for (peer_id, addr) in peers {
                    if !self.known_peers.contains(&peer_id) {
                        info!("mDNS discovered (local): {} at {}", peer_id, addr);
                        self.known_peers.insert(peer_id);
                        // Mark as local peer (discovered via mDNS)
                        self.local_peers.insert(peer_id);
                        
                        // Add to health monitor as local peer
                        self.health_monitor.add_peer(peer_id, true);
                        
                        // Add to Kademlia
                        self.swarm.behaviour_mut().kademlia.add_address(&peer_id, addr);
                        
                        let _ = self.event_tx.send(NodeEvent::PeerDiscovered(peer_id)).await;
                    }
                }
            }

            SwarmEvent::Behaviour(FireCloudEvent::Mdns(mdns::Event::Expired(peers))) => {
                for (peer_id, _) in peers {
                    debug!("mDNS peer expired: {}", peer_id);
                    // Remove from local_peers but keep in known_peers (may reconnect via other means)
                    self.local_peers.remove(&peer_id);
                    // Note: Keep in health monitor, will be cleaned up by periodic task if truly offline
                }
            }

            SwarmEvent::Behaviour(FireCloudEvent::Kademlia(kad::Event::RoutingUpdated { 
                peer, 
                addresses, 
                .. 
            })) => {
                debug!("Kademlia routing updated: {} with {} addresses", peer, addresses.len());
                if !self.known_peers.contains(&peer) {
                    self.known_peers.insert(peer);
                    let _ = self.event_tx.send(NodeEvent::PeerDiscovered(peer)).await;
                }
            }

            // Handle Kademlia query results
            SwarmEvent::Behaviour(FireCloudEvent::Kademlia(kad::Event::OutboundQueryProgressed { 
                result, 
                .. 
            })) => {
                match result {
                    kad::QueryResult::GetProviders(Ok(kad::GetProvidersOk::FoundProviders { key, providers, .. })) => {
                        let key_str = String::from_utf8_lossy(key.as_ref()).to_string();
                        let provider_ids: Vec<PeerId> = providers.into_iter().collect();
                        info!("Found {} providers for {}", provider_ids.len(), key_str);
                        let _ = self.event_tx.send(NodeEvent::ProvidersFound {
                            key: key_str,
                            providers: provider_ids,
                        }).await;
                    }
                    kad::QueryResult::GetProviders(Ok(kad::GetProvidersOk::FinishedWithNoAdditionalRecord { .. })) => {
                        debug!("GetProviders finished with no additional records");
                    }
                    kad::QueryResult::GetProviders(Err(e)) => {
                        warn!("GetProviders failed: {:?}", e);
                        let _ = self.event_tx.send(NodeEvent::DhtQueryFailed {
                            key: String::from_utf8_lossy(e.key().as_ref()).to_string(),
                            error: format!("{:?}", e),
                        }).await;
                    }
                    kad::QueryResult::StartProviding(Ok(kad::AddProviderOk { key })) => {
                        let key_str = String::from_utf8_lossy(key.as_ref()).to_string();
                        info!("Started providing: {}", key_str);
                        let _ = self.event_tx.send(NodeEvent::ProvideStarted {
                            key: key_str,
                        }).await;
                    }
                    kad::QueryResult::StartProviding(Err(e)) => {
                        warn!("StartProviding failed: {:?}", e);
                    }
                    kad::QueryResult::GetRecord(Ok(kad::GetRecordOk::FoundRecord(peer_record))) => {
                        let key_str = String::from_utf8_lossy(peer_record.record.key.as_ref()).to_string();
                        info!("Found record: {} ({} bytes)", key_str, peer_record.record.value.len());
                        let _ = self.event_tx.send(NodeEvent::RecordFound {
                            key: key_str,
                            value: peer_record.record.value,
                        }).await;
                    }
                    kad::QueryResult::GetRecord(Ok(kad::GetRecordOk::FinishedWithNoAdditionalRecord { .. })) => {
                        debug!("GetRecord finished with no additional records");
                    }
                    kad::QueryResult::GetRecord(Err(e)) => {
                        warn!("GetRecord failed: {:?}", e);
                        let key_bytes = match &e {
                            kad::GetRecordError::NotFound { key, .. } => key.as_ref(),
                            kad::GetRecordError::QuorumFailed { key, .. } => key.as_ref(),
                            kad::GetRecordError::Timeout { key, .. } => key.as_ref(),
                        };
                        let _ = self.event_tx.send(NodeEvent::DhtQueryFailed {
                            key: String::from_utf8_lossy(key_bytes).to_string(),
                            error: format!("{:?}", e),
                        }).await;
                    }
                    kad::QueryResult::PutRecord(Ok(kad::PutRecordOk { key })) => {
                        let key_str = String::from_utf8_lossy(key.as_ref()).to_string();
                        info!("Record stored: {}", key_str);
                        let _ = self.event_tx.send(NodeEvent::RecordStored {
                            key: key_str,
                        }).await;
                    }
                    kad::QueryResult::PutRecord(Err(e)) => {
                        warn!("PutRecord failed: {:?}", e);
                    }
                    kad::QueryResult::Bootstrap(Ok(_)) => {
                        info!("Kademlia bootstrap successful");
                    }
                    kad::QueryResult::Bootstrap(Err(e)) => {
                        warn!("Kademlia bootstrap failed: {:?}", e);
                    }
                    _ => {}
                }
            }

            SwarmEvent::Behaviour(FireCloudEvent::Gossipsub(gossipsub::Event::Message {
                propagation_source,
                message,
                ..
            })) => {
                info!(
                    "Received message from {} on topic {}",
                    propagation_source,
                    message.topic
                );
                let _ = self.event_tx.send(NodeEvent::Message {
                    source: propagation_source,
                    topic: message.topic.to_string(),
                    data: message.data,
                }).await;
            }

            SwarmEvent::Behaviour(FireCloudEvent::Identify(identify::Event::Received {
                peer_id,
                info,
                ..
            })) => {
                debug!(
                    "Identified peer {}: {} with {} addresses",
                    peer_id,
                    info.protocol_version,
                    info.listen_addrs.len()
                );
                
                // IMPORTANT: Add the peer's addresses to Kademlia's routing table
                // This is necessary for DHT peer discovery beyond bootstrap nodes
                for addr in &info.listen_addrs {
                    self.swarm.behaviour_mut().kademlia.add_address(&peer_id, addr.clone());
                }
                debug!("Kademlia routing updated: {} with {} addresses", peer_id, info.listen_addrs.len());
            }

            SwarmEvent::Behaviour(FireCloudEvent::Ping(ping::Event { peer, result, .. })) => {
                match result {
                    Ok(rtt) => {
                        debug!("Ping to {}: {:?}", peer, rtt);
                        // Store/update latest RTT for this peer
                        self.peer_latencies.insert(peer, rtt);
                        // Update health monitor
                        self.health_monitor.record_ping_success(&peer, rtt);
                    }
                    Err(e) => {
                        warn!("Ping to {} failed: {}", peer, e);
                        // Update health monitor
                        self.health_monitor.record_ping_failure(&peer);
                    }
                }
            }

            // Handle relay client events for circuit relay
            SwarmEvent::Behaviour(FireCloudEvent::RelayClient(event)) => {
                use libp2p::relay;
                match event {
                    relay::client::Event::ReservationReqAccepted { relay_peer_id, .. } => {
                        info!("✅ Relay reservation accepted by {}", relay_peer_id);
                        
                        // Find the relay address that matches this peer ID
                        let bootstrap_relays: Vec<_> = self.relay_manager.config().bootstrap_relays.clone();
                        for relay_addr in bootstrap_relays {
                            // Extract peer ID from multiaddr
                            if let Some(peer) = extract_peer_id(&relay_addr) {
                                if peer == relay_peer_id {
                                    self.relay_manager.mark_connected(&relay_addr, relay_peer_id);
                                    
                                    // Build p2p-circuit address for listening
                                    let circuit_addr = relay_addr
                                        .with(libp2p::multiaddr::Protocol::P2pCircuit)
                                        .with(libp2p::multiaddr::Protocol::P2p(self.local_peer_id));
                                    
                                    info!("🔊 Listening on relay circuit: {}", circuit_addr);
                                    
                                    // Listen on the circuit relay address
                                    match self.swarm.listen_on(circuit_addr.clone()) {
                                        Ok(_) => {
                                            self.relay_manager.mark_listening(&relay_peer_id, circuit_addr);
                                        }
                                        Err(e) => {
                                            warn!("Failed to listen on relay circuit: {}", e);
                                        }
                                    }
                                    break;
                                }
                            }
                        }
                    }
                    relay::client::Event::OutboundCircuitEstablished { relay_peer_id, .. } => {
                        info!("🔗 Outbound circuit established via relay {}", relay_peer_id);
                    }
                    relay::client::Event::InboundCircuitEstablished { src_peer_id, .. } => {
                        info!("🔗 Inbound circuit established from {}", src_peer_id);
                    }
                    _ => {
                        debug!("Relay client event: {:?}", event);
                    }
                }
            }

            // Handle DCUtR events for hole punching
            SwarmEvent::Behaviour(FireCloudEvent::Dcutr(dcutr::Event {
                remote_peer_id,
                result,
            })) => {
                match result {
                    Ok(_connection_id) => {
                        info!("✅ Hole punch SUCCESS with {} - direct connection established", remote_peer_id);
                        // Direct connection established - no longer need relay for this peer
                    }
                    Err(error) => {
                        warn!("❌ Hole punch FAILED with {}: {:?}", remote_peer_id, error);
                        // Fall back to relayed connection
                    }
                }
            }

            SwarmEvent::Behaviour(FireCloudEvent::Transfer(event)) => {
                match event {
                    request_response::Event::Message { peer, message } => {
                        match message {
                            request_response::Message::Request { request, channel, .. } => {
                                info!("Transfer request from {}: {:?}", peer, request);
                                // Note: Clone is not implemented for ResponseChannel,
                                // so we need to handle this differently - expose it to caller
                                // The caller needs to handle the request and respond
                                let _ = self.event_tx.send(NodeEvent::TransferRequest {
                                    peer,
                                    request,
                                    channel,
                                }).await;
                            }
                            request_response::Message::Response { request_id, response } => {
                                info!("Transfer response for {:?}: {:?}", request_id, response);
                                let _ = self.event_tx.send(NodeEvent::TransferResponse {
                                    peer,
                                    request_id,
                                    response,
                                }).await;
                            }
                        }
                    }
                    request_response::Event::OutboundFailure { peer, request_id, error, .. } => {
                        warn!("Transfer to {} failed: {:?}", peer, error);
                        let _ = self.event_tx.send(NodeEvent::TransferFailed {
                            peer,
                            request_id,
                            error: format!("{:?}", error),
                        }).await;
                    }
                    request_response::Event::InboundFailure { peer, error, .. } => {
                        warn!("Inbound transfer from {} failed: {:?}", peer, error);
                    }
                    request_response::Event::ResponseSent { peer, .. } => {
                        debug!("Response sent to {}", peer);
                    }
                }
            }

            // ========================================================================
            // Messaging Events (Phase 1.5)
            // ========================================================================
            SwarmEvent::Behaviour(FireCloudEvent::Messaging(event)) => {
                match event {
                    request_response::Event::Message { peer, message } => {
                        match message {
                            request_response::Message::Request { request, channel, .. } => {
                                info!("📨 Message request from {}: {:?}", peer, request);
                                // Forward to caller for handling (friend requests, messages, etc.)
                                let _ = self.event_tx.send(NodeEvent::MessageRequest {
                                    peer,
                                    request,
                                    channel,
                                }).await;
                            }
                            request_response::Message::Response { request_id, response } => {
                                info!("✅ Message response for {:?}: {:?}", request_id, response);
                                let _ = self.event_tx.send(NodeEvent::MessageResponse {
                                    peer,
                                    request_id,
                                    response,
                                }).await;
                            }
                        }
                    }
                    request_response::Event::OutboundFailure { peer, request_id, error, .. } => {
                        warn!("❌ Message to {} failed: {:?}", peer, error);
                        let _ = self.event_tx.send(NodeEvent::MessageFailed {
                            peer,
                            request_id,
                            error: format!("{:?}", error),
                        }).await;
                    }
                    request_response::Event::InboundFailure { peer, error, .. } => {
                        warn!("❌ Inbound message from {} failed: {:?}", peer, error);
                    }
                    request_response::Event::ResponseSent { peer, .. } => {
                        debug!("✅ Message response sent to {}", peer);
                    }
                }
            }

            SwarmEvent::ConnectionEstablished { peer_id, .. } => {
                info!("Connected to {}", peer_id);
                if !self.known_peers.contains(&peer_id) {
                    self.known_peers.insert(peer_id);
                    let _ = self.event_tx.send(NodeEvent::PeerDiscovered(peer_id)).await;
                }
                
                // Check if this is a relay server we're supposed to use
                let bootstrap_relays: Vec<_> = self.relay_manager.config().bootstrap_relays.clone();
                for relay_addr in bootstrap_relays {
                    if let Some(relay_peer) = extract_peer_id(&relay_addr) {
                        if relay_peer == peer_id {
                            info!("🔗 Connected to relay server, requesting reservation...");
                            self.relay_manager.mark_connected(&relay_addr, peer_id);
                            
                            // Listen on relay circuit to trigger reservation
                            let circuit_addr = relay_addr
                                .with(libp2p::multiaddr::Protocol::P2pCircuit);
                            
                            match self.swarm.listen_on(circuit_addr.clone()) {
                                Ok(_) => {
                                    info!("📡 Listening on relay circuit: {}", circuit_addr);
                                }
                                Err(e) => {
                                    warn!("Failed to listen on relay circuit: {}", e);
                                }
                            }
                            break;
                        }
                    }
                }
            }

            SwarmEvent::ConnectionClosed { peer_id, .. } => {
                debug!("Disconnected from {}", peer_id);
                let _ = self.event_tx.send(NodeEvent::PeerDisconnected(peer_id)).await;
            }

            SwarmEvent::OutgoingConnectionError { peer_id, error, .. } => {
                warn!("Outgoing connection error to {:?}: {}", peer_id, error);
            }

            SwarmEvent::IncomingConnectionError { error, .. } => {
                warn!("Incoming connection error: {}", error);
            }

            _ => {}
        }
    }

    // ========== PHASE 2: STORAGE PROVIDER METHODS ==========

    /// Announce this node as a storage provider to the DHT
    pub async fn announce_as_provider(&mut self, available_space: u64) -> NetResult<()> {
        use crate::provider::ProviderInfo;
        
        info!("📢 Announcing as storage provider: {} GB available", available_space / (1024 * 1024 * 1024));

        // Get our listen addresses
        let listen_addrs: Vec<Multiaddr> = self.swarm.listeners().cloned().collect();

        // Create provider info
        let provider_info = ProviderInfo::new(
            self.local_peer_id,
            available_space,
            listen_addrs,
        );

        // Serialize provider info
        let provider_data = bincode::serialize(&provider_info)
            .map_err(|e| NetError::Other(format!("Failed to serialize provider info: {}", e)))?;

        // Store in DHT with key "storage-providers:<peer_id>"
        let key_string = format!("storage-providers:{}", self.local_peer_id);
        let key = libp2p::kad::RecordKey::new(&key_string.as_bytes());
        
        let record = libp2p::kad::Record {
            key: key.clone(),
            value: provider_data,
            publisher: Some(self.local_peer_id),
            expires: None, // We'll refresh periodically
        };

        self.swarm
            .behaviour_mut()
            .kademlia
            .put_record(record, libp2p::kad::Quorum::One)
            .map_err(|e| NetError::Other(format!("Failed to put DHT record: {}", e)))?;

        info!("✓ Provider announcement sent to DHT");
        Ok(())
    }

    /// Find available storage providers from the DHT
    pub async fn find_storage_providers(&mut self, min_count: usize) -> NetResult<Vec<crate::provider::ProviderInfo>> {
        info!("🔍 Searching for {} storage providers...", min_count);

        // For now, we'll use a simplified approach:
        // In a full implementation, we'd query the DHT for all "storage-providers:*" keys
        // For MVP, we'll return known peers and assume they're providers
        
        let mut providers = Vec::new();

        // Get all known peers from Kademlia routing table
        for bucket in self.swarm.behaviour_mut().kademlia.kbuckets() {
            for entry in bucket.iter() {
                let peer_id = entry.node.key.preimage().clone();
                
                // Skip ourselves
                if peer_id == self.local_peer_id {
                    continue;
                }

                // Create provider info for this peer
                // In a full implementation, we'd query their actual provider record
                let provider_info = crate::provider::ProviderInfo::new(
                    peer_id,
                    10 * 1024 * 1024 * 1024, // Assume 10GB available (TODO: query actual)
                    vec![], // We'll get addresses from connection
                );

                providers.push(provider_info);

                if providers.len() >= min_count * 2 {
                    break;
                }
            }
        }

        if providers.is_empty() {
            warn!("⚠️  No storage providers found in DHT!");
            warn!("    Make sure other nodes are running with:");
            warn!("    firecloud storage init --quota 10GB");
            warn!("    firecloud node --port 4001");
        } else {
            info!("✓ Found {} potential storage providers", providers.len());
        }

        Ok(providers)
    }

    /// Send a chunk to a storage provider
    pub async fn send_chunk_to_provider(
        &mut self,
        provider: PeerId,
        chunk_hash_str: &str,
        chunk_data: &[u8],
        original_size: u64,
    ) -> NetResult<()> {
        use crate::protocol::TransferRequest;

        debug!("📤 Sending chunk {}... to provider {}", &chunk_hash_str[..16], provider);

        let request = TransferRequest::StoreChunk {
            hash: chunk_hash_str.to_string(),
            data: chunk_data.to_vec(),
            original_size,
        };

        // Send via request-response protocol
        self.swarm
            .behaviour_mut()
            .transfer
            .send_request(&provider, request);

        // In a full implementation, we'd wait for response
        // For now, we'll assume success
        debug!("✓ Chunk sent to provider");
        
        Ok(())
    }
}


/// Extract peer ID from a multiaddr if present
fn extract_peer_id(addr: &Multiaddr) -> Option<PeerId> {
    addr.iter().find_map(|p| {
        if let Protocol::P2p(peer_id) = p {
            Some(peer_id)
        } else {
            None
        }
    })
}