ferripfs_network/

host.rs

// Ported from: kubo/core/node/libp2p/host.go
// Kubo version: v0.39.0
//
// Original work: Copyright (c) Protocol Labs, Inc.
// Port: Copyright (c) 2026 ferripfs contributors
// SPDX-License-Identifier: MIT OR Apache-2.0

//! Network host initialization and management.

use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;

use futures::StreamExt;
use libp2p::{identity::Keypair, swarm::SwarmEvent, Multiaddr, PeerId, Swarm};
use tokio::sync::{mpsc, oneshot, RwLock};

use crate::{
    behavior::FerripfsBehavior,
    swarm::{parse_multiaddr, parse_peer_id, SwarmBuilder, SwarmConfig},
    ConnectedPeer, NetworkError, NetworkResult, PeerInfo, PingResult,
};

/// Configuration for creating a network host.
#[derive(Debug, Clone)]
pub struct HostConfig {
    /// Keypair for node identity
    pub keypair: Keypair,
    /// Addresses to listen on
    pub listen_addrs: Vec<String>,
    /// Bootstrap peers
    pub bootstrap_peers: Vec<String>,
    /// Swarm configuration
    pub swarm_config: SwarmConfig,
}

impl HostConfig {
    /// Create host config from ferripfs config.
    pub fn from_config(config: &ferripfs_config::Config, keypair: Keypair) -> Self {
        Self {
            keypair,
            listen_addrs: config.addresses.swarm.clone(),
            bootstrap_peers: config.bootstrap.clone(),
            swarm_config: SwarmConfig::from_config(config),
        }
    }
}

/// Commands that can be sent to the network host.
#[derive(Debug)]
pub enum HostCommand {
    /// Get local peer info
    GetPeerInfo(oneshot::Sender<PeerInfo>),
    /// Get connected peers
    GetConnectedPeers(oneshot::Sender<Vec<ConnectedPeer>>),
    /// Connect to a peer
    Connect(Multiaddr, oneshot::Sender<Result<(), String>>),
    /// Disconnect from a peer
    Disconnect(PeerId, oneshot::Sender<Result<(), String>>),
    /// Get known addresses for all peers
    GetKnownAddresses(oneshot::Sender<HashMap<String, Vec<String>>>),
    /// Ping a peer
    Ping(PeerId, oneshot::Sender<PingResult>),
    /// Get listen addresses
    GetListenAddresses(oneshot::Sender<Vec<String>>),
    /// Shutdown the host
    Shutdown(oneshot::Sender<()>),
    /// Find providers for a CID (DHT)
    FindProviders(Vec<u8>, oneshot::Sender<crate::DhtQueryResult>),
    /// Find a peer in the DHT
    FindPeer(PeerId, oneshot::Sender<crate::DhtQueryResult>),
    /// Provide content to the DHT
    Provide(Vec<u8>, oneshot::Sender<crate::DhtQueryResult>),
    /// Put a value in the DHT
    PutValue(Vec<u8>, Vec<u8>, oneshot::Sender<crate::DhtQueryResult>),
    /// Get a value from the DHT
    GetValue(Vec<u8>, oneshot::Sender<crate::DhtQueryResult>),
    /// Get DHT statistics
    GetDhtStats(oneshot::Sender<crate::DhtStats>),
    /// Bootstrap the DHT
    Bootstrap(oneshot::Sender<Result<(), String>>),
}

/// Network host providing high-level network operations.
pub struct NetworkHost {
    /// Local peer ID
    peer_id: PeerId,
    /// Command sender
    cmd_tx: mpsc::Sender<HostCommand>,
    /// Listen addresses (updated by swarm)
    #[allow(dead_code)]
    listen_addrs: Arc<RwLock<Vec<Multiaddr>>>,
}

impl NetworkHost {
    /// Create and start a new network host.
    pub async fn start(config: HostConfig) -> NetworkResult<Self> {
        let local_peer_id = PeerId::from(config.keypair.public());
        let _local_public_key = config.keypair.public();

        // Parse listen addresses
        let listen_addrs: Vec<Multiaddr> = config
            .listen_addrs
            .iter()
            .filter_map(|s| parse_multiaddr(s).ok())
            .collect();

        // Build swarm
        let mut swarm = SwarmBuilder::new(config.keypair.clone())
            .with_config(config.swarm_config)
            .with_listen_addrs(listen_addrs.clone())
            .build()?;

        // Start listening on configured addresses
        for addr in &listen_addrs {
            swarm
                .listen_on(addr.clone())
                .map_err(|e| NetworkError::Transport(e.to_string()))?;
        }

        // Add bootstrap peers to Kademlia
        for peer_str in &config.bootstrap_peers {
            if let Ok((peer_id, addr)) = crate::swarm::parse_bootstrap_peer(peer_str) {
                swarm.behaviour_mut().add_address(&peer_id, addr);
            }
        }

        // Create command channel
        let (cmd_tx, cmd_rx) = mpsc::channel(32);
        let actual_listen_addrs = Arc::new(RwLock::new(Vec::new()));
        let listen_addrs_clone = actual_listen_addrs.clone();

        // Spawn swarm event loop
        let keypair = config.keypair.clone();
        tokio::spawn(async move {
            Self::run_event_loop(swarm, cmd_rx, listen_addrs_clone, keypair).await;
        });

        Ok(Self {
            peer_id: local_peer_id,
            cmd_tx,
            listen_addrs: actual_listen_addrs,
        })
    }

    /// Run the swarm event loop.
    async fn run_event_loop(
        mut swarm: Swarm<FerripfsBehavior>,
        mut cmd_rx: mpsc::Receiver<HostCommand>,
        listen_addrs: Arc<RwLock<Vec<Multiaddr>>>,
        keypair: Keypair,
    ) {
        let local_peer_id = PeerId::from(keypair.public());
        let local_public_key = keypair.public();

        // Track pending pings
        let mut pending_pings: HashMap<PeerId, oneshot::Sender<PingResult>> = HashMap::new();

        // Track pending DHT queries
        let mut pending_find_providers: HashMap<
            kad::QueryId,
            oneshot::Sender<crate::DhtQueryResult>,
        > = HashMap::new();
        let mut pending_find_peers: HashMap<kad::QueryId, oneshot::Sender<crate::DhtQueryResult>> =
            HashMap::new();
        let mut pending_provide: HashMap<kad::QueryId, oneshot::Sender<crate::DhtQueryResult>> =
            HashMap::new();
        let mut pending_put_value: HashMap<kad::QueryId, oneshot::Sender<crate::DhtQueryResult>> =
            HashMap::new();
        let mut pending_get_value: HashMap<kad::QueryId, oneshot::Sender<crate::DhtQueryResult>> =
            HashMap::new();

        loop {
            tokio::select! {
                // Handle swarm events
                event = swarm.select_next_some() => {
                    match event {
                        SwarmEvent::NewListenAddr { address, .. } => {
                            let mut addrs = listen_addrs.write().await;
                            addrs.push(address.clone());
                            tracing::info!("Listening on {}", address);
                        }
                        SwarmEvent::Behaviour(event) => {
                            // Handle behavior events
                            Self::handle_behavior_event(
                                event,
                                &mut pending_pings,
                                &mut pending_find_providers,
                                &mut pending_find_peers,
                                &mut pending_provide,
                                &mut pending_put_value,
                                &mut pending_get_value,
                            );
                        }
                        SwarmEvent::ConnectionEstablished { peer_id, .. } => {
                            tracing::debug!("Connected to {}", peer_id);
                        }
                        SwarmEvent::ConnectionClosed { peer_id, .. } => {
                            tracing::debug!("Disconnected from {}", peer_id);
                        }
                        SwarmEvent::IncomingConnection { .. } => {}
                        SwarmEvent::IncomingConnectionError { .. } => {}
                        SwarmEvent::OutgoingConnectionError { .. } => {}
                        SwarmEvent::ExpiredListenAddr { address, .. } => {
                            let mut addrs = listen_addrs.write().await;
                            addrs.retain(|a| a != &address);
                        }
                        SwarmEvent::ListenerClosed { .. } => {}
                        SwarmEvent::ListenerError { .. } => {}
                        SwarmEvent::Dialing { .. } => {}
                        SwarmEvent::NewExternalAddrCandidate { .. } => {}
                        SwarmEvent::ExternalAddrConfirmed { .. } => {}
                        SwarmEvent::ExternalAddrExpired { .. } => {}
                        SwarmEvent::NewExternalAddrOfPeer { .. } => {}
                        _ => {}
                    }
                }
                // Handle commands
                Some(cmd) = cmd_rx.recv() => {
                    match cmd {
                        HostCommand::GetPeerInfo(tx) => {
                            let addrs = listen_addrs.read().await;
                            let info = PeerInfo {
                                id: local_peer_id.to_string(),
                                public_key: bs58::encode(local_public_key.encode_protobuf()).into_string(),
                                addresses: addrs.iter().map(|a| {
                                    format!("{}/p2p/{}", a, local_peer_id)
                                }).collect(),
                                agent_version: crate::AGENT_VERSION.to_string(),
                                protocol_version: crate::PROTOCOL_VERSION.to_string(),
                                protocols: vec![
                                    "/ipfs/id/1.0.0".to_string(),
                                    "/ipfs/ping/1.0.0".to_string(),
                                    "/ipfs/kad/1.0.0".to_string(),
                                ],
                            };
                            let _ = tx.send(info);
                        }
                        HostCommand::GetConnectedPeers(tx) => {
                            let peers: Vec<ConnectedPeer> = swarm
                                .connected_peers()
                                .map(|peer_id| {
                                    let _info = swarm.network_info();
                                    ConnectedPeer {
                                        addr: String::new(), // Would need connection info
                                        peer: peer_id.to_string(),
                                        latency: None,
                                        muxer: Some("yamux".into()),
                                        direction: "unknown".into(),
                                    }
                                })
                                .collect();
                            let _ = tx.send(peers);
                        }
                        HostCommand::Connect(addr, tx) => {
                            match swarm.dial(addr) {
                                Ok(_) => { let _ = tx.send(Ok(())); }
                                Err(e) => { let _ = tx.send(Err(e.to_string())); }
                            }
                        }
                        HostCommand::Disconnect(peer_id, tx) => {
                            let _ = swarm.disconnect_peer_id(peer_id);
                            let _ = tx.send(Ok(()));
                        }
                        HostCommand::GetKnownAddresses(tx) => {
                            let mut addrs: HashMap<String, Vec<String>> = HashMap::new();
                            // Get addresses from Kademlia routing table
                            for bucket in swarm.behaviour_mut().kademlia.kbuckets() {
                                for entry in bucket.iter() {
                                    let peer_id = entry.node.key.preimage().to_string();
                                    let peer_addrs: Vec<String> = entry.node.value
                                        .iter()
                                        .map(|a| a.to_string())
                                        .collect();
                                    addrs.insert(peer_id, peer_addrs);
                                }
                            }
                            let _ = tx.send(addrs);
                        }
                        HostCommand::Ping(peer_id, tx) => {
                            pending_pings.insert(peer_id, tx);
                            // Ping will be handled by ping behavior automatically
                        }
                        HostCommand::GetListenAddresses(tx) => {
                            let addrs = listen_addrs.read().await;
                            let _ = tx.send(addrs.iter().map(|a| a.to_string()).collect());
                        }
                        HostCommand::Shutdown(tx) => {
                            let _ = tx.send(());
                            break;
                        }
                        HostCommand::FindProviders(key, tx) => {
                            let record_key = kad::RecordKey::new(&key);
                            let query_id = swarm.behaviour_mut().get_providers(record_key);
                            pending_find_providers.insert(query_id, tx);
                        }
                        HostCommand::FindPeer(peer_id, tx) => {
                            let query_id = swarm.behaviour_mut().get_closest_peers(peer_id);
                            pending_find_peers.insert(query_id, tx);
                        }
                        HostCommand::Provide(key, tx) => {
                            let record_key = kad::RecordKey::new(&key);
                            match swarm.behaviour_mut().start_providing(record_key) {
                                Ok(query_id) => {
                                    pending_provide.insert(query_id, tx);
                                }
                                Err(e) => {
                                    let _ = tx.send(crate::DhtQueryResult::failure(
                                        "provide",
                                        format!("failed to start providing: {}", e),
                                    ));
                                }
                            }
                        }
                        HostCommand::PutValue(key, value, tx) => {
                            let record_key = kad::RecordKey::new(&key);
                            let query_id = swarm.behaviour_mut().put_value(record_key, value);
                            pending_put_value.insert(query_id, tx);
                        }
                        HostCommand::GetValue(key, tx) => {
                            let record_key = kad::RecordKey::new(&key);
                            let query_id = swarm.behaviour_mut().get_value(record_key);
                            pending_get_value.insert(query_id, tx);
                        }
                        HostCommand::GetDhtStats(tx) => {
                            let (buckets, total_peers) = swarm.behaviour_mut().get_routing_table_info();
                            let stats = crate::DhtStats {
                                name: "kademlia".to_string(),
                                buckets: buckets as u32,
                                total_peers: total_peers as u32,
                                mode: "client".to_string(), // Default mode
                            };
                            let _ = tx.send(stats);
                        }
                        HostCommand::Bootstrap(tx) => {
                            match swarm.behaviour_mut().bootstrap() {
                                Ok(_) => { let _ = tx.send(Ok(())); }
                                Err(e) => { let _ = tx.send(Err(format!("bootstrap failed: {:?}", e))); }
                            }
                        }
                    }
                }
            }
        }
    }

    /// Handle behavior-specific events.
    fn handle_behavior_event(
        event: crate::behavior::FerripfsBehaviorEvent,
        pending_pings: &mut HashMap<PeerId, oneshot::Sender<PingResult>>,
        pending_find_providers: &mut HashMap<kad::QueryId, oneshot::Sender<crate::DhtQueryResult>>,
        pending_find_peers: &mut HashMap<kad::QueryId, oneshot::Sender<crate::DhtQueryResult>>,
        pending_provide: &mut HashMap<kad::QueryId, oneshot::Sender<crate::DhtQueryResult>>,
        pending_put_value: &mut HashMap<kad::QueryId, oneshot::Sender<crate::DhtQueryResult>>,
        pending_get_value: &mut HashMap<kad::QueryId, oneshot::Sender<crate::DhtQueryResult>>,
    ) {
        use crate::behavior::FerripfsBehaviorEvent;

        match event {
            FerripfsBehaviorEvent::Ping(ping::Event { peer, result, .. }) => {
                if let Some(tx) = pending_pings.remove(&peer) {
                    let ping_result = match result {
                        Ok(rtt) => PingResult {
                            success: true,
                            time: rtt.as_millis() as u64,
                            text: format!("Pong received: time={}ms", rtt.as_millis()),
                        },
                        Err(e) => PingResult {
                            success: false,
                            time: 0,
                            text: format!("Ping failed: {}", e),
                        },
                    };
                    let _ = tx.send(ping_result);
                }
            }
            FerripfsBehaviorEvent::Mdns(mdns::Event::Discovered(peers)) => {
                for (peer_id, addr) in peers {
                    tracing::debug!("mDNS discovered {} at {}", peer_id, addr);
                }
            }
            FerripfsBehaviorEvent::Mdns(mdns::Event::Expired(peers)) => {
                for (peer_id, addr) in peers {
                    tracing::debug!("mDNS peer expired {} at {}", peer_id, addr);
                }
            }
            FerripfsBehaviorEvent::Identify(identify::Event::Received {
                peer_id, info, ..
            }) => {
                tracing::debug!("Identified peer {} running {}", peer_id, info.agent_version);
            }
            FerripfsBehaviorEvent::Kademlia(kad::Event::RoutingUpdated { peer, .. }) => {
                tracing::debug!("Kademlia routing updated for {}", peer);
            }
            FerripfsBehaviorEvent::Kademlia(kad::Event::OutboundQueryProgressed {
                id,
                result,
                ..
            }) => match result {
                kad::QueryResult::GetProviders(Ok(kad::GetProvidersOk::FoundProviders {
                    providers,
                    ..
                })) => {
                    if let Some(tx) = pending_find_providers.remove(&id) {
                        let dht_providers: Vec<crate::DhtProvider> = providers
                            .into_iter()
                            .map(|p| crate::DhtProvider {
                                id: p.to_string(),
                                addrs: vec![],
                            })
                            .collect();
                        let _ = tx.send(crate::DhtQueryResult::success("providers", dht_providers));
                    }
                }
                kad::QueryResult::GetProviders(Ok(
                    kad::GetProvidersOk::FinishedWithNoAdditionalRecord { .. },
                )) => {
                    if let Some(tx) = pending_find_providers.remove(&id) {
                        let _ = tx.send(crate::DhtQueryResult::success("providers", vec![]));
                    }
                }
                kad::QueryResult::GetProviders(Err(e)) => {
                    if let Some(tx) = pending_find_providers.remove(&id) {
                        let _ = tx.send(crate::DhtQueryResult::failure(
                            "providers",
                            format!("{:?}", e),
                        ));
                    }
                }
                kad::QueryResult::GetClosestPeers(Ok(kad::GetClosestPeersOk { peers, .. })) => {
                    if let Some(tx) = pending_find_peers.remove(&id) {
                        let dht_peers: Vec<crate::DhtProvider> = peers
                            .into_iter()
                            .map(|p| crate::DhtProvider {
                                id: p.peer_id.to_string(),
                                addrs: p.addrs.iter().map(|a| a.to_string()).collect(),
                            })
                            .collect();
                        let _ = tx.send(crate::DhtQueryResult::success("closest_peers", dht_peers));
                    }
                }
                kad::QueryResult::GetClosestPeers(Err(e)) => {
                    if let Some(tx) = pending_find_peers.remove(&id) {
                        let _ = tx.send(crate::DhtQueryResult::failure(
                            "closest_peers",
                            format!("{:?}", e),
                        ));
                    }
                }
                kad::QueryResult::StartProviding(Ok(kad::AddProviderOk { .. })) => {
                    if let Some(tx) = pending_provide.remove(&id) {
                        let _ = tx.send(crate::DhtQueryResult::success("provide", vec![]));
                    }
                }
                kad::QueryResult::StartProviding(Err(e)) => {
                    if let Some(tx) = pending_provide.remove(&id) {
                        let _ = tx.send(crate::DhtQueryResult::failure(
                            "provide",
                            format!("{:?}", e),
                        ));
                    }
                }
                kad::QueryResult::PutRecord(Ok(kad::PutRecordOk { .. })) => {
                    if let Some(tx) = pending_put_value.remove(&id) {
                        let _ = tx.send(crate::DhtQueryResult::success("put_value", vec![]));
                    }
                }
                kad::QueryResult::PutRecord(Err(e)) => {
                    if let Some(tx) = pending_put_value.remove(&id) {
                        let _ = tx.send(crate::DhtQueryResult::failure(
                            "put_value",
                            format!("{:?}", e),
                        ));
                    }
                }
                kad::QueryResult::GetRecord(Ok(kad::GetRecordOk::FoundRecord(peer_record))) => {
                    if let Some(tx) = pending_get_value.remove(&id) {
                        let value = String::from_utf8_lossy(&peer_record.record.value).to_string();
                        let _ = tx.send(crate::DhtQueryResult::success_with_extra(
                            "get_value",
                            value,
                        ));
                    }
                }
                kad::QueryResult::GetRecord(Ok(
                    kad::GetRecordOk::FinishedWithNoAdditionalRecord { .. },
                )) => {
                    if let Some(tx) = pending_get_value.remove(&id) {
                        let _ = tx.send(crate::DhtQueryResult::failure(
                            "get_value",
                            "record not found".into(),
                        ));
                    }
                }
                kad::QueryResult::GetRecord(Err(e)) => {
                    if let Some(tx) = pending_get_value.remove(&id) {
                        let _ = tx.send(crate::DhtQueryResult::failure(
                            "get_value",
                            format!("{:?}", e),
                        ));
                    }
                }
                _ => {}
            },
            _ => {}
        }
    }

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

    /// Get peer info.
    pub async fn peer_info(&self) -> NetworkResult<PeerInfo> {
        let (tx, rx) = oneshot::channel();
        self.cmd_tx
            .send(HostCommand::GetPeerInfo(tx))
            .await
            .map_err(|_| NetworkError::NotRunning)?;
        rx.await.map_err(|_| NetworkError::NotRunning)
    }

    /// Get connected peers.
    pub async fn connected_peers(&self) -> NetworkResult<Vec<ConnectedPeer>> {
        let (tx, rx) = oneshot::channel();
        self.cmd_tx
            .send(HostCommand::GetConnectedPeers(tx))
            .await
            .map_err(|_| NetworkError::NotRunning)?;
        rx.await.map_err(|_| NetworkError::NotRunning)
    }

    /// Connect to a peer.
    pub async fn connect(&self, addr: &str) -> NetworkResult<()> {
        let multiaddr = parse_multiaddr(addr)?;
        let (tx, rx) = oneshot::channel();
        self.cmd_tx
            .send(HostCommand::Connect(multiaddr, tx))
            .await
            .map_err(|_| NetworkError::NotRunning)?;
        rx.await
            .map_err(|_| NetworkError::NotRunning)?
            .map_err(NetworkError::Connection)
    }

    /// Disconnect from a peer.
    pub async fn disconnect(&self, peer: &str) -> NetworkResult<()> {
        let peer_id = parse_peer_id(peer)?;
        let (tx, rx) = oneshot::channel();
        self.cmd_tx
            .send(HostCommand::Disconnect(peer_id, tx))
            .await
            .map_err(|_| NetworkError::NotRunning)?;
        rx.await
            .map_err(|_| NetworkError::NotRunning)?
            .map_err(NetworkError::Connection)
    }

    /// Get known addresses for all peers.
    pub async fn known_addresses(&self) -> NetworkResult<HashMap<String, Vec<String>>> {
        let (tx, rx) = oneshot::channel();
        self.cmd_tx
            .send(HostCommand::GetKnownAddresses(tx))
            .await
            .map_err(|_| NetworkError::NotRunning)?;
        rx.await.map_err(|_| NetworkError::NotRunning)
    }

    /// Ping a peer.
    pub async fn ping(&self, peer: &str) -> NetworkResult<PingResult> {
        let peer_id = parse_peer_id(peer)?;
        let (tx, rx) = oneshot::channel();
        self.cmd_tx
            .send(HostCommand::Ping(peer_id, tx))
            .await
            .map_err(|_| NetworkError::NotRunning)?;
        // Wait with timeout
        match tokio::time::timeout(Duration::from_secs(10), rx).await {
            Ok(Ok(result)) => Ok(result),
            Ok(Err(_)) => Err(NetworkError::NotRunning),
            Err(_) => Ok(PingResult {
                success: false,
                time: 0,
                text: "Ping timeout".into(),
            }),
        }
    }

    /// Get listen addresses.
    pub async fn listen_addresses(&self) -> NetworkResult<Vec<String>> {
        let (tx, rx) = oneshot::channel();
        self.cmd_tx
            .send(HostCommand::GetListenAddresses(tx))
            .await
            .map_err(|_| NetworkError::NotRunning)?;
        rx.await.map_err(|_| NetworkError::NotRunning)
    }

    /// Shutdown the host.
    pub async fn shutdown(&self) -> NetworkResult<()> {
        let (tx, rx) = oneshot::channel();
        let _ = self.cmd_tx.send(HostCommand::Shutdown(tx)).await;
        let _ = rx.await;
        Ok(())
    }

    /// Find providers for content.
    pub async fn find_providers(&self, key: &[u8]) -> NetworkResult<crate::DhtQueryResult> {
        let (tx, rx) = oneshot::channel();
        self.cmd_tx
            .send(HostCommand::FindProviders(key.to_vec(), tx))
            .await
            .map_err(|_| NetworkError::NotRunning)?;
        match tokio::time::timeout(Duration::from_secs(30), rx).await {
            Ok(Ok(result)) => Ok(result),
            Ok(Err(_)) => Err(NetworkError::NotRunning),
            Err(_) => Ok(crate::DhtQueryResult::failure(
                "providers",
                "query timeout".into(),
            )),
        }
    }

    /// Find a peer in the DHT.
    pub async fn find_peer(&self, peer: &str) -> NetworkResult<crate::DhtQueryResult> {
        let peer_id = parse_peer_id(peer)?;
        let (tx, rx) = oneshot::channel();
        self.cmd_tx
            .send(HostCommand::FindPeer(peer_id, tx))
            .await
            .map_err(|_| NetworkError::NotRunning)?;
        match tokio::time::timeout(Duration::from_secs(30), rx).await {
            Ok(Ok(result)) => Ok(result),
            Ok(Err(_)) => Err(NetworkError::NotRunning),
            Err(_) => Ok(crate::DhtQueryResult::failure(
                "find_peer",
                "query timeout".into(),
            )),
        }
    }

    /// Announce content to the DHT.
    pub async fn provide(&self, key: &[u8]) -> NetworkResult<crate::DhtQueryResult> {
        let (tx, rx) = oneshot::channel();
        self.cmd_tx
            .send(HostCommand::Provide(key.to_vec(), tx))
            .await
            .map_err(|_| NetworkError::NotRunning)?;
        match tokio::time::timeout(Duration::from_secs(30), rx).await {
            Ok(Ok(result)) => Ok(result),
            Ok(Err(_)) => Err(NetworkError::NotRunning),
            Err(_) => Ok(crate::DhtQueryResult::failure(
                "provide",
                "query timeout".into(),
            )),
        }
    }

    /// Put a value in the DHT.
    pub async fn put_value(
        &self,
        key: &[u8],
        value: Vec<u8>,
    ) -> NetworkResult<crate::DhtQueryResult> {
        let (tx, rx) = oneshot::channel();
        self.cmd_tx
            .send(HostCommand::PutValue(key.to_vec(), value, tx))
            .await
            .map_err(|_| NetworkError::NotRunning)?;
        match tokio::time::timeout(Duration::from_secs(30), rx).await {
            Ok(Ok(result)) => Ok(result),
            Ok(Err(_)) => Err(NetworkError::NotRunning),
            Err(_) => Ok(crate::DhtQueryResult::failure(
                "put_value",
                "query timeout".into(),
            )),
        }
    }

    /// Get a value from the DHT.
    pub async fn get_value(&self, key: &[u8]) -> NetworkResult<crate::DhtQueryResult> {
        let (tx, rx) = oneshot::channel();
        self.cmd_tx
            .send(HostCommand::GetValue(key.to_vec(), tx))
            .await
            .map_err(|_| NetworkError::NotRunning)?;
        match tokio::time::timeout(Duration::from_secs(30), rx).await {
            Ok(Ok(result)) => Ok(result),
            Ok(Err(_)) => Err(NetworkError::NotRunning),
            Err(_) => Ok(crate::DhtQueryResult::failure(
                "get_value",
                "query timeout".into(),
            )),
        }
    }

    /// Get DHT statistics.
    pub async fn dht_stats(&self) -> NetworkResult<crate::DhtStats> {
        let (tx, rx) = oneshot::channel();
        self.cmd_tx
            .send(HostCommand::GetDhtStats(tx))
            .await
            .map_err(|_| NetworkError::NotRunning)?;
        rx.await.map_err(|_| NetworkError::NotRunning)
    }

    /// Bootstrap the DHT.
    pub async fn bootstrap(&self) -> NetworkResult<()> {
        let (tx, rx) = oneshot::channel();
        self.cmd_tx
            .send(HostCommand::Bootstrap(tx))
            .await
            .map_err(|_| NetworkError::NotRunning)?;
        rx.await
            .map_err(|_| NetworkError::NotRunning)?
            .map_err(NetworkError::Swarm)
    }
}

use libp2p::{identify, kad, mdns, ping};