guardian-db 0.19.0

/// Optimized Iroh backend - native embedded Iroh Endpoint in Rust.
///
/// Uses the embedded Iroh Endpoint with advanced optimizations:
/// - Intelligent cache with automatic compression
/// - Connection pool with load balancing
/// - Batch processing for optimized throughput
/// - Real-time performance monitoring
use crate::guardian::error::{GuardianError, Result};
use crate::p2p::network::{config::ClientConfig, types::*};
use bytes::Bytes;
use iroh::SecretKey;
use iroh::endpoint::Endpoint;
use iroh::protocol::Router;
use iroh::{EndpointAddr as NodeAddr, EndpointId as NodeId};
use iroh_blobs::api::Tag;
use iroh_blobs::store::fs::FsStore;
use iroh_blobs::{BlobFormat, BlobsProtocol, Hash as IrohHash, HashAndFormat};
use iroh_docs::protocol::Docs;
use iroh_gossip::net::Gossip;
use iroh_mdns_address_lookup::MdnsAddressLookup;
use std::collections::HashMap;
use std::path::PathBuf;
use std::pin::Pin;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::io::{AsyncRead, AsyncReadExt};
use tokio::sync::{Mutex, RwLock};
use tracing::{debug, info, warn};
// Main modules.
pub mod blobs;
pub mod docs;
pub mod gossip;
pub mod key_synchronizer;
pub mod networking_metrics;
pub mod ticket_exchange;

// Optimization modules.
pub mod batch_processor;
pub mod connection_pool;
pub mod optimized_cache;

pub use blobs::BlobStore;
pub use docs::WillowDocs;
pub use gossip::EpidemicPubSub;
pub use optimized_cache::OptimizedCache;

/// Information about a pinned object.
#[derive(Debug, Clone)]
pub struct PinInfo {
    /// BLAKE3 hash of the content (hex string).
    pub hash: String,
    pub pin_type: PinType,
}

/// Pin type.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum PinType {
    /// Direct pin of the object.
    Direct,
    /// Recursive pin (includes references).
    Recursive,
    /// Indirect pin (referenced by another pin).
    Indirect,
}

/// Statistics of a block.
#[derive(Debug, Clone)]
pub struct BlockStats {
    /// BLAKE3 hash of the block.
    pub hash: IrohHash,
    pub size: u64,
    pub exists_locally: bool,
}

/// Garbage collection statistics.
#[derive(Debug, Clone)]
pub struct GcStats {
    pub blocks_removed: u64,
    pub bytes_freed: u64,
    pub duration_ms: u64,
}

/// Backend performance metrics.
#[derive(Debug, Clone)]
pub struct BackendMetrics {
    /// Operations per second.
    pub ops_per_second: f64,
    /// Average latency in ms.
    pub avg_latency_ms: f64,
    /// Total number of operations.
    pub total_operations: u64,
    /// Number of errors.
    pub error_count: u64,
    /// Memory usage in bytes.
    pub memory_usage_bytes: u64,
}

/// Backend health status.
#[derive(Debug, Clone)]
pub struct HealthStatus {
    /// Whether the backend is healthy.
    pub healthy: bool,
    /// Descriptive message.
    pub message: String,
    /// Response time in ms.
    pub response_time_ms: u64,
    /// Verified components.
    pub checks: Vec<HealthCheck>,
}

/// Individual health check.
#[derive(Debug, Clone)]
pub struct HealthCheck {
    pub name: String,
    pub passed: bool,
    pub message: String,
}

/// iroh-blobs store (only FsStore is currently used).
enum StoreType {
    Fs(FsStore),
}

/// Optimized Iroh backend.
///
/// Snapshot of one home-relay's connection status, as reported by
/// [`IrohBackend::relay_status`] (C2). Derived from `Endpoint::home_relay_status`.
#[derive(Debug, Clone)]
pub struct RelayInfo {
    /// URL of the home relay.
    pub url: String,
    /// Whether the endpoint is currently connected to this relay.
    pub connected: bool,
    /// Most recent connection error, if currently disconnected.
    pub last_error: Option<String>,
}

/// High-performance Iroh backend with native optimizations:
/// - Multi-level cache with intelligent compression
/// - Connection pool with circuit breaking
/// - Batch processing for maximum throughput
/// - Continuous performance monitoring
pub struct IrohBackend {
    /// Backend configuration.
    #[allow(dead_code)]
    config: ClientConfig,
    /// Node data directory.
    data_dir: PathBuf,
    /// Iroh Endpoint for P2P communication.
    endpoint: Arc<RwLock<Option<Endpoint>>>,
    /// iroh-bytes store for storage.
    store: Arc<RwLock<Option<StoreType>>>,
    /// Gossip protocol instance for pub/sub.
    gossip: Arc<RwLock<Option<Gossip>>>,
    /// Docs protocol instance for the distributed KV store.
    docs: Arc<RwLock<Option<Docs>>>,
    /// Router for protocol multiplexing via ALPN.
    router: Arc<RwLock<Option<Router>>>,
    /// Node secret key.
    secret_key: SecretKey,
    /// Performance metrics.
    metrics: Arc<RwLock<BackendMetrics>>,
    /// Cache of pinned objects.
    pinned_cache: Arc<Mutex<HashMap<String, PinType>>>,
    /// Node status.
    node_status: Arc<RwLock<NodeStatus>>,
    /// Cache of peers discovered via Iroh Discovery Services (Pkarr/DNS/mDNS).
    discovery_cache: Arc<RwLock<DiscoveryCache>>,
    /// Optimized cache with integrated metrics, compression and intelligent eviction.
    optimized_cache: Arc<OptimizedCache>,
    /// Pool of active connections.
    connection_pool: Arc<RwLock<HashMap<NodeId, ConnectionInfo>>>,
    /// Real-time performance monitor.
    performance_monitor: Arc<RwLock<PerformanceMonitor>>,
    /// Advanced networking metrics collector.
    networking_metrics:
        Arc<crate::p2p::network::core::networking_metrics::NetworkingMetricsCollector>,
    /// Key synchronizer for consistency between peers.
    key_synchronizer: Arc<crate::p2p::network::core::key_synchronizer::KeySynchronizer>,
    /// Registry of `DocTicket` providers per store address (secure automatic exchange).
    ticket_registry: crate::p2p::network::core::ticket_exchange::TicketRegistry,
    /// Peers we have already connected to (candidates for requesting tickets).
    known_peers: Arc<RwLock<std::collections::HashSet<NodeId>>>,
    /// Per-peer latency sample history (bounded ring), for per-peer p95/p99 (C1).
    /// Fed by `update_connection_latency`; independent of the EMA `avg_latency_ms`.
    peer_latency_history: Arc<RwLock<HashMap<NodeId, std::collections::VecDeque<f64>>>>,
    /// Guardian Compute executor handler (registered on the Router under the
    /// compute ALPN); kept here so the owner can adjust the admission policy.
    #[cfg(feature = "compute")]
    compute_handler: Arc<RwLock<Option<crate::compute::ComputeProtocolHandler>>>,
    /// Guardian Compute capability telemetry service: gossips this node's
    /// capacity and collects peers' vectors into the scheduler directory.
    #[cfg(feature = "compute")]
    compute_gossip: Arc<RwLock<Option<Arc<crate::compute::CapabilityGossip>>>>,
    /// Guardian Compute reputation book, persisted across the ephemeral
    /// schedulers returned by `compute_scheduler()` so redundant-run penalties
    /// (RFC §6.5) accumulate instead of resetting each call.
    #[cfg(feature = "compute")]
    compute_reputation: Arc<crate::compute::ReputationBook>,
}

/// Max latency samples kept per peer for percentile estimation (C1).
const PEER_LATENCY_HISTORY_CAP: usize = 128;

/// Internal status of the Iroh node.
#[derive(Debug, Clone)]
struct NodeStatus {
    /// Whether the node is online and operational.
    is_online: bool,
    /// Last error encountered.
    last_error: Option<String>,
    /// Timestamp of the last activity.
    last_activity: Instant,
    /// Number of connected peers.
    connected_peers: u32,
}

/// Information about a peer discovered via Iroh Discovery Services.
///
/// This structure stores information about peers discovered via Pkarr, DNS or mDNS.
#[derive(Debug, Clone)]
struct DiscoveredPeerInfo {
    /// Node ID.
    node_id: NodeId,
    /// Known addresses (SocketAddr formatted as strings).
    addresses: Vec<String>,
    /// Last time it was seen.
    last_seen: Instant,
    /// Approximate latency.
    #[allow(dead_code)]
    latency: Option<Duration>,
    /// Supported protocols (informational identifiers).
    protocols: Vec<String>,
}

/// Discovery information cache for peers.
///
/// This cache stores discovery information (Pkarr/DNS/mDNS) obtained via Discovery Services.
#[derive(Debug, Default)]
struct DiscoveryCache {
    /// Known peers indexed by NodeId.
    peers: HashMap<NodeId, DiscoveredPeerInfo>,
}

/// Cached data with metadata.
#[derive(Debug, Clone)]
pub struct CachedData {
    /// Blob data.
    pub data: Bytes,
    /// Cache timestamp.
    pub cached_at: Instant,
    /// Number of accesses.
    pub access_count: u64,
    /// Data size.
    pub size: usize,
}

/// Optimized connection information.
#[derive(Debug, Clone)]
pub struct ConnectionInfo {
    /// Node ID.
    pub node_id: NodeId,
    /// Connection address.
    pub address: String,
    /// Connection timestamp.
    pub connected_at: Instant,
    /// Last use.
    pub last_used: Instant,
    /// Average latency (ms).
    pub avg_latency_ms: f64,
    /// Number of operations.
    pub operations_count: u64,
}

/// Real-time performance monitor.
#[derive(Debug, Default)]
pub struct PerformanceMonitor {
    /// Throughput metrics.
    pub throughput_metrics: ThroughputMetrics,
    /// Latency metrics.
    pub latency_metrics: LatencyMetrics,
    /// Resource metrics.
    pub resource_metrics: ResourceMetrics,
    /// Performance history.
    pub performance_history: Vec<PerformanceSnapshot>,
}

/// Throughput metrics.
#[derive(Debug, Default, Clone)]
pub struct ThroughputMetrics {
    /// Operations per second.
    pub ops_per_second: f64,
    /// Bytes per second.
    pub bytes_per_second: u64,
    /// Peak throughput.
    pub peak_throughput: f64,
    /// Average throughput.
    pub avg_throughput: f64,
}

/// Latency metrics.
#[derive(Debug, Default, Clone)]
pub struct LatencyMetrics {
    /// Average latency (ms).
    pub avg_latency_ms: f64,
    /// P95 latency (ms).
    pub p95_latency_ms: f64,
    /// P99 latency (ms).
    pub p99_latency_ms: f64,
    /// Minimum latency (ms).
    pub min_latency_ms: f64,
    /// Maximum latency (ms).
    pub max_latency_ms: f64,
}

/// Resource metrics.
#[derive(Debug, Default, Clone)]
pub struct ResourceMetrics {
    /// CPU usage (0.0-1.0).
    pub cpu_usage: f64,
    /// Memory usage (bytes).
    pub memory_usage_bytes: u64,
    /// Disk I/O (bytes/s).
    pub disk_io_bps: u64,
    /// Bandwidth (bytes/s).
    pub network_bandwidth_bps: u64,
}

/// Performance snapshot at a specific moment.
#[derive(Debug, Clone)]
pub struct PerformanceSnapshot {
    /// Snapshot timestamp.
    pub timestamp: Instant,
    /// Throughput metrics.
    pub throughput: ThroughputMetrics,
    /// Latency metrics.
    pub latency: LatencyMetrics,
    /// Resource metrics.
    pub resources: ResourceMetrics,
}

/// Cached content with metadata.
#[derive(Debug, Clone)]
#[allow(dead_code)]
struct CachedContent {
    /// Content data.
    data: bytes::Bytes,
    /// Timestamp of when it was cached.
    cached_at: Instant,
    /// Number of cache accesses.
    access_count: u64,
    /// Last access.
    last_accessed: Instant,
    /// Size in bytes.
    size: usize,
    /// Cache priority (0-10).
    priority: u8,
}

/// Content metadata (reserved for future use).
#[derive(Debug, Clone)]
#[allow(dead_code)]
struct ContentMetadata {
    #[allow(dead_code)]
    hash_str: String,
    /// Content size.
    #[allow(dead_code)]
    size: usize,
    /// Content type.
    #[allow(dead_code)]
    content_type: Option<String>,
    /// Content hash.
    #[allow(dead_code)]
    hash: String,
    /// Peers that hold the content.
    #[allow(dead_code)]
    providers: Vec<NodeId>,
    /// Discovery timestamp.
    #[allow(dead_code)]
    discovered_at: Instant,
}

/// Simple structure for cache statistics (public API).
#[derive(Debug, Clone, Default)]
pub struct SimpleCacheStats {
    pub entries_count: u32,
    pub hit_ratio: f64,
    pub total_size_bytes: u64,
}

impl IrohBackend {
    // ╔════════════════════════════════════════════════════════════════════════════════╗
    // ║                          INITIALIZATION AND CONSTRUCTION                          ║
    // ╚════════════════════════════════════════════════════════════════════════════════╝

    /// Creates a new instance of the Iroh backend.
    ///
    /// # Arguments
    /// * `config` - Client configuration containing the data path
    ///
    /// # Returns
    /// A new configured instance of the Iroh backend
    ///
    /// # Errors
    /// Returns an error if the Iroh node cannot be initialized
    pub async fn new(config: &ClientConfig) -> Result<Self> {
        let data_dir = config
            .data_store_path
            .as_ref()
            .ok_or_else(|| {
                GuardianError::Other(
                    "Data directory not configured for the Iroh backend".to_string(),
                )
            })?
            .clone();

        debug!("Initializing Iroh backend in directory: {:?}", data_dir);

        // Ensure the directory exists.
        tokio::fs::create_dir_all(&data_dir)
            .await
            .map_err(|e| GuardianError::Other(format!("Error creating data directory: {}", e)))?;

        // Generate or load the node's persistent secret key.
        let secret_key = Self::load_or_generate_node_secret_key(&data_dir).await?;

        let data_dir_clone = data_dir.clone();

        // Initialize the optimized components.
        debug!("Initializing optimization components...");

        // Optimized cache with compression, integrated metrics and intelligent eviction.
        let cache_config = optimized_cache::CacheConfig {
            max_data_cache_size: 256 * 1024 * 1024, // 256 MB
            max_data_entries: 10_000,
            max_compressed_cache_size: 512 * 1024 * 1024, // 512 MB
            max_compressed_entries: 50_000,
            default_ttl_secs: 3600,
            compression_threshold: 64 * 1024, // 64 KB
            compression_level: 6,
            eviction_threshold: 0.85,
            enable_access_prediction: true,
        };
        let optimized_cache = Arc::new(OptimizedCache::new(cache_config));

        // Initially empty connection pool.
        let connection_pool = Arc::new(RwLock::new(HashMap::new()));

        let backend = Self {
            config: config.clone(),
            data_dir,
            endpoint: Arc::new(RwLock::new(None)),
            store: Arc::new(RwLock::new(None)),
            gossip: Arc::new(RwLock::new(None)),
            docs: Arc::new(RwLock::new(None)),
            router: Arc::new(RwLock::new(None)),
            secret_key,
            metrics: Arc::new(RwLock::new(BackendMetrics {
                ops_per_second: 0.0,
                avg_latency_ms: 0.0,
                total_operations: 0,
                error_count: 0,
                memory_usage_bytes: 0,
            })),
            pinned_cache: Arc::new(Mutex::new(HashMap::new())),
            node_status: Arc::new(RwLock::new(NodeStatus {
                is_online: false, // Starts offline until it connects.
                last_error: None,
                last_activity: Instant::now(),
                connected_peers: 0,
            })),
            discovery_cache: Arc::new(RwLock::new(DiscoveryCache::default())),

            // Optimized components.
            optimized_cache,
            connection_pool,
            performance_monitor: Arc::new(RwLock::new(PerformanceMonitor::default())),

            networking_metrics: Arc::new(
                crate::p2p::network::core::networking_metrics::NetworkingMetricsCollector::new(),
            ),
            key_synchronizer: Arc::new(
                crate::p2p::network::core::key_synchronizer::KeySynchronizer::new(config).await?,
            ),
            ticket_registry: crate::p2p::network::core::ticket_exchange::new_registry(),
            known_peers: Arc::new(RwLock::new(std::collections::HashSet::new())),
            peer_latency_history: Arc::new(RwLock::new(HashMap::new())),
            #[cfg(feature = "compute")]
            compute_handler: Arc::new(RwLock::new(None)),
            #[cfg(feature = "compute")]
            compute_gossip: Arc::new(RwLock::new(None)),
            #[cfg(feature = "compute")]
            compute_reputation: Arc::new(crate::compute::ReputationBook::new()),
        };
        // Initialize the Iroh node asynchronously.
        backend.initialize_node().await?;
        info!(
            "Optimized Iroh backend initialized successfully at {:?}",
            data_dir_clone
        );
        info!("Active optimizations: intelligent cache, connection pooling, batch processing");
        Ok(backend)
    }

    /// Loads an existing secret key or securely generates a new one.
    ///
    /// - Looks for an existing key file in the data directory
    /// - Generates a new cryptographically secure key if needed
    /// - Saves the generated key for future reuse
    async fn load_or_generate_node_secret_key(data_dir: &std::path::Path) -> Result<SecretKey> {
        let key_file = data_dir.join("node_secret.key");

        // Try to load an existing key.
        if key_file.exists() {
            debug!("Loading existing secret key from {:?}", key_file);

            match tokio::fs::read(&key_file).await {
                Ok(key_bytes) if key_bytes.len() == 32 => {
                    let mut key_array = [0u8; 32];
                    key_array.copy_from_slice(&key_bytes);

                    let secret_key = SecretKey::from_bytes(&key_array);
                    info!("Node secret key loaded successfully");
                    return Ok(secret_key);
                }
                Ok(_) => {
                    warn!("Key file has an invalid size, generating a new one");
                }
                Err(e) => {
                    warn!("Error reading key file: {}, generating a new one", e);
                }
            }
        }

        // Generate a new cryptographic key.
        debug!("Generating a new secret key for the node");
        let secret_key = SecretKey::generate();

        // Save the key for future use.
        if let Err(e) = tokio::fs::write(&key_file, secret_key.to_bytes()).await {
            warn!("Error saving secret key: {} - Using a temporary key", e);
        } else {
            info!("New secret key saved to {:?}", key_file);
        }

        Ok(secret_key)
    }

    /// Initializes the embedded Iroh node.
    async fn initialize_node(&self) -> Result<()> {
        debug!("Initializing Iroh node with FsStore for persistence...");

        // Create a specific directory for the store.
        let store_dir = self.data_dir.join("iroh_store");
        tokio::fs::create_dir_all(&store_dir)
            .await
            .map_err(|e| GuardianError::Other(format!("Error creating store directory: {}", e)))?;

        // Initialize the FsStore with persistence.
        let fs_store = FsStore::load(&store_dir)
            .await
            .map_err(|e| GuardianError::Other(format!("Error initializing FsStore: {}", e)))?;

        // Store the store.
        {
            let mut store_lock = self.store.write().await;
            *store_lock = Some(StoreType::Fs(fs_store));
        }

        // Initialize the Endpoint for P2P communication with native address lookup services.
        // Iroh 1.0 uses the N0 preset, which enables DNS + Pkarr discovery via n0.computer (global).
        // Local mDNS discovery (LAN) is added after binding via iroh-mdns-address-lookup.
        let endpoint = Endpoint::builder(iroh::endpoint::presets::N0)
            .secret_key(self.secret_key.clone())
            .bind()
            .await
            .map_err(|e| GuardianError::Other(format!("Error initializing Endpoint: {}", e)))?;

        // mDNS discovery on the local network (LAN), equivalent to the former discovery_local_network().
        match MdnsAddressLookup::builder().build(endpoint.id()) {
            Ok(mdns) => match endpoint.address_lookup() {
                Ok(services) => {
                    services.add(mdns);
                    debug!("Local mDNS discovery (LAN) enabled");
                }
                Err(e) => warn!("Address lookup unavailable for mDNS: {}", e),
            },
            Err(e) => warn!("Could not start local mDNS discovery: {}", e),
        }

        // Store the endpoint.
        {
            let mut endpoint_lock = self.endpoint.write().await;
            *endpoint_lock = Some(endpoint.clone());
        }

        // Initialize Gossip with the shared Endpoint.
        debug!("Initializing the Gossip protocol...");
        let gossip = Gossip::builder()
            .max_message_size(self.config.gossip.max_message_size)
            .spawn(endpoint.clone());
        {
            let mut gossip_lock = self.gossip.write().await;
            *gossip_lock = Some(gossip.clone());
        }
        info!("Gossip protocol initialized successfully");

        // Initialize the Router for ALPN protocol multiplexing.
        debug!("Configuring the Router for ALPN multiplexing...");

        // Initialize BlobsProtocol with the shared store and endpoint.
        debug!("Initializing BlobsProtocol...");
        let store_lock = self.store.read().await;
        let store_for_blobs = store_lock
            .as_ref()
            .ok_or_else(|| GuardianError::Other("Store not initialized".to_string()))?;

        let blobs = match store_for_blobs {
            StoreType::Fs(fs_store) => BlobsProtocol::new(fs_store.as_ref(), None),
        };
        drop(store_lock);

        // Initialize the Docs protocol.
        debug!("Initializing the Docs protocol...");
        let docs_dir = self.data_dir.join("iroh_docs");
        tokio::fs::create_dir_all(&docs_dir)
            .await
            .map_err(|e| GuardianError::Other(format!("Error creating docs directory: {}", e)))?;

        // Get the store for Docs (FsStore implements AsRef<Store>).
        let store_lock = self.store.read().await;
        let blobs_store = match store_lock.as_ref() {
            Some(StoreType::Fs(fs_store)) => fs_store.as_ref().clone(),
            None => return Err(GuardianError::Other("Store not initialized".into())),
        };
        drop(store_lock);

        // Create Docs using the Builder pattern.
        let docs = Docs::persistent(docs_dir)
            .spawn(endpoint.clone(), blobs_store, gossip.clone())
            .await
            .map_err(|e| GuardianError::Other(format!("Error initializing Docs: {}", e)))?;

        // Store Docs.
        {
            let mut docs_lock = self.docs.write().await;
            *docs_lock = Some(docs.clone());
        }
        info!("Docs protocol initialized successfully");

        // Configure the Router with Gossip, Blobs, Docs and the ticket exchange protocol.
        let ticket_handler = crate::p2p::network::core::ticket_exchange::TicketProtocolHandler::new(
            self.ticket_registry.clone(),
        );
        let router_builder = Router::builder(endpoint.clone())
            .accept(iroh_gossip::ALPN, gossip.clone())
            .accept(iroh_blobs::ALPN, blobs)
            .accept(iroh_docs::ALPN, docs)
            .accept(
                crate::p2p::network::core::ticket_exchange::TICKET_ALPN,
                ticket_handler,
            );

        // Guardian Compute executor: serves delegated WASM tasks over the
        // compute ALPN, fetching task blobs from the requester via iroh-blobs.
        // Reciprocity-by-default (RFC 0002 §8.3) with the owner's policy in
        // control — adjustable at runtime through `compute_handler()`.
        #[cfg(feature = "compute")]
        let router_builder = {
            let store = self.get_store_for_blobs().await?;
            let blob_store = crate::p2p::network::core::blobs::BlobStore::new_with_endpoint(
                store,
                endpoint.clone(),
            );
            let handler = crate::compute::ComputeProtocolHandler::new(
                Arc::new(blob_store),
                crate::compute::ExecutorPolicy::default(),
            )
            .map_err(|e| {
                GuardianError::Other(format!("Error initializing Guardian Compute: {e}"))
            })?;
            *self.compute_handler.write().await = Some(handler.clone());
            info!("Guardian Compute executor registered on the compute ALPN");
            router_builder.accept(crate::compute::COMPUTE_ALPN, handler)
        };

        let router = router_builder.spawn();

        {
            let mut router_lock = self.router.write().await;
            *router_lock = Some(router);
        }
        info!("Router configured with ALPN multiplexing: Gossip + Blobs + Docs active");

        // Guardian Compute capability telemetry (Phase 3): sample this node's
        // capacity and gossip it with hysteresis; collect peers' vectors into
        // the scheduler directory. Peers join the mesh as connections are made
        // (see `compute_join_capability_mesh`).
        //
        // Failure here is fatal, matching the handler init above: a node that
        // serves inbound tasks (handler registered) but cannot announce its
        // capacity or schedule (no telemetry) is a silent free-rider. The
        // compute subsystem is all-or-nothing — surface the error rather than
        // leaving it half-initialized.
        #[cfg(feature = "compute")]
        if let Some(handler) = self.compute_handler.read().await.clone() {
            let directory = Arc::new(crate::compute::CapabilityDirectory::new());
            let capability_gossip = crate::compute::CapabilityGossip::spawn(
                gossip.clone(),
                endpoint.id(),
                handler,
                directory,
                Vec::new(),
                crate::compute::TelemetryConfig::default(),
            )
            .await
            .map_err(|e| {
                GuardianError::Other(format!(
                    "Error starting Guardian Compute capability telemetry: {e}"
                ))
            })?;
            *self.compute_gossip.write().await = Some(Arc::new(capability_gossip));
            info!("Guardian Compute capability telemetry active");
        }

        // Update the status to online.
        {
            let mut status = self.node_status.write().await;
            status.is_online = true;
            status.last_activity = Instant::now();
            status.last_error = None;
        }

        // Discovery is managed automatically by the Endpoint via discovery_n0() and discovery_local_network().
        // Iroh publishes and discovers peers automatically via PkarrPublisher, DnsDiscovery and MdnsDiscovery.
        debug!("Iroh's native discovery services enabled on the Endpoint");
        info!("Iroh backend initialized with active discovery services");
        Ok(())
    }

    /// Shuts down the backend, ensuring all pending operations
    /// are finished and the data is persisted to disk.
    ///
    /// This method is especially important to ensure FsStore tags are
    /// synchronized to the SQLite database (blobs.db) before shutdown.
    pub async fn shutdown(&self) -> Result<()> {
        debug!("Starting IrohBackend shutdown");

        // 1. Stop accepting new connections on the endpoint.
        if let Ok(endpoint_arc) = self.get_endpoint().await {
            let endpoint_lock = endpoint_arc.read().await;
            if let Some(endpoint) = endpoint_lock.as_ref() {
                // Wait a bit for pending operations.
                tokio::time::sleep(Duration::from_millis(100)).await;

                // Close all active connections.
                endpoint.close().await;
                debug!("Endpoint closed");
            }
        }

        // 2. Force a flush of pending tags by performing a read.
        // This helps ensure the SQLite WAL is synchronized.
        if (self.pin_ls().await).is_ok() {
            debug!("Tags listed to force a sync");
        }

        // 3. Wait a bit to ensure asynchronous operations finish.
        tokio::time::sleep(Duration::from_millis(500)).await;

        // 4. Clear the optimized cache.
        let _ = self.optimized_cache.clear().await;
        debug!("Optimized cache cleared");

        // 5. Update the node status.
        {
            let mut status = self.node_status.write().await;
            status.is_online = false;
            status.last_activity = Instant::now();
        }

        info!("IrohBackend shutdown complete");
        Ok(())
    }

    /// Returns a reference to the store if available.
    async fn get_store(&self) -> Result<Arc<RwLock<Option<StoreType>>>> {
        let store_lock = self.store.read().await;
        if store_lock.is_none() {
            drop(store_lock);
            return Err(GuardianError::Other("Store not initialized".to_string()));
        }
        Ok(self.store.clone())
    }

    /// Returns the specific store for BlobStore.
    ///
    /// Returns Arc<RwLock<FsStore>> for direct use by BlobStore.
    /// Ensures the store is initialized and unwraps the StoreType::Fs.
    pub async fn get_store_for_blobs(&self) -> Result<Arc<RwLock<FsStore>>> {
        let store_lock = self.store.read().await;
        match store_lock.as_ref() {
            Some(StoreType::Fs(fs_store)) => Ok(Arc::new(RwLock::new(fs_store.clone()))),
            None => {
                drop(store_lock);
                Err(GuardianError::Other("Store not initialized".to_string()))
            }
        }
    }

    /// Returns the Guardian Compute executor handler, if initialized — the
    /// hook for the node owner to inspect and adjust the admission policy
    /// (`set_policy`), which always overrides the reciprocity default.
    #[cfg(feature = "compute")]
    pub async fn compute_handler(&self) -> Option<crate::compute::ComputeProtocolHandler> {
        self.compute_handler.read().await.clone()
    }

    /// Returns a requester-side client for delegating tasks to other nodes
    /// over the compute ALPN (Phase 2: the caller picks the executor).
    #[cfg(feature = "compute")]
    pub async fn compute_client(&self) -> Result<crate::compute::ComputeClient> {
        let endpoint_arc = self.get_endpoint().await?;
        let endpoint = endpoint_arc
            .read()
            .await
            .as_ref()
            .cloned()
            .ok_or_else(|| GuardianError::Other("Endpoint not initialized".to_string()))?;
        Ok(crate::compute::ComputeClient::new(endpoint))
    }

    /// Returns the capability telemetry service, if running.
    #[cfg(feature = "compute")]
    pub async fn compute_capability_gossip(&self) -> Option<Arc<crate::compute::CapabilityGossip>> {
        self.compute_gossip.read().await.clone()
    }

    /// Adds peers to the capability gossip mesh — call after connecting to
    /// peers so their capacity vectors start flowing both ways.
    #[cfg(feature = "compute")]
    pub async fn compute_join_capability_mesh(&self, peers: Vec<NodeId>) -> Result<()> {
        let gossip = self.compute_gossip.read().await.clone().ok_or_else(|| {
            GuardianError::Other("Guardian Compute capability gossip not running".to_string())
        })?;
        gossip.join_peers(peers).await.map_err(GuardianError::Other)
    }

    /// Returns the capability-aware scheduler (Phase 3): `execute` with no
    /// explicit destination — the network decides where the task runs.
    #[cfg(feature = "compute")]
    pub async fn compute_scheduler(&self) -> Result<crate::compute::ComputeScheduler> {
        let gossip = self.compute_gossip.read().await.clone().ok_or_else(|| {
            GuardianError::Other("Guardian Compute capability gossip not running".to_string())
        })?;
        let client = self.compute_client().await?;
        let endpoint_arc = self.get_endpoint().await?;
        let local = endpoint_arc
            .read()
            .await
            .as_ref()
            .map(|ep| ep.id())
            .ok_or_else(|| GuardianError::Other("Endpoint not initialized".to_string()))?;
        Ok(crate::compute::ComputeScheduler::with_reputation(
            client,
            gossip.directory(),
            self.compute_reputation.clone(),
            local,
            crate::compute::SchedulerConfig::default(),
        ))
    }

    /// Returns a reference to the endpoint if available.
    pub async fn get_endpoint(&self) -> Result<Arc<RwLock<Option<Endpoint>>>> {
        let endpoint_lock = self.endpoint.read().await;
        if endpoint_lock.is_none() {
            drop(endpoint_lock);
            return Err(GuardianError::Other("Endpoint not initialized".to_string()));
        }
        Ok(self.endpoint.clone())
    }

    /// Snapshot of the endpoint's home-relay connection status (C2).
    ///
    /// Returns one entry per home relay whose URL is known; empty when no relay is
    /// configured or before one is selected. Wraps `Endpoint::home_relay_status`.
    pub async fn relay_status(&self) -> Vec<RelayInfo> {
        use iroh::Watcher;
        let Ok(endpoint_arc) = self.get_endpoint().await else {
            return Vec::new();
        };
        let endpoint_lock = endpoint_arc.read().await;
        let Some(endpoint) = endpoint_lock.as_ref() else {
            return Vec::new();
        };
        let mut watcher = endpoint.home_relay_status();
        watcher
            .get()
            .into_iter()
            .map(|s| RelayInfo {
                url: s.url().to_string(),
                connected: s.is_connected(),
                last_error: s.last_error().map(|e| e.to_string()),
            })
            .collect()
    }

    /// Real connection type to `node_id` derived from its *active* transport
    /// address (C1): `"relay"`, `"direct"` (IP), or `"unknown"`. `None` when the
    /// peer is not known to the endpoint. Wraps `Endpoint::remote_info`.
    pub async fn conn_type(&self, node_id: NodeId) -> Option<String> {
        use iroh::endpoint::TransportAddrUsage;
        let endpoint_arc = self.get_endpoint().await.ok()?;
        let endpoint_lock = endpoint_arc.read().await;
        let endpoint = endpoint_lock.as_ref()?;
        let info = endpoint.remote_info(node_id).await?;
        // Prefer the actively-used address; fall back to any known address.
        let chosen = info
            .addrs()
            .find(|a| matches!(a.usage(), TransportAddrUsage::Active))
            .or_else(|| info.addrs().next())?;
        let kind = if chosen.addr().is_relay() {
            "relay"
        } else if chosen.addr().is_ip() {
            "direct"
        } else {
            "unknown"
        };
        Some(kind.to_string())
    }

    // ─── Automatic DocTicket exchange (secure replication of iroh-docs stores) ─────────────

    /// Registers a store as a `DocTicket` provider, indexed by its address.
    ///
    /// When an authorized peer requests this address's ticket via
    /// [`ticket_exchange::TICKET_ALPN`], the handler delivers the capability matching the
    /// peer's role: `write_ticket` (carries the namespace secret) for write-authorized peers,
    /// `read_ticket` (namespace public key only) for read-only peers.
    pub async fn register_ticket_provider(
        &self,
        address: String,
        read_ticket: String,
        write_ticket: String,
        access_controller: Arc<dyn crate::access_control::traits::AccessController>,
    ) {
        let provider = crate::p2p::network::core::ticket_exchange::TicketProvider {
            read_ticket,
            write_ticket,
            access_controller,
        };
        self.ticket_registry.write().await.insert(address, provider);
    }

    /// Registers a peer we have connected to (a candidate for requesting tickets).
    pub async fn note_known_peer(&self, peer: NodeId) {
        self.known_peers.write().await.insert(peer);
    }

    /// Requests the `DocTicket` for `address` from each known peer, returning the first granted one.
    ///
    /// Used by iroh-docs stores when opening without a ticket: it tries to join the shared
    /// namespace of a peer that already holds it (and authorizes this node), instead of creating
    /// an isolated namespace.
    pub async fn request_ticket_from_known_peers(&self, address: &str) -> Option<String> {
        let peers: Vec<NodeId> = {
            let kp = self.known_peers.read().await;
            kp.iter().copied().collect()
        };
        if peers.is_empty() {
            return None;
        }

        let endpoint_arc = self.get_endpoint().await.ok()?;
        let endpoint_lock = endpoint_arc.read().await;
        let endpoint = endpoint_lock.as_ref()?.clone();
        drop(endpoint_lock);

        for peer in peers {
            match crate::p2p::network::core::ticket_exchange::request_ticket(
                &endpoint, peer, address,
            )
            .await
            {
                Ok(Some(ticket)) => {
                    info!(peer = %peer.fmt_short(), address, "DocTicket obtained from peer via automatic exchange");
                    return Some(ticket);
                }
                Ok(None) => {
                    debug!(peer = %peer.fmt_short(), address, "Peer did not provide a ticket (denied/unavailable)");
                }
                Err(e) => {
                    debug!(peer = %peer.fmt_short(), address, error = %e, "Failed to request ticket from peer");
                }
            }
        }
        None
    }

    /// Resolves a store's shared namespace deterministically, avoiding split-brain
    /// when multiple nodes open the same store simultaneously.
    ///
    /// Rule: the node with the **smallest `EndpointId`** among {self, known peers} is the namespace
    /// "creator"; the others wait and import its ticket.
    ///
    /// - Tries to obtain the ticket immediately (common case: a peer already created and registered it).
    /// - If no peer provided one and a peer with a smaller id exists (which should be the creator),
    ///   it makes a few short retries to give it time to create/register.
    /// - If this node has the smallest id (or no one responded after the retries), returns `None`
    ///   and the caller creates a new namespace (taking the creator role).
    pub async fn resolve_shared_ticket(&self, store_key: &str) -> Option<String> {
        // Immediate attempt.
        if let Some(ticket) = self.request_ticket_from_known_peers(store_key).await {
            return Some(ticket);
        }

        // Is there any known peer with a smaller EndpointId than ours?
        let my_id = self.secret_key().public();
        let lower_peer_exists = {
            let kp = self.known_peers.read().await;
            kp.iter().any(|p| p.as_bytes() < my_id.as_bytes())
        };

        if !lower_peer_exists {
            // We are the node with the smallest id (or have no peers): we take the creator role.
            return None;
        }

        // There is a peer that should be the creator — give it time to create/register and try again.
        const MAX_RETRIES: u32 = 10;
        const RETRY_DELAY: Duration = Duration::from_millis(300);
        for attempt in 1..=MAX_RETRIES {
            tokio::time::sleep(RETRY_DELAY).await;
            if let Some(ticket) = self.request_ticket_from_known_peers(store_key).await {
                debug!(
                    store_key,
                    attempt, "DocTicket obtained from the creator after a retry"
                );
                return Some(ticket);
            }
        }

        // Fallback: the creator did not respond in time; we take the namespace to avoid blocking.
        warn!(
            store_key,
            "Expected creator did not provide the ticket in time; creating a local namespace (possible split-brain)"
        );
        None
    }

    /// Returns a reference to the Gossip if available.
    pub async fn get_gossip(&self) -> Result<Arc<RwLock<Option<Gossip>>>> {
        let gossip_lock = self.gossip.read().await;
        if gossip_lock.is_none() {
            drop(gossip_lock);
            return Err(GuardianError::Other("Gossip not initialized".to_string()));
        }
        Ok(self.gossip.clone())
    }

    /// Returns a reference to the Router if available.
    pub async fn get_router(&self) -> Result<Arc<RwLock<Option<Router>>>> {
        let router_lock = self.router.read().await;
        if router_lock.is_none() {
            drop(router_lock);
            return Err(GuardianError::Other("Router not initialized".to_string()));
        }
        Ok(self.router.clone())
    }

    /// Returns a reference to Docs if available.
    pub async fn get_docs(&self) -> Result<Arc<RwLock<Option<Docs>>>> {
        let docs_lock = self.docs.read().await;
        if docs_lock.is_none() {
            drop(docs_lock);
            return Err(GuardianError::Other("Docs not initialized".to_string()));
        }
        Ok(self.docs.clone())
    }

    /// Actively discovers peers using the Discovery trait's subscribe().
    ///
    /// Uses discovery services (Pkarr/DNS/mDNS) for active, real-time discovery.
    /// Polls the subscribe() stream to capture passive discovery events.
    pub async fn discover_peers_active(&self, _timeout: Duration) -> Result<Vec<NodeAddr>> {
        // API CHANGE (Iroh 1.0): passive discovery via Discovery::subscribe()
        // was replaced by a pull-based model in AddressLookupServices::resolve(endpoint_id),
        // which resolves a specific peer. There is no longer passive enumeration of all peers.
        // To resolve a specific peer, use discover_peer_integrated(node_id).
        debug!(
            "discover_peers_active: passive enumeration is not supported in Iroh 1.0; \
             use discover_peer_integrated(node_id) to resolve a specific peer"
        );
        Ok(Vec::new())
    }

    /// Discovers a specific peer using the Iroh Endpoint.
    ///
    /// First tries remote_info() (known peers), then active discovery.
    pub async fn discover_peer_integrated(&self, node_id: NodeId) -> Result<Vec<NodeAddr>> {
        debug!("Discovering peer {} via the Iroh Endpoint", node_id);

        let endpoint_arc = self.get_endpoint().await?;
        let endpoint_lock = endpoint_arc.read().await;
        let endpoint = endpoint_lock
            .as_ref()
            .ok_or_else(|| GuardianError::Other("Endpoint not initialized".to_string()))?;

        // First try remote_info() for already-known peers (now asynchronous in Iroh 1.0).
        if let Some(remote_info) = endpoint.remote_info(node_id).await {
            // Build EndpointAddr from the RemoteInfo (TransportAddr unifies IP + relay).
            let node_addr = NodeAddr::from_parts(
                remote_info.id(),
                remote_info.into_addrs().map(|a| a.into_addr()),
            );
            if !node_addr.addrs.is_empty() {
                info!("Peer {} found via remote_info()", node_id);
                return Ok(vec![node_addr]);
            }
        }

        debug!(
            "Peer {} is not in remote_info(), trying address lookup (resolve)",
            node_id
        );

        // Iroh 1.0 pull-based model: resolve(endpoint_id) via AddressLookupServices.
        let services = endpoint
            .address_lookup()
            .map_err(|e| GuardianError::Other(format!("Address lookup not configured: {}", e)))?;

        use futures::StreamExt;
        let mut stream = services.resolve(node_id);
        let mut discovered: Vec<NodeAddr> = Vec::new();
        let deadline = tokio::time::sleep(Duration::from_secs(5));
        tokio::pin!(deadline);
        loop {
            tokio::select! {
                _ = &mut deadline => break,
                item = stream.next() => match item {
                    Some(Ok(Ok(item))) => discovered.push(item.into_endpoint_addr()),
                    Some(_) => continue,
                    None => break,
                }
            }
        }

        if !discovered.is_empty() {
            info!("Peer {} discovered via address lookup", node_id);
            return Ok(discovered);
        }

        debug!("Peer {} not found after address lookup", node_id);
        Err(GuardianError::Other(format!(
            "Peer {} not found via remote_info() or address lookup",
            node_id
        )))
    }

    /// Gets content from the optimized cache if available.
    async fn get_from_cache(&self, hash_str: &str) -> Option<bytes::Bytes> {
        // OptimizedCache already updates metrics automatically (hits/misses).
        self.optimized_cache.get(hash_str).await
    }

    /// Adds content to the optimized cache.
    async fn add_to_cache(&self, hash_str: &str, data: bytes::Bytes) -> Result<()> {
        // OptimizedCache manages automatically:
        // - Compression (if data.len() >= compression_threshold)
        // - Metrics (hits, misses, bytes_cached)
        // - Intelligent eviction (when needed)
        self.optimized_cache.put(hash_str, data.clone()).await?;

        debug!(
            "Content added to the cache: {} ({} bytes)",
            hash_str,
            data.len()
        );
        Ok(())
    }

    /// Updates metrics after an operation.
    async fn update_metrics(&self, duration: Duration, success: bool) {
        // Update the basic metrics.
        {
            let mut metrics = self.metrics.write().await;
            metrics.total_operations += 1;
            if !success {
                metrics.error_count += 1;
            }

            // Update the average latency.
            let new_latency = duration.as_millis() as f64;
            if metrics.total_operations == 1 {
                metrics.avg_latency_ms = new_latency;
            } else {
                metrics.avg_latency_ms = (metrics.avg_latency_ms * 0.9) + (new_latency * 0.1);
            }

            // Compute ops/second.
            let ops_window = std::cmp::min(metrics.total_operations, 3600);
            metrics.ops_per_second = ops_window as f64 / 3600.0;
        } // Drop the metrics lock here.

        // Update the performance monitor with detailed metrics.
        {
            let mut monitor = self.performance_monitor.write().await;
            let latency_ms = duration.as_millis() as f64;

            // Update the latency metrics.
            if monitor.latency_metrics.min_latency_ms == 0.0
                || latency_ms < monitor.latency_metrics.min_latency_ms
            {
                monitor.latency_metrics.min_latency_ms = latency_ms;
            }
            if latency_ms > monitor.latency_metrics.max_latency_ms {
                monitor.latency_metrics.max_latency_ms = latency_ms;
            }

            // Update the average latency with a moving average.
            if monitor.latency_metrics.avg_latency_ms == 0.0 {
                monitor.latency_metrics.avg_latency_ms = latency_ms;
            } else {
                monitor.latency_metrics.avg_latency_ms =
                    (monitor.latency_metrics.avg_latency_ms * 0.95) + (latency_ms * 0.05);
            }

            // Update the throughput metrics.
            monitor.throughput_metrics.ops_per_second = (monitor.throughput_metrics.ops_per_second
                * 0.95)
                + (1.0 / duration.as_secs_f64() * 0.05);

            if monitor.throughput_metrics.ops_per_second
                > monitor.throughput_metrics.peak_throughput
            {
                monitor.throughput_metrics.peak_throughput =
                    monitor.throughput_metrics.ops_per_second;
            }
        }

        // Update the node status in a separate scope.
        {
            let mut status = self.node_status.write().await;
            status.last_activity = Instant::now();
            if success {
                status.last_error = None;
            }
        } // Drop the status lock here.
    }

    /// Runs an operation with metrics tracking.
    async fn execute_with_metrics<F, T>(&self, operation: F) -> Result<T>
    where
        F: std::future::Future<Output = Result<T>> + Send,
    {
        let start = Instant::now();
        let result = operation.await;
        let duration = start.elapsed();

        self.update_metrics(duration, result.is_ok()).await;

        // Update the error in the status if needed.
        if let Err(ref e) = result {
            let mut status = self.node_status.write().await;
            status.last_error = Some(e.to_string());
        }

        result
    }

    /// Converts an Iroh error into a GuardianError.
    fn map_iroh_error(error: impl std::fmt::Display) -> GuardianError {
        GuardianError::Other(format!("Iroh error: {}", error))
    }

    /// Converts a hexadecimal string into an Iroh BLAKE3 Hash.
    fn parse_hash(hash_str: &str) -> Result<IrohHash> {
        let hash_bytes = hex::decode(hash_str)
            .map_err(|e| GuardianError::Other(format!("Invalid hex hash '{}': {}", hash_str, e)))?;

        if hash_bytes.len() != 32 {
            return Err(GuardianError::Other(format!(
                "Hash must be 32 bytes, found: {}",
                hash_bytes.len()
            )));
        }

        let mut hash_array = [0u8; 32];
        hash_array.copy_from_slice(&hash_bytes);
        Ok(IrohHash::from(hash_array))
    }

    /// Converts an Iroh BLAKE3 Hash into a hexadecimal string.
    fn hash_to_string(hash: &IrohHash) -> String {
        hex::encode(hash.as_bytes())
    }

    // ╔════════════════════════════════════════════════════════════════════════════════╗
    // ║                              CONTENT OPERATIONS                                   ║
    // ╚════════════════════════════════════════════════════════════════════════════════╝

    pub async fn add(&self, mut data: Pin<Box<dyn AsyncRead + Send>>) -> Result<AddResponse> {
        let start = Instant::now();

        debug!("Adding content via Iroh");

        // Read the data into a buffer.
        let mut buffer = Vec::new();
        data.read_to_end(&mut buffer)
            .await
            .map_err(|e| GuardianError::Other(format!("Error reading data: {}", e)))?;

        // Convert to bytes::Bytes and save the size.
        let bytes_data = Bytes::from(buffer);
        let data_size = bytes_data.len();

        // Get a reference to the store and clone the reference.
        let store_arc = self.get_store().await?;
        let (temp_tag, store_type_name) = {
            let store_lock = store_arc.read().await;
            match store_lock
                .as_ref()
                .ok_or_else(|| GuardianError::Other("Store not available".to_string()))?
            {
                StoreType::Fs(fs_store) => {
                    let outcome = fs_store
                        .add_bytes(bytes_data.clone())
                        .await
                        .map_err(Self::map_iroh_error)?;
                    (outcome.hash, "FsStore")
                }
            }
        }; // Drop the lock here.

        // Get the hash from the outcome.
        let hash = temp_tag;

        // Convert the BLAKE3 Hash into a hex string.
        let hash_str = Self::hash_to_string(&hash);

        // Add the content to the intelligent cache for fast future access.
        if let Err(e) = self.add_to_cache(&hash_str, bytes_data.clone()).await {
            warn!("Error adding content to the cache: {}", e);
        }

        // Cache already added in the add_to_cache method.

        debug!(
            "Content added with hash: {} using {} (cached)",
            hash_str, store_type_name
        );

        // Update metrics manually.
        let duration = start.elapsed();
        self.update_metrics(duration, true).await;

        // Record the add operation in NetworkingMetrics.
        self.networking_metrics
            .record_add_operation(duration.as_millis() as f64, data_size as u64)
            .await;

        // Use the size saved earlier.
        Ok(AddResponse {
            hash: hash_str,
            name: "unnamed".to_string(),
            size: data_size.to_string(),
        })
    }

    /// Retrieves content from the store by its BLAKE3 hash.
    ///
    /// # Arguments
    /// * `hash_str` - BLAKE3 hash in hexadecimal format
    pub async fn cat(&self, hash_str: &str) -> Result<Pin<Box<dyn AsyncRead + Send>>> {
        let start = Instant::now();

        debug!(
            "Retrieving content {} via Iroh (checking cache first)",
            hash_str
        );

        // First, try to get it from the cache for optimized performance.
        if let Some(cached_data) = self.get_from_cache(hash_str).await {
            debug!(
                "Cache hit! Returning content of {} bytes from the cache",
                cached_data.len()
            );

            // Update metrics with the cache time (very fast).
            let duration = start.elapsed();
            self.update_metrics(duration, true).await;

            // Record the cache cat operation in NetworkingMetrics.
            self.networking_metrics
                .record_cat_operation(duration.as_millis() as f64, cached_data.len() as u64)
                .await;

            // Return the cached data as AsyncRead.
            let cursor = std::io::Cursor::new(cached_data.to_vec());
            return Ok(Box::pin(cursor));
        }

        debug!("Cache miss for {}, fetching from the store", hash_str);

        // Parse the hexadecimal hash into an IrohHash.
        let hash = Self::parse_hash(hash_str)?;

        // Fetch the content from the store.
        let buffer_vec = {
            let store_guard = self.store.read().await;
            let buffer_bytes: bytes::Bytes = match store_guard.as_ref() {
                Some(StoreType::Fs(store)) => {
                    // API 0.94.0: use a direct reader to get the data.
                    let mut reader = store.reader(hash);

                    // Read all the content using read_to_end() with a buffer.
                    let mut buffer = Vec::new();
                    reader
                        .read_to_end(&mut buffer)
                        .await
                        .map_err(Self::map_iroh_error)?;

                    Bytes::from(buffer)
                }
                None => {
                    return Err(GuardianError::Other(
                        "Iroh store not initialized".to_string(),
                    ));
                }
            };

            // Convert from bytes::Bytes to Vec<u8>.
            buffer_bytes.to_vec()
        };

        // Add the retrieved data to the cache for future lookups.
        let buffer_bytes = bytes::Bytes::from(buffer_vec.clone());
        if let Err(e) = self.add_to_cache(hash_str, buffer_bytes).await {
            warn!("Error adding retrieved content to the cache: {}", e);
        } else {
            debug!(
                "Content {} added to the cache after retrieval from the store",
                hash_str
            );
        }

        debug!(
            "Content {} retrieved, {} bytes (cached for the future)",
            hash_str,
            buffer_vec.len()
        );

        // Update success metrics.
        let duration = start.elapsed();
        self.update_metrics(duration, true).await;

        // Record the store cat operation in NetworkingMetrics.
        self.networking_metrics
            .record_cat_operation(duration.as_millis() as f64, buffer_vec.len() as u64)
            .await;

        let cursor = std::io::Cursor::new(buffer_vec);
        Ok(Box::pin(cursor))
    }

    /// Pins an object in the store using Iroh's persistent Tags system.
    ///
    /// Tag lifecycle:
    /// 1. TempTag - Temporarily protects during the operation (automatic drop)
    /// 2. Persistent tag - Created with set_tag(), protects against GC permanently
    /// 3. The tag persists even after the node restarts
    ///
    /// # Arguments
    /// * `hash_str` - BLAKE3 hash in hexadecimal format of the content to pin
    pub async fn pin_add(&self, hash_str: &str) -> Result<()> {
        self.execute_with_metrics(async {
            debug!("Pinning object {} via Iroh using persistent Tags", hash_str);

            // Get a reference to the store.
            let store_arc = self.get_store().await?;

            // Parse the hexadecimal hash.
            let hash = Self::parse_hash(hash_str)?;
            let hash_and_format = HashAndFormat::new(hash, BlobFormat::Raw);

            // Check that the content exists and create a TempTag for protection during the operation.
            let _temp_tag = {
                let store_lock = store_arc.read().await;
                match store_lock.as_ref().unwrap() {
                    StoreType::Fs(fs_store) => {
                        // API 0.94.0: use has to check existence.
                        let has_blob = fs_store.has(hash).await.unwrap_or(false);

                        if !has_blob {
                            return Err(GuardianError::Other(format!(
                                "Content {} not found in the store",
                                hash_str
                            )));
                        }

                        // Return the hash to create a permanent tag.
                        hash_and_format.hash
                    }
                }
            };

            // Create a persistent Tag that survives GC.
            let permanent_tag = {
                let store_lock = store_arc.read().await;
                match store_lock.as_ref().unwrap() {
                    StoreType::Fs(fs_store) => {
                        // Create a permanent tag with a name based on the hash.
                        let tag_name = format!("pin-{}", hash_str);
                        let tag = Tag::from(tag_name.as_str());

                        // Set the tag in the store - this persists to disk.
                        fs_store
                            .tags()
                            .set(tag.as_ref(), hash_and_format)
                            .await
                            .map_err(Self::map_iroh_error)?;

                        debug!("Persistent tag '{}' created for hash {}", tag_name, hash);
                        tag
                    }
                }
            };

            // Add it to the local cache for fast tracking.
            {
                let mut pinned = self.pinned_cache.lock().await;
                pinned.insert(hash_str.to_string(), PinType::Direct);
            }

            info!(
                "Object {} pinned successfully using persistent Tag: {}",
                hash_str, permanent_tag
            );
            Ok(())
        })
        .await
    }

    /// Removes the pin from an object using Store::delete_tag().
    ///
    /// Removes the persistent Tag associated with the hash, allowing GC
    /// to remove the content in future runs.
    ///
    /// # Arguments
    /// * `hash_str` - BLAKE3 hash in hexadecimal format of the content to unpin
    pub async fn pin_rm(&self, hash_str: &str) -> Result<()> {
        self.execute_with_metrics(async {
            debug!(
                "Unpinning object {} via Iroh by removing the permanent Tag",
                hash_str
            );

            // First check whether it is pinned in the local cache.
            let was_cached = {
                let mut cache = self.pinned_cache.lock().await;
                cache.remove(hash_str).is_some()
            };

            if !was_cached {
                return Err(GuardianError::Other(format!(
                    "Object {} was not pinned",
                    hash_str
                )));
            }

            // Get a reference to the store to remove the permanent tag.
            let store_arc = self.get_store().await?;

            // Remove the permanent tag from the Iroh store.
            {
                let store_lock = store_arc.read().await;
                match store_lock.as_ref().unwrap() {
                    StoreType::Fs(fs_store) => {
                        // Tag name based on the hash (same pattern used in pin_add).
                        let tag_name = format!("pin-{}", hash_str);
                        let tag = Tag::from(tag_name.as_str());

                        // Remove the tag from the store using delete_tag.
                        fs_store
                            .tags()
                            .delete(tag.as_ref())
                            .await
                            .map_err(Self::map_iroh_error)?;

                        debug!("Permanent tag '{}' removed from the store", tag_name);
                    }
                }
            }

            info!(
                "Object {} unpinned successfully - permanent Tag removed from Iroh",
                hash_str
            );
            Ok(())
        })
        .await
    }

    /// Lists all pinned objects using the Store::tags() iterator.
    ///
    /// Iterates over all persistent Tags in the store and filters those that
    /// start with "pin-" (the convention used in pin_add()).
    ///
    /// # Returns
    /// A Vec with information about each pinned object (hash and pin type)
    pub async fn pin_ls(&self) -> Result<Vec<PinInfo>> {
        self.execute_with_metrics(async {
            debug!("Listing pinned objects via Iroh through the persistent Tags");

            // Get a reference to the store to list tags.
            let store_arc = self.get_store().await?;
            let mut pins = Vec::new();

            // List all tags in the Iroh store.
            {
                let store_lock = store_arc.read().await;
                match store_lock.as_ref().unwrap() {
                    StoreType::Fs(fs_store) => {
                        use futures::stream::StreamExt; // To use next().

                        // Get a stream of all tags in the store.
                        let mut tags_stream =
                            fs_store.tags().list().await.map_err(Self::map_iroh_error)?;

                        // Process each tag to find pins (tags that start with "pin-").
                        while let Some(tag_result) = tags_stream.next().await {
                            match tag_result {
                                Ok(tag_info) => {
                                    let tag_name = String::from_utf8_lossy(tag_info.name.as_ref());

                                    // Check whether it is a pin tag.
                                    if let Some(hash_str) = tag_name.strip_prefix("pin-") {
                                        // Extract the hash from the tag name.

                                        // Determine the pin type based on the format.
                                        let pin_type = match tag_info.format {
                                            BlobFormat::Raw => PinType::Recursive,
                                            BlobFormat::HashSeq => PinType::Direct,
                                        };

                                        pins.push(PinInfo {
                                            hash: hash_str.to_string(),
                                            pin_type: pin_type.clone(),
                                        });

                                        debug!("Pin found: {} (type: {:?})", hash_str, pin_type);
                                    }
                                }
                                Err(e) => {
                                    warn!("Error processing tag during pin listing: {}", e);
                                    // Continue with the other tags.
                                }
                            }
                        }
                    }
                }
            }

            // Also check the local cache for compatibility (it may have unsynced pins).
            {
                let cache = self.pinned_cache.lock().await;
                for (hash_str, pin_type) in cache.iter() {
                    // Avoid duplicates - only add if not already found in the tags.
                    if !pins.iter().any(|p| &p.hash == hash_str) {
                        pins.push(PinInfo {
                            hash: hash_str.clone(),
                            pin_type: pin_type.clone(),
                        });
                        debug!(
                            "Pin from local cache added: {} (type: {:?})",
                            hash_str, pin_type
                        );
                    }
                }
            }

            info!("Found {} pinned objects via Iroh Tags", pins.len());
            Ok(pins)
        })
        .await
    }

    // ╔════════════════════════════════════════════════════════════════════════════════╗
    // ║                       NETWORK AND CONNECTIVITY OPERATIONS                         ║
    // ╚════════════════════════════════════════════════════════════════════════════════╝

    pub async fn peers(&self) -> Result<Vec<PeerInfo>> {
        self.execute_with_metrics(async {
            debug!("Listing connected peers via the Iroh Endpoint and Connection Pool");

            // Get a reference to the endpoint.
            let endpoint_arc = self.get_endpoint().await?;
            let endpoint_lock = endpoint_arc.read().await;
            let endpoint = endpoint_lock
                .as_ref()
                .ok_or_else(|| GuardianError::Other("Endpoint not available".to_string()))?;

            // Get connection information from the endpoint.
            let local_addr = endpoint
                .bound_sockets()
                .into_iter()
                .next()
                .map(|socket_addr| socket_addr.to_string())
                .unwrap_or_else(|| "0.0.0.0:0".to_string());

            let mut peers = Vec::new();
            let mut node_ids_seen = std::collections::HashSet::new();

            debug!("Local endpoint bound at: {}", local_addr);

            // First, get peers from the connection pool (confirmed active connections).
            {
                let pool = self.connection_pool.read().await;
                debug!("Connection pool contains {} active connections", pool.len());

                for conn_info in pool.values() {
                    node_ids_seen.insert(conn_info.node_id);

                    peers.push(PeerInfo {
                        id: conn_info.node_id,
                        addresses: vec![conn_info.address.clone()],
                        protocols: vec![
                            "iroh/blobs/0.92.0".to_string(),
                            "iroh/gossip/0.92.0".to_string(),
                            "iroh/docs/0.92.0".to_string(),
                        ],
                        connected: conn_info.last_used.elapsed() < Duration::from_secs(60),
                    });
                }
            }

            // Then, add peers from the discovery cache that are not in the pool.
            let discovered_peers = {
                let discovery_cache = self.discovery_cache.read().await;
                discovery_cache.peers.values().cloned().collect::<Vec<_>>()
            };

            // Convert discovery-cache peers into PeerInfo (avoiding duplicates).
            for discovered_peer in discovered_peers {
                // Avoid duplicates.
                if node_ids_seen.contains(&discovered_peer.node_id) {
                    continue;
                }
                node_ids_seen.insert(discovered_peer.node_id);

                peers.push(PeerInfo {
                    id: discovered_peer.node_id,
                    addresses: discovered_peer.addresses.clone(),
                    protocols: discovered_peer.protocols.clone(),
                    connected: discovered_peer.last_seen.elapsed() < Duration::from_secs(30),
                });
            }

            // API CHANGE (Iroh 1.0): Endpoint::remote_info_iter() was removed — there is no
            // longer enumeration of all known remotes. The peer list is assembled from the
            // connection pool and the discovery cache above. For a specific peer, use
            // remote_info(id).await or address_lookup().resolve(id).
            let _ = &node_ids_seen;

            info!(
                "Found {} peers (connection pool + discovery cache)",
                peers.len()
            );
            Ok(peers)
        })
        .await
    }

    pub async fn id(&self) -> Result<NodeInfo> {
        self.execute_with_metrics(async {
            debug!("Getting node information via the Iroh Endpoint");

            // Get a reference to the endpoint.
            let endpoint_arc = self.get_endpoint().await?;
            let endpoint_lock = endpoint_arc.read().await;
            let endpoint = endpoint_lock
                .as_ref()
                .ok_or_else(|| GuardianError::Other("Endpoint not available".to_string()))?;

            // Get the EndpointId from the endpoint (Iroh 1.0: node_id() -> id()).
            let node_id = endpoint.id();

            // Get the endpoint's network addresses.
            let addresses: Vec<String> = endpoint
                .bound_sockets()
                .into_iter()
                .map(|addr| addr.to_string())
                .collect();

            debug!("Iroh NodeId: {}", node_id);
            debug!("Bound addresses: {:?}", addresses);

            Ok(NodeInfo {
                id: node_id,
                public_key: format!("iroh-node-{}", node_id),
                addresses,
                agent_version: "guardian-db-iroh/0.1.0".to_string(),
                protocol_version: "iroh-protocols/0.92.0".to_string(),
            })
        })
        .await
    }

    // ╔════════════════════════════════════════════════════════════════════════════════╗
    // ║                      REPOSITORY AND VERSION OPERATIONS                            ║
    // ╚════════════════════════════════════════════════════════════════════════════════╝

    pub async fn repo_stat(&self) -> Result<RepoStats> {
        self.execute_with_metrics(async {
            debug!("Getting repository statistics via the Iroh FsStore");

            let store_path = self.data_dir.join("iroh_store");

            // Try to get statistics from the store directory.
            let (num_objects, repo_size) = match tokio::fs::read_dir(&store_path).await {
                Ok(mut entries) => {
                    let mut count = 0;
                    let mut total_size = 0;

                    while let Some(entry) = entries.next_entry().await.unwrap_or(None) {
                        if let Ok(metadata) = entry.metadata().await
                            && metadata.is_file()
                        {
                            count += 1;
                            total_size += metadata.len();
                        }
                    }

                    (count, total_size)
                }
                Err(_) => (0, 0), // Fallback if the directory cannot be read.
            };

            Ok(RepoStats {
                num_objects: num_objects as u64,
                repo_size,
                repo_path: store_path.to_string_lossy().to_string(),
                version: "15".to_string(), // Version compatible with FsStore.
            })
        })
        .await
    }

    pub async fn version(&self) -> Result<VersionInfo> {
        self.execute_with_metrics(async {
            Ok(VersionInfo {
                version: "iroh-0.92.0".to_string(),
                commit: "embedded".to_string(),
                repo: "15".to_string(), // iroh repo version.
                system: std::env::consts::OS.to_string(),
            })
        })
        .await
    }

    // ╔════════════════════════════════════════════════════════════════════════════════╗
    // ║                    METADATA, STATUS AND HEALTH CHECKS                             ║
    // ╚════════════════════════════════════════════════════════════════════════════════╝

    pub async fn is_online(&self) -> bool {
        let status = self.node_status.read().await;
        status.is_online
    }

    pub async fn metrics(&self) -> Result<BackendMetrics> {
        let mut metrics = self.metrics.read().await.clone();

        // Add cache information to the metrics using OptimizedCache.
        let cache_stats = self.optimized_cache.get_stats().await;
        let hit_ratio = cache_stats.hit_rate;

        // Add estimated memory usage including the cache.
        metrics.memory_usage_bytes = self.estimate_memory_usage().await;

        // Update ops_per_second based on cache performance.
        if hit_ratio > 0.0 {
            // Cache hits significantly improve performance.
            metrics.ops_per_second *= 1.0 + (hit_ratio * 2.0); // Boost based on hit ratio.
        }

        debug!(
            "Performance metrics - Hit ratio: {:.2}%, Total bytes cached: {}",
            hit_ratio * 100.0,
            cache_stats.total_bytes_cached
        );

        Ok(metrics)
    }

    pub async fn health_check(&self) -> Result<HealthStatus> {
        let start = Instant::now();
        let mut checks = Vec::new();
        let mut healthy = true;

        // Check 1: Node status.
        {
            let status = self.node_status.read().await;
            checks.push(HealthCheck {
                name: "node_status".to_string(),
                passed: status.is_online,
                message: if status.is_online {
                    "Iroh node online".to_string()
                } else {
                    format!(
                        "Iroh node offline: {}",
                        status.last_error.as_deref().unwrap_or("unknown reason")
                    )
                },
            });

            if !status.is_online {
                healthy = false;
            }
        }

        // Check 2: Data directory accessible.
        let data_check = tokio::fs::metadata(&self.data_dir).await.is_ok();
        checks.push(HealthCheck {
            name: "data_directory".to_string(),
            passed: data_check,
            message: if data_check {
                "Data directory accessible".to_string()
            } else {
                "Data directory inaccessible".to_string()
            },
        });

        if !data_check {
            healthy = false;
        }

        // Check 3: Basic metrics.
        let metrics_check = self.metrics().await.is_ok();
        checks.push(HealthCheck {
            name: "metrics".to_string(),
            passed: metrics_check,
            message: if metrics_check {
                "Metrics available".to_string()
            } else {
                "Error accessing metrics".to_string()
            },
        });

        let response_time = start.elapsed();

        let message = if healthy {
            "Iroh backend operational".to_string()
        } else {
            "Iroh backend has problems".to_string()
        };

        Ok(HealthStatus {
            healthy,
            message,
            response_time_ms: response_time.as_millis() as u64,
            checks,
        })
    }

    // ╔════════════════════════════════════════════════════════════════════════════════╗
    // ║                      OPTIMIZATIONS AND CACHE MANAGEMENT                           ║
    // ╚════════════════════════════════════════════════════════════════════════════════╝

    // === METRICS AND MONITORING ===
    /// Estimates the backend's memory usage.
    async fn estimate_memory_usage(&self) -> u64 {
        let pinned_cache_size = self.pinned_cache.lock().await.len() as u64 * 64;

        // Use statistics from OptimizedCache.
        let cache_stats = self.optimized_cache.get_stats().await;
        let data_cache_size = cache_stats.total_bytes_cached;

        // Estimate the discovery cache overhead.
        let discovery_cache_size = {
            let discovery_cache = self.discovery_cache.read().await;
            discovery_cache.peers.len() as u64 * 256 // Estimate per peer.
        };

        pinned_cache_size + data_cache_size + discovery_cache_size
    }

    // === PEER DISCOVERY ===
    /// Discovers a specific peer.
    pub async fn discover_peer_with_endpoint(&mut self, node_id: NodeId) -> Result<Vec<NodeAddr>> {
        debug!(
            "Discovering peer {} using the IrohBackend's concrete resources",
            node_id
        );

        // Use the Endpoint directly for discovery.
        let discovered_addresses = self.discover_peer_integrated(node_id).await?;

        if discovered_addresses.is_empty() {
            debug!("No address found for peer {}", node_id);
            return Err(GuardianError::Other(format!(
                "No address found for peer {}",
                node_id
            )));
        }

        debug!(
            "Peer {} discovered successfully: {} addresses",
            node_id,
            discovered_addresses.len()
        );

        // Log discovery success.
        info!(
            "Successful discovery: {} addresses for peer {}",
            discovered_addresses.len(),
            node_id
        );

        Ok(discovered_addresses)
    }

    /// Gets statistics from the optimized cache.
    pub async fn get_cache_statistics(&self) -> Result<SimpleCacheStats> {
        let cache_stats = self.optimized_cache.get_stats().await;

        // Convert OptimizedCache's CacheStats into SimpleCacheStats (public API).
        Ok(SimpleCacheStats {
            entries_count: 0, // OptimizedCache does not expose a direct count.
            hit_ratio: cache_stats.hit_rate,
            total_size_bytes: cache_stats.total_bytes_cached,
        })
    }

    /// Runs automatic performance optimization.
    pub async fn optimize_performance(&self) -> Result<()> {
        debug!("Starting automatic performance optimization");

        // Optimize the cache based on metrics.
        self.optimize_cache_with_metrics().await?;

        // 3. Update performance metrics.
        {
            let stats = self.get_cache_statistics().await?;
            let mut metrics = self.metrics.write().await;

            // Adjust ops_per_second based on cache performance.
            let hit_ratio = stats.hit_ratio;

            // Performance boost based on the hit ratio.
            if hit_ratio > 0.5 {
                metrics.ops_per_second = (metrics.ops_per_second * (1.0 + hit_ratio)).max(10.0);
            }

            metrics.avg_latency_ms = if hit_ratio > 0.8 { 0.5 } else { 1.0 };
        }

        info!(
            "Performance optimization complete with hit ratio: {:.2}",
            self.get_cache_statistics().await?.hit_ratio
        );
        Ok(())
    }

    /// Optimizes the cache based on usage metrics.
    async fn optimize_cache_with_metrics(&self) -> Result<()> {
        let cache_stats = self.optimized_cache.get_stats().await;
        let hit_ratio = cache_stats.hit_rate;

        debug!(
            "Optimizing cache - Current Hit Ratio: {:.2}%",
            hit_ratio * 100.0
        );

        // OptimizedCache manages intelligent eviction automatically
        // when the configured threshold is reached.
        if hit_ratio < 0.3 {
            info!(
                "Low hit ratio detected ({:.1}%) - OptimizedCache will manage eviction automatically",
                hit_ratio * 100.0
            );
        }

        Ok(())
    }

    /// Uses the configuration for dynamic adjustments.
    pub async fn get_config_info(&self) -> String {
        format!(
            "Backend configured with data_store_path: {:?}",
            self.config.data_store_path
        )
    }

    /// Gets information about the connection pool.
    pub async fn get_connection_pool_status(&self) -> String {
        let pool = self.connection_pool.read().await;
        format!("Connection pool active with {} peers", pool.len())
    }

    /// Gets a connection from the pool, or returns an error if it does not exist.
    pub async fn get_connection_from_pool(&self, node_id: &NodeId) -> Result<ConnectionInfo> {
        let mut pool = self.connection_pool.write().await;

        if let Some(conn_info) = pool.get_mut(node_id) {
            // Update the last-used timestamp.
            conn_info.last_used = Instant::now();
            conn_info.operations_count += 1;

            debug!(
                "Connection obtained from the pool: {} (operations: {})",
                node_id.fmt_short(),
                conn_info.operations_count
            );

            Ok(conn_info.clone())
        } else {
            Err(GuardianError::Other(format!(
                "Connection not found in the pool: {}",
                node_id.fmt_short()
            )))
        }
    }

    /// Removes a connection from the pool.
    pub async fn remove_connection_from_pool(&self, node_id: &NodeId) -> Result<()> {
        let mut pool = self.connection_pool.write().await;

        if pool.remove(node_id).is_some() {
            info!(
                "Connection removed from the pool: {} ({} connections remaining)",
                node_id.fmt_short(),
                pool.len()
            );

            // Update the connected-peers counter.
            let mut status = self.node_status.write().await;
            status.connected_peers = status.connected_peers.saturating_sub(1);

            Ok(())
        } else {
            Err(GuardianError::Other(format!(
                "Connection not found in the pool: {}",
                node_id.fmt_short()
            )))
        }
    }

    /// Clears stale connections from the pool (unused for longer than the timeout).
    pub async fn cleanup_stale_connections(&self, timeout: Duration) -> Result<u32> {
        let mut pool = self.connection_pool.write().await;
        let mut removed_count = 0;

        let now = Instant::now();
        let stale_peers: Vec<NodeId> = pool
            .iter()
            .filter(|(_, conn)| now.saturating_duration_since(conn.last_used) > timeout)
            .map(|(id, _)| *id)
            .collect();

        for node_id in stale_peers {
            pool.remove(&node_id);
            removed_count += 1;
            debug!(
                "Stale connection removed from the pool: {}",
                node_id.fmt_short()
            );
        }

        if removed_count > 0 {
            info!(
                "Connection pool cleanup: {} stale connections removed",
                removed_count
            );

            // Update the connected-peers counter.
            let mut status = self.node_status.write().await;
            status.connected_peers = pool.len() as u32;
        }

        Ok(removed_count)
    }

    /// Lists all active connections in the pool.
    pub async fn list_active_connections(&self) -> Vec<ConnectionInfo> {
        let pool = self.connection_pool.read().await;
        pool.values().cloned().collect()
    }

    /// Updates the latency of a connection in the pool.
    pub async fn update_connection_latency(&self, node_id: &NodeId, latency_ms: f64) -> Result<()> {
        // Record the raw sample for per-peer percentiles (C1), independent of
        // whether the peer is currently in the connection pool.
        self.record_peer_latency_sample(node_id, latency_ms).await;

        let mut pool = self.connection_pool.write().await;

        if let Some(conn_info) = pool.get_mut(node_id) {
            // Exponential moving average to smooth out fluctuations.
            conn_info.avg_latency_ms = if conn_info.avg_latency_ms == 0.0 {
                latency_ms
            } else {
                conn_info.avg_latency_ms * 0.7 + latency_ms * 0.3
            };

            debug!(
                "Latency updated for {}: {:.2}ms",
                node_id.fmt_short(),
                conn_info.avg_latency_ms
            );

            Ok(())
        } else {
            Err(GuardianError::Other(format!(
                "Connection not found in the pool: {}",
                node_id.fmt_short()
            )))
        }
    }

    /// Appends a latency sample to a peer's bounded history ring (C1).
    async fn record_peer_latency_sample(&self, node_id: &NodeId, latency_ms: f64) {
        let mut hist = self.peer_latency_history.write().await;
        let samples = hist.entry(*node_id).or_default();
        samples.push_back(latency_ms);
        while samples.len() > PEER_LATENCY_HISTORY_CAP {
            samples.pop_front();
        }
    }

    /// Per-peer p95/p99 latency (ms) from the peer's sample history (C1). Returns
    /// `None` until at least a few samples exist, so callers don't show noise.
    pub async fn peer_latency_percentiles(&self, node_id: &NodeId) -> Option<(f64, f64)> {
        let hist = self.peer_latency_history.read().await;
        let samples = hist.get(node_id)?;
        if samples.len() < 4 {
            return None;
        }
        let mut sorted: Vec<f64> = samples.iter().copied().collect();
        sorted.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
        let pick = |q: f64| {
            let idx = ((sorted.len() as f64 * q) as usize).min(sorted.len() - 1);
            sorted[idx]
        };
        Some((pick(0.95), pick(0.99)))
    }

    /// Peers we have learned about (via connection/ticket exchange) that are **not**
    /// currently in the active connection pool (C3, honest version).
    ///
    /// Note: iroh 1.0 exposes no passive enumeration of *discovery-only* peers, so
    /// this is the set of previously-known peers minus the live connections — an
    /// observed view, not the full discovery table.
    pub async fn discovered_not_connected(&self) -> Vec<NodeId> {
        let connected: std::collections::HashSet<NodeId> = {
            let pool = self.connection_pool.read().await;
            pool.keys().copied().collect()
        };
        let known = self.known_peers.read().await;
        known
            .iter()
            .filter(|p| !connected.contains(*p))
            .copied()
            .collect()
    }

    // === NODE INFO ===
    /// Returns the node's secret key.
    pub fn secret_key(&self) -> &SecretKey {
        &self.secret_key
    }

    /// Returns a reference to the backend configuration.
    pub fn config(&self) -> &ClientConfig {
        &self.config
    }

    // === KEY SYNCHRONIZATION ===
    /// Gets a reference to the key synchronizer.
    pub fn get_key_synchronizer(
        &self,
    ) -> Arc<crate::p2p::network::core::key_synchronizer::KeySynchronizer> {
        self.key_synchronizer.clone()
    }

    /// Adds a trusted peer to the key synchronizer.
    pub async fn add_trusted_peer_for_sync(
        &self,
        node_id: NodeId,
        public_key: ed25519_dalek::VerifyingKey,
    ) -> Result<()> {
        self.key_synchronizer
            .add_trusted_peer(node_id, public_key)
            .await
    }

    /// Removes a trusted peer from the key synchronizer.
    pub async fn remove_trusted_peer_from_sync(&self, node_id: &NodeId) -> Result<bool> {
        self.key_synchronizer.remove_trusted_peer(node_id).await
    }

    /// Synchronizes a specific key with peers.
    pub async fn sync_key_with_peers(
        &self,
        key_id: &str,
        operation: crate::p2p::network::core::key_synchronizer::SyncOperation,
    ) -> Result<()> {
        self.key_synchronizer.sync_key(key_id, operation).await
    }

    /// Gets key synchronization statistics.
    pub async fn get_key_sync_statistics(
        &self,
    ) -> crate::p2p::network::core::key_synchronizer::SyncStatistics {
        self.key_synchronizer.get_statistics().await
    }

    /// Gets the synchronization status of a key.
    pub async fn get_key_sync_status(
        &self,
        key_id: &str,
    ) -> Option<crate::p2p::network::core::key_synchronizer::KeySyncStatus> {
        self.key_synchronizer.get_key_sync_status(key_id).await
    }

    /// Lists synchronized keys.
    pub async fn list_synchronized_keys(&self) -> Vec<String> {
        self.key_synchronizer.list_synchronized_keys().await
    }

    /// Lists trusted peers for synchronization.
    pub async fn list_trusted_peers_for_sync(&self) -> Vec<NodeId> {
        self.key_synchronizer.list_trusted_peers().await
    }

    /// Processes a received synchronization message.
    pub async fn handle_sync_message(
        &self,
        message: crate::p2p::network::core::key_synchronizer::SyncMessage,
    ) -> Result<()> {
        self.key_synchronizer.handle_sync_message(message).await
    }

    /// Forces a full synchronization of all keys.
    pub async fn force_full_key_sync(&self) -> Result<()> {
        self.key_synchronizer.force_full_sync().await
    }

    /// Exports the synchronization configuration.
    pub async fn export_key_sync_config(&self) -> Result<Vec<u8>> {
        self.key_synchronizer.export_sync_config().await
    }

    /// Clears the cache of old messages (simplified method).
    pub async fn cleanup_sync_cache(&self) -> Result<u64> {
        // KeySynchronizer does not expose a public cleanup method.
        // This is a placeholder for future compatibility.
        Ok(0)
    }

    /// Exports synchronization statistics as JSON.
    pub async fn export_sync_statistics_json(&self) -> Result<String> {
        let stats = self.get_key_sync_statistics().await;
        serde_json::to_string_pretty(&stats)
            .map_err(|e| GuardianError::Other(format!("Error serializing statistics: {}", e)))
    }

    /// Generates a key synchronization report.
    pub async fn generate_key_sync_report(&self) -> String {
        let stats = self.get_key_sync_statistics().await;
        let trusted_peers = self.list_trusted_peers_for_sync().await;

        format!(
            r#"
=== KEY SYNCHRONIZATION REPORT ===

General Statistics:
   - Messages synchronized: {}
   - Pending messages: {}
   - Success rate: {:.1}%
   - Average latency: {:.2}ms

Conflicts:
   - Detected: {}
   - Resolved: {}
   - Resolution rate: {:.1}%

Peers:
   - Active peers: {}
   - Trusted peers: {}

Status: {}
"#,
            stats.messages_synced,
            stats.pending_messages,
            stats.success_rate * 100.0,
            stats.avg_sync_latency_ms,
            stats.conflicts_detected,
            stats.conflicts_resolved,
            if stats.conflicts_detected > 0 {
                (stats.conflicts_resolved as f64 / stats.conflicts_detected as f64) * 100.0
            } else {
                100.0
            },
            stats.active_peers,
            trusted_peers.len(),
            if stats.success_rate > 0.95 {
                "✓ Healthy"
            } else if stats.success_rate > 0.80 {
                "⚠ Attention"
            } else {
                "✗ Critical"
            }
        )
    }

    // === NETWORKING METRICS ===
    /// Gets up-to-date networking metrics.
    pub async fn get_networking_metrics(&self) -> Result<networking_metrics::NetworkingMetrics> {
        // Update the computed metrics before returning.
        self.networking_metrics.update_computed_metrics().await;
        Ok(self.networking_metrics.get_metrics().await)
    }

    /// Generates a detailed networking metrics report.
    pub async fn generate_networking_report(&self) -> String {
        self.networking_metrics.update_computed_metrics().await;
        self.networking_metrics.generate_report().await
    }

    /// Exports networking metrics as JSON.
    pub async fn export_networking_metrics_json(&self) -> Result<String> {
        self.networking_metrics.update_computed_metrics().await;
        self.networking_metrics.export_json().await
    }

    // === PERFORMANCE MONITORING ===
    /// Gets the performance monitor status.
    pub async fn get_performance_monitor_status(&self) -> String {
        let monitor = self.performance_monitor.read().await;
        format!(
            "Performance monitor active - Throughput: {:.2} ops/s",
            monitor.throughput_metrics.ops_per_second
        )
    }

    /// Gets a reference to the performance monitor.
    pub fn get_performance_monitor(&self) -> Arc<RwLock<PerformanceMonitor>> {
        self.performance_monitor.clone()
    }

    /// Gets the throughput metrics.
    pub async fn get_throughput_metrics(&self) -> ThroughputMetrics {
        let monitor = self.performance_monitor.read().await;
        monitor.throughput_metrics.clone()
    }

    /// Gets the latency metrics.
    pub async fn get_latency_metrics(&self) -> LatencyMetrics {
        let monitor = self.performance_monitor.read().await;
        monitor.latency_metrics.clone()
    }

    /// Gets the resource metrics.
    pub async fn get_resource_metrics(&self) -> ResourceMetrics {
        let monitor = self.performance_monitor.read().await;
        monitor.resource_metrics.clone()
    }

    /// Creates a snapshot of the current performance.
    pub async fn create_performance_snapshot(&self) -> PerformanceSnapshot {
        let monitor = self.performance_monitor.read().await;
        PerformanceSnapshot {
            timestamp: Instant::now(),
            throughput: monitor.throughput_metrics.clone(),
            latency: monitor.latency_metrics.clone(),
            resources: monitor.resource_metrics.clone(),
        }
    }

    /// Gets the history of performance snapshots.
    pub async fn get_performance_history(&self) -> Vec<PerformanceSnapshot> {
        let monitor = self.performance_monitor.read().await;
        monitor.performance_history.clone()
    }

    /// Adds a snapshot to the history (limited to the last 100).
    pub async fn record_performance_snapshot(&self) -> Result<()> {
        let snapshot = self.create_performance_snapshot().await;
        let mut monitor = self.performance_monitor.write().await;

        monitor.performance_history.push(snapshot);

        // Keep only the last 100 snapshots.
        if monitor.performance_history.len() > 100 {
            monitor.performance_history.remove(0);
        }

        Ok(())
    }

    /// Updates resource metrics manually.
    pub async fn update_resource_metrics(
        &self,
        cpu_usage: f64,
        memory_bytes: u64,
        disk_io_bps: u64,
        network_bps: u64,
    ) -> Result<()> {
        let mut monitor = self.performance_monitor.write().await;

        monitor.resource_metrics.cpu_usage = cpu_usage.clamp(0.0, 1.0);
        monitor.resource_metrics.memory_usage_bytes = memory_bytes;
        monitor.resource_metrics.disk_io_bps = disk_io_bps;
        monitor.resource_metrics.network_bandwidth_bps = network_bps;

        Ok(())
    }

    /// Resets the performance metrics.
    pub async fn reset_performance_metrics(&self) -> Result<()> {
        let mut monitor = self.performance_monitor.write().await;

        *monitor = PerformanceMonitor::default();

        info!("Performance metrics reset");
        Ok(())
    }

    /// Computes latency percentiles (P95, P99).
    pub async fn calculate_latency_percentiles(&self) -> Result<(f64, f64)> {
        let monitor = self.performance_monitor.read().await;

        if monitor.performance_history.is_empty() {
            return Ok((0.0, 0.0));
        }

        let mut latencies: Vec<f64> = monitor
            .performance_history
            .iter()
            .map(|s| s.latency.avg_latency_ms)
            .collect();

        latencies.sort_by(|a, b| a.partial_cmp(b).unwrap());

        let p95_idx = (latencies.len() as f64 * 0.95) as usize;
        let p99_idx = (latencies.len() as f64 * 0.99) as usize;

        let p95 = latencies.get(p95_idx).copied().unwrap_or(0.0);
        let p99 = latencies.get(p99_idx).copied().unwrap_or(0.0);

        // Update it in the monitor.
        drop(monitor);
        let mut monitor_mut = self.performance_monitor.write().await;
        monitor_mut.latency_metrics.p95_latency_ms = p95;
        monitor_mut.latency_metrics.p99_latency_ms = p99;

        Ok((p95, p99))
    }

    /// Generates a detailed performance monitor report.
    pub async fn generate_performance_monitor_report(&self) -> String {
        let monitor = self.performance_monitor.read().await;
        let (p95, p99) = self
            .calculate_latency_percentiles()
            .await
            .unwrap_or((0.0, 0.0));

        format!(
            r#"
=== PERFORMANCE MONITOR REPORT ===

Throughput:
   - Operations/second: {:.2}
   - Bytes/second: {}
   - Peak throughput: {:.2} ops/s
   - Average throughput: {:.2} ops/s

Latency:
   - Average latency: {:.2}ms
   - Minimum latency: {:.2}ms
   - Maximum latency: {:.2}ms
   - P95 latency: {:.2}ms
   - P99 latency: {:.2}ms

Resources:
   - CPU usage: {:.1}%
   - Memory usage: {:.2}MB
   - Disk I/O: {:.2}MB/s
   - Bandwidth: {:.2}MB/s

History:
   - Snapshots recorded: {}
   - Monitored period: {} snapshots

Status: {}
"#,
            monitor.throughput_metrics.ops_per_second,
            monitor.throughput_metrics.bytes_per_second,
            monitor.throughput_metrics.peak_throughput,
            monitor.throughput_metrics.avg_throughput,
            monitor.latency_metrics.avg_latency_ms,
            monitor.latency_metrics.min_latency_ms,
            monitor.latency_metrics.max_latency_ms,
            p95,
            p99,
            monitor.resource_metrics.cpu_usage * 100.0,
            monitor.resource_metrics.memory_usage_bytes as f64 / 1_048_576.0,
            monitor.resource_metrics.disk_io_bps as f64 / 1_048_576.0,
            monitor.resource_metrics.network_bandwidth_bps as f64 / 1_048_576.0,
            monitor.performance_history.len(),
            monitor.performance_history.len(),
            if monitor.latency_metrics.avg_latency_ms < 50.0 {
                "✓ Excellent"
            } else if monitor.latency_metrics.avg_latency_ms < 100.0 {
                "✓ Good"
            } else if monitor.latency_metrics.avg_latency_ms < 200.0 {
                "⚠ Moderate"
            } else {
                "✗ Critical"
            }
        )
    }
    /// Generates a detailed performance report.
    pub async fn generate_performance_report(&self) -> String {
        let cache_stats = self.get_cache_statistics().await.unwrap_or_default();
        let metrics = self.metrics.read().await;
        let memory_usage = self.estimate_memory_usage().await;

        let hit_ratio = cache_stats.hit_ratio;

        // Connection pool information.
        let (pool_size, avg_pool_latency, total_pool_operations) = {
            let pool = self.connection_pool.read().await;
            let size = pool.len();
            let avg_latency = if !pool.is_empty() {
                pool.values().map(|c| c.avg_latency_ms).sum::<f64>() / size as f64
            } else {
                0.0
            };
            let total_ops = pool.values().map(|c| c.operations_count).sum::<u64>();
            (size, avg_latency, total_ops)
        };

        // Key synchronizer information.
        let sync_stats = self.get_key_sync_statistics().await;
        let trusted_peers_count = self.list_trusted_peers_for_sync().await.len();

        // Performance monitor information.
        let perf_throughput = self.get_throughput_metrics().await;
        let perf_latency = self.get_latency_metrics().await;
        let perf_resources = self.get_resource_metrics().await;
        let perf_history_count = self.get_performance_history().await.len();

        format!(
            r#"
IROH BACKEND PERFORMANCE REPORT

General Metrics:
   - Operations per second: {:.2}
   - Average latency: {:.2}ms
   - Total operations: {}
   - Errors: {}
   - Memory usage: {:.2}MB

Cache Statistics:
   - Cache hits: {}
   - Cache misses: {}
   - Hit ratio: {:.1}%
   - Bytes cached: {:.2}MB
   - Cache entries: {}
   - Bytes saved: {:.2}MB
   - Average access time: {:.2}ms

Connection Pool:
   - Active connections: {}
   - Average pool latency: {:.2}ms
   - Total operations via pool: {}
   - Reuse efficiency: {:.1}%

Key Synchronization:
   - Messages synchronized: {}
   - Pending messages: {}
   - Success rate: {:.1}%
   - Conflicts (resolved/total): {}/{}
   - Trusted peers: {}
   - Average sync latency: {:.2}ms

Performance Monitor:
   - Throughput: {:.2} ops/s (peak: {:.2})
   - Bytes/second: {}
   - Average latency: {:.2}ms
   - Latency (min/max): {:.2}ms / {:.2}ms
   - Latency P95/P99: {:.2}ms / {:.2}ms
   - CPU usage: {:.1}%
   - Memory usage: {:.2}MB
   - Disk I/O: {:.2}MB/s
   - Snapshots recorded: {}

Optimizations:
   - Intelligent cache: ✓ Active
   - Connection pooling: ✓ Active
   - Key synchronization: ✓ Active
   - Performance monitoring: ✓ Active
   - Adaptive eviction: ✓ Configured
   - Dynamic prioritization: ✓ Working
   - Discovery caching: ✓ Integrated

Performance Score: {:.1}/10
"#,
            metrics.ops_per_second,
            metrics.avg_latency_ms,
            metrics.total_operations,
            metrics.error_count,
            memory_usage as f64 / 1_048_576.0,
            cache_stats.entries_count, // estimated hits
            0,                         // misses (not available in SimpleCacheStats)
            hit_ratio * 100.0,
            cache_stats.total_size_bytes as f64 / 1_048_576.0,
            cache_stats.entries_count,
            cache_stats.total_size_bytes as f64 / 1_048_576.0, // estimated bytes saved
            1.0,                                               // fast access time for LRU
            pool_size,
            avg_pool_latency,
            total_pool_operations,
            if pool_size > 0 {
                (total_pool_operations as f64 / pool_size as f64) * 10.0
            } else {
                0.0
            },
            sync_stats.messages_synced,
            sync_stats.pending_messages,
            sync_stats.success_rate * 100.0,
            sync_stats.conflicts_resolved,
            sync_stats.conflicts_detected,
            trusted_peers_count,
            sync_stats.avg_sync_latency_ms,
            perf_throughput.ops_per_second,
            perf_throughput.peak_throughput,
            perf_throughput.bytes_per_second,
            perf_latency.avg_latency_ms,
            perf_latency.min_latency_ms,
            perf_latency.max_latency_ms,
            perf_latency.p95_latency_ms,
            perf_latency.p99_latency_ms,
            perf_resources.cpu_usage * 100.0,
            perf_resources.memory_usage_bytes as f64 / 1_048_576.0,
            perf_resources.disk_io_bps as f64 / 1_048_576.0,
            perf_history_count,
            (hit_ratio * 10.0).clamp(1.0, 10.0)
        )
    }
}