saorsa_core/dht/

rsps_integration.rs

//! RSPS Integration with DHT Storage
//!
//! This module integrates Root-Scoped Provider Summaries (RSPS) with the DHT storage layer,
//! enabling efficient content discovery and cache admission control.

use crate::dht::optimized_storage::OptimizedDHTStorage;
use crate::dht::{Key, Record};
use crate::error::{P2PError, P2pResult as Result, StorageError};
use crate::{Multiaddr, PeerId};
use saorsa_rsps::{
    CachePolicy, Cid, RootAnchoredCache, RootCid, Rsps, RspsConfig, TtlConfig, TtlEngine, TtlStats,
    WitnessKey, WitnessReceipt, witness::ReceiptMetadata,
};
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use std::time::{Duration, SystemTime};
use tokio::sync::RwLock;
use tracing::{debug, info, warn};

/// RSPS-enhanced DHT storage that uses provider summaries for efficient routing
pub struct RspsDhtStorage {
    /// Base DHT storage layer
    base_storage: Arc<OptimizedDHTStorage>,
    /// Root-anchored cache with RSPS admission control
    cache: Arc<RootAnchoredCache>,
    /// Map of root CIDs to their provider summaries
    provider_summaries: Arc<RwLock<HashMap<RootCid, ProviderRecord>>>,
    /// TTL management engine
    ttl_manager: Arc<TtlEngine>,
    /// Witness key for generating receipts
    witness_key: Arc<WitnessKey>,
    /// Local peer ID
    local_peer: PeerId,
    /// Configuration
    config: RspsDhtConfig,
}

/// Configuration for RSPS-DHT integration (wrapper around dht_rsps::RspsConfig)
#[derive(Debug, Clone)]
pub struct RspsDhtConfig {
    /// Maximum cache size in bytes
    pub max_cache_size: usize,
    /// Maximum items per root in cache
    pub max_items_per_root: usize,
    /// Base TTL for cached items
    pub base_ttl: Duration,
    /// Minimum receipts for TTL extension
    pub min_receipts_for_extension: usize,
    /// Maximum TTL multiplier
    pub max_ttl_multiplier: f64,
    /// Witness pseudonym refresh interval
    pub pseudonym_refresh_interval: Duration,
    /// Provider summary update interval
    pub summary_update_interval: Duration,
}

impl Default for RspsDhtConfig {
    fn default() -> Self {
        Self {
            max_cache_size: 100 * 1024 * 1024, // 100MB
            max_items_per_root: 1000,
            base_ttl: Duration::from_secs(3600), // 1 hour
            min_receipts_for_extension: 3,
            max_ttl_multiplier: 8.0,
            pseudonym_refresh_interval: Duration::from_secs(86400), // 24 hours
            summary_update_interval: Duration::from_secs(300),      // 5 minutes
        }
    }
}

impl From<RspsDhtConfig> for RspsConfig {
    fn from(_config: RspsDhtConfig) -> Self {
        // Use defaults for dht_rsps::RspsConfig since the structures don't match
        RspsConfig::default()
    }
}

/// Provider record containing RSPS and metadata
#[derive(Debug, Clone)]
pub struct ProviderRecord {
    /// The provider's peer ID
    pub provider: PeerId,
    /// Provider's network addresses
    pub addresses: Vec<Multiaddr>,
    /// Root-scoped provider summary
    pub rsps: Arc<Rsps>,
    /// Last update timestamp
    pub last_updated: SystemTime,
    /// Provider's witness receipts
    pub receipts: Vec<WitnessReceipt>,
}

impl RspsDhtStorage {
    /// Create a new RSPS-enhanced DHT storage
    pub async fn new(
        base_storage: Arc<OptimizedDHTStorage>,
        local_peer: PeerId,
        config: RspsDhtConfig,
    ) -> Result<Self> {
        // Initialize cache with policy
        let cache_policy = CachePolicy {
            max_size: config.max_cache_size,
            max_items_per_root: config.max_items_per_root,
            min_root_depth: 2,
            pledge_ratio: 1.5,
        };
        let cache = Arc::new(RootAnchoredCache::new(cache_policy));

        // Initialize TTL manager
        let ttl_config = TtlConfig {
            base_ttl: config.base_ttl,
            ttl_per_hit: Duration::from_secs(30 * 60), // 30 minutes per hit
            max_hit_ttl: Duration::from_secs(12 * 3600), // 12 hours max from hits
            ttl_per_receipt: Duration::from_secs(10 * 60), // 10 minutes per receipt
            max_receipt_ttl: Duration::from_secs(2 * 3600), // 2 hours max from receipts
            bucket_window: Duration::from_secs(5 * 60), // 5 minute buckets
        };
        let ttl_manager = Arc::new(TtlEngine::new(ttl_config));

        // Generate witness key
        let witness_key = Arc::new(WitnessKey::generate());

        Ok(Self {
            base_storage,
            cache,
            provider_summaries: Arc::new(RwLock::new(HashMap::new())),
            ttl_manager,
            witness_key,
            local_peer,
            config,
        })
    }

    /// Store a provider record with RSPS
    pub async fn store_provider(
        &self,
        root_cid: RootCid,
        provider: PeerId,
        addresses: Vec<Multiaddr>,
        rsps: Rsps,
    ) -> Result<()> {
        info!(
            "Storing provider record for root {:?} from peer {:?}",
            root_cid, provider
        );

        // Register RSPS with the cache for admission control
        self.cache.register_rsps(rsps.clone());

        // Create provider record
        let record = ProviderRecord {
            provider: provider.clone(),
            addresses,
            rsps: Arc::new(rsps),
            last_updated: SystemTime::now(),
            receipts: Vec::new(),
        };

        // Store in provider summaries
        let mut summaries = self.provider_summaries.write().await;
        summaries.insert(root_cid, record.clone());

        // Create DHT record for provider announcement
        let key = self.provider_key(&root_cid, &provider);
        let value = self.serialize_provider_record(&record)?;

        let dht_record = Record {
            key,
            value,
            publisher: {
                // Convert PeerId (String) to NodeId
                let peer_bytes = self.local_peer.as_bytes();
                let mut node_id_bytes = [0u8; 32];
                let len = peer_bytes.len().min(32);
                node_id_bytes[..len].copy_from_slice(&peer_bytes[..len]);
                crate::identity::node_identity::NodeId::from_bytes(node_id_bytes)
            },
            expires_at: SystemTime::now() + self.config.summary_update_interval,
            created_at: SystemTime::now(),
            signature: Some(Vec::new()),
        };

        // Store in base DHT
        self.base_storage.store(dht_record).await?;

        debug!("Provider record stored successfully");
        Ok(())
    }

    /// Find providers for a root CID
    pub async fn find_providers(&self, root_cid: &RootCid) -> Result<Vec<ProviderRecord>> {
        let summaries = self.provider_summaries.read().await;

        // Check local cache first
        if let Some(record) = summaries.get(root_cid) {
            debug!("Found provider in local cache");
            return Ok(vec![record.clone()]);
        }

        // Query DHT for providers
        let pattern = self.provider_key_pattern(root_cid);
        let records = self
            .base_storage
            .get_records_by_publisher(&pattern, None)
            .await
            .iter()
            .filter_map(|record| {
                let key_str = std::str::from_utf8(&record.key).ok()?;
                if key_str.starts_with(&pattern) {
                    Some(record.clone())
                } else {
                    None
                }
            })
            .collect::<Vec<_>>();

        let mut providers = Vec::new();
        for record in records {
            if let Ok(provider_record) = self.deserialize_provider_record(&record.value) {
                providers.push(provider_record);
            }
        }

        info!(
            "Found {} providers for root {:?}",
            providers.len(),
            root_cid
        );
        Ok(providers)
    }

    /// Cache a CID if it's in the RSPS for the root
    pub async fn cache_if_allowed(
        &self,
        root_cid: RootCid,
        cid: Cid,
        data: Vec<u8>,
    ) -> Result<bool> {
        // Check if we have RSPS for this root
        let summaries = self.provider_summaries.read().await;
        let _provider_record =
            summaries
                .get(&root_cid)
                .ok_or(P2PError::Storage(StorageError::Database(
                    std::borrow::Cow::Borrowed("No RSPS for root"),
                )))?;

        // Use cache admission control
        let admitted = self.cache.admit(root_cid, cid, data.clone()).map_err(|e| {
            P2PError::Storage(StorageError::Database(std::borrow::Cow::Owned(format!(
                "Cache admission failed: {}",
                e
            ))))
        })?;

        if admitted {
            // Record in TTL manager
            let ttl = self.ttl_manager.record_hit(&cid).map_err(|e| {
                P2PError::Storage(StorageError::Database(std::borrow::Cow::Owned(format!(
                    "TTL record failed: {}",
                    e
                ))))
            })?;

            info!("Cached CID {:?} with TTL {:?}", cid, ttl);
        } else {
            debug!("CID {:?} not admitted to cache", cid);
        }

        Ok(admitted)
    }

    /// Generate a witness receipt for retrieved content
    pub async fn generate_receipt(&self, cid: &Cid) -> Result<WitnessReceipt> {
        // Get current epoch (simplified)
        let epoch = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_secs();

        // Create receipt metadata
        let metadata = ReceiptMetadata {
            latency_ms: 0,
            content_size: 0,
            valid: true,
            error: None,
        };

        // Create receipt using proper API
        let receipt = self.witness_key.create_receipt(*cid, epoch, metadata);

        // Record receipt in TTL manager for extension logic
        let witness_id = self.witness_key.public_key();
        self.ttl_manager
            .record_receipt(cid, witness_id)
            .map_err(|e| {
                P2PError::Storage(StorageError::Database(std::borrow::Cow::Owned(format!(
                    "Failed to record receipt: {}",
                    e
                ))))
            })?;

        receipt.map_err(|e| {
            P2PError::Storage(StorageError::Database(std::borrow::Cow::Owned(format!(
                "RSPS receipt creation failed: {}",
                e
            ))))
        })
    }

    /// Batch generate receipts for multiple CIDs
    pub async fn generate_receipt_batch(&self, cids: &[Cid]) -> Result<Vec<WitnessReceipt>> {
        let mut batch = Vec::new();

        for cid in cids {
            let receipt = self.generate_receipt(cid).await?;
            batch.push(receipt);
        }

        Ok(batch)
    }

    /// Verify a witness receipt
    pub async fn verify_receipt(&self, receipt: &WitnessReceipt) -> Result<bool> {
        // In production, this would verify against known witness keys
        // For now, we just check the signature format
        Ok(!receipt.signature.is_empty())
    }

    /// Update RSPS for a root based on new content
    pub async fn update_rsps(&self, root_cid: &RootCid, new_cids: Vec<Cid>) -> Result<()> {
        let mut summaries = self.provider_summaries.write().await;

        if let Some(record) = summaries.get_mut(root_cid) {
            // Create new RSPS with updated CIDs
            let mut all_cids = HashSet::new();

            // Get existing CIDs from RSPS
            // Note: This requires iterating through possible CIDs to check membership
            // In production, we'd maintain a separate index

            // Add new CIDs
            for cid in new_cids {
                all_cids.insert(cid);
            }

            // Create updated RSPS
            let cid_vec: Vec<Cid> = all_cids.into_iter().collect();
            let new_rsps =
                Rsps::new(*root_cid, 1, &cid_vec, &RspsConfig::default()).map_err(|e| {
                    P2PError::Storage(StorageError::Database(std::borrow::Cow::Owned(format!(
                        "RSPS creation failed: {}",
                        e
                    ))))
                })?;

            // Update record
            record.rsps = Arc::new(new_rsps);
            record.last_updated = SystemTime::now();

            info!("Updated RSPS for root {:?}", root_cid);
        } else {
            warn!("No existing RSPS for root {:?}", root_cid);
        }

        Ok(())
    }

    /// Clean up expired entries
    pub async fn cleanup_expired(&self) -> Result<()> {
        // Note: RootAnchoredCache doesn't have cleanup_expired method
        // Cache cleanup happens automatically during eviction
        debug!("Cache cleanup managed automatically");

        // Clean up TTL manager
        let expired_cids = self.ttl_manager.cleanup_expired();
        debug!("Removed {} expired TTL entries", expired_cids.len());

        // Clean up old provider summaries
        let mut summaries = self.provider_summaries.write().await;
        let now = SystemTime::now();
        let expired_roots: Vec<RootCid> = summaries
            .iter()
            .filter(|(_, record)| {
                now.duration_since(record.last_updated)
                    .unwrap_or(Duration::ZERO)
                    > self.config.summary_update_interval * 2
            })
            .map(|(root, _)| *root)
            .collect();

        for root in expired_roots {
            summaries.remove(&root);
            debug!("Removed expired provider summary for root {:?}", root);
        }

        Ok(())
    }

    /// Get cache statistics
    pub async fn get_cache_stats(&self) -> CacheStats {
        CacheStats {
            total_cached_items: self.cache.stats().total_items,
            total_cache_size: self.cache.stats().total_size,
            roots_tracked: self.provider_summaries.read().await.len(),
            ttl_stats: TtlStats {
                hit_count: 0,
                receipt_count: 0,
                active_buckets: 0,
                remaining_ttl: std::time::Duration::ZERO,
                total_ttl: std::time::Duration::ZERO,
            },
        }
    }

    // Helper methods

    fn provider_key(&self, root_cid: &RootCid, provider: &PeerId) -> Key {
        let key_str = format!("/rsps/provider/{}/{}", hex::encode(root_cid), provider);
        let hash = blake3::hash(key_str.as_bytes());
        *hash.as_bytes()
    }

    fn provider_key_pattern(&self, root_cid: &RootCid) -> String {
        format!("/rsps/provider/{}/", hex::encode(root_cid))
    }

    fn serialize_provider_record(&self, record: &ProviderRecord) -> Result<Vec<u8>> {
        // In production, use proper serialization (e.g., protobuf)
        Ok(format!("{:?}", record).into_bytes())
    }

    fn deserialize_provider_record(&self, _data: &[u8]) -> Result<ProviderRecord> {
        // In production, use proper deserialization
        Err(P2PError::Serialization("Not implemented".into()))
    }
}

/// Cache statistics
#[derive(Debug, Clone)]
pub struct CacheStats {
    pub total_cached_items: usize,
    pub total_cache_size: usize,
    pub roots_tracked: usize,
    pub ttl_stats: saorsa_rsps::TtlStats,
}

#[cfg(test)]
mod tests {

    #[tokio::test]
    async fn test_rsps_integration() {
        // Test will be implemented
    }
}