ant_core/bootstrap/

merge.rs

//! Multi-Instance Cache Merge Coordination
//!
//! Handles conflict resolution and merging when multiple P2P Foundation instances
//! are running locally and sharing the same bootstrap cache.

use crate::{PeerId, Result, P2PError};
use crate::bootstrap::{ContactEntry, BootstrapCache};
use std::collections::HashMap;
use std::path::PathBuf;
use std::time::SystemTime;
use serde::{Deserialize, Serialize};
use tracing::{debug, info, warn};

/// Merge coordinator for handling multi-instance cache coordination
#[derive(Clone)]
pub struct MergeCoordinator {
    /// Main cache directory path
    cache_dir: PathBuf,
    /// Directory containing instance-specific cache files
    instance_cache_dir: PathBuf,
    /// Strategy used for resolving merge conflicts
    merge_strategy: MergeStrategy,
}

/// Strategy for resolving conflicts during merge operations
#[derive(Debug, Clone)]
pub enum MergeStrategy {
    /// Use quality score to determine the best contact
    QualityBased,
    /// Use most recent timestamp
    TimestampBased,
    /// Combine metrics from both contacts
    MetricsCombined,
    /// Use success rate as primary factor
    SuccessRateBased,
}

/// Instance cache data structure
#[derive(Debug, Serialize, Deserialize)]
struct InstanceCacheData {
    instance_id: String,
    timestamp: chrono::DateTime<chrono::Utc>,
    process_id: u32,
    contacts: HashMap<PeerId, ContactEntry>,
    version: u32,
}

/// Merge operation result
#[derive(Debug)]
pub struct MergeResult {
    /// Number of contacts that were merged from other instances
    pub contacts_merged: usize,
    /// Number of existing contacts that were updated
    pub contacts_updated: usize,
    /// Number of new contacts that were added
    pub contacts_added: usize,
    /// Number of conflicts that were resolved during merge
    pub conflicts_resolved: usize,
    /// Number of instance cache files that were processed
    pub instances_processed: usize,
    /// Total time taken for the merge operation in milliseconds
    pub merge_duration_ms: u64,
}

/// Conflict resolution information
#[derive(Debug)]
struct ConflictInfo {
    peer_id: PeerId,
    main_contact: ContactEntry,
    instance_contact: ContactEntry,
    resolution_strategy: MergeStrategy,
}

impl MergeCoordinator {
    /// Create a new merge coordinator
    pub fn new(cache_dir: PathBuf) -> Result<Self> {
        let instance_cache_dir = cache_dir.join("instance_caches");
        
        // Ensure instance cache directory exists
        std::fs::create_dir_all(&instance_cache_dir)
            .map_err(|e| P2PError::Bootstrap(format!("Failed to create instance cache directory: {}", e)))?;
        
        Ok(Self {
            cache_dir,
            instance_cache_dir,
            merge_strategy: MergeStrategy::QualityBased,
        })
    }
    
    /// Create coordinator with custom merge strategy
    pub fn with_strategy(cache_dir: PathBuf, strategy: MergeStrategy) -> Result<Self> {
        let mut coordinator = Self::new(cache_dir)?;
        coordinator.merge_strategy = strategy;
        Ok(coordinator)
    }
    
    /// Merge all instance caches into the main cache
    pub async fn merge_instance_caches(&self, main_cache: &BootstrapCache) -> Result<MergeResult> {
        let merge_start = SystemTime::now();
        
        debug!("Starting merge of instance caches");
        
        // Discover all instance cache files
        let instance_files = self.discover_instance_caches()?;
        
        if instance_files.is_empty() {
            debug!("No instance caches found to merge");
            return Ok(MergeResult::empty());
        }
        
        // Load all instance caches
        let instance_caches = self.load_instance_caches(instance_files).await?;
        
        // Perform merge operation
        let merge_result = self.perform_merge(main_cache, instance_caches).await?;
        
        // Cleanup old instance caches
        self.cleanup_processed_caches().await?;
        
        let merge_duration = merge_start.elapsed()
            .unwrap_or_default()
            .as_millis() as u64;
        
        info!("Merge completed: {} contacts processed, {} conflicts resolved in {}ms",
              merge_result.contacts_merged, merge_result.conflicts_resolved, merge_duration);
        
        Ok(MergeResult {
            merge_duration_ms: merge_duration,
            ..merge_result
        })
    }
    
    /// Discover all instance cache files
    fn discover_instance_caches(&self) -> Result<Vec<PathBuf>> {
        let mut cache_files = Vec::new();
        
        if !self.instance_cache_dir.exists() {
            return Ok(cache_files);
        }
        
        let entries = std::fs::read_dir(&self.instance_cache_dir)
            .map_err(|e| P2PError::Bootstrap(format!("Failed to read instance cache directory: {}", e)))?;
        
        for entry in entries {
            let entry = entry?;
            let path = entry.path();
            
            if path.is_file() && path.extension().and_then(|s| s.to_str()) == Some("cache") {
                // Check if the process is still running
                if let Some(process_id) = self.extract_process_id(&path) {
                    if self.is_process_running(process_id) {
                        cache_files.push(path);
                    } else {
                        // Process is dead, we can safely include this cache
                        cache_files.push(path);
                    }
                }
            }
        }
        
        debug!("Discovered {} instance cache files", cache_files.len());
        
        Ok(cache_files)
    }
    
    /// Load instance caches from files
    async fn load_instance_caches(&self, cache_files: Vec<PathBuf>) -> Result<Vec<InstanceCacheData>> {
        let mut instance_caches = Vec::new();
        
        for cache_file in cache_files {
            match self.load_instance_cache(&cache_file).await {
                Ok(cache_data) => {
                    if self.validate_instance_cache(&cache_data) {
                        instance_caches.push(cache_data);
                    } else {
                        warn!("Invalid instance cache found: {:?}", cache_file);
                    }
                }
                Err(e) => {
                    warn!("Failed to load instance cache {:?}: {}", cache_file, e);
                }
            }
        }
        
        debug!("Loaded {} valid instance caches", instance_caches.len());
        
        Ok(instance_caches)
    }
    
    /// Load a single instance cache
    async fn load_instance_cache(&self, cache_file: &PathBuf) -> Result<InstanceCacheData> {
        let json_data = std::fs::read_to_string(cache_file)
            .map_err(|e| P2PError::Bootstrap(format!("Failed to read instance cache: {}", e)))?;
        
        let cache_data: InstanceCacheData = serde_json::from_str(&json_data)
            .map_err(|e| P2PError::Bootstrap(format!("Failed to parse instance cache: {}", e)))?;
        
        Ok(cache_data)
    }
    
    /// Validate instance cache data
    fn validate_instance_cache(&self, cache_data: &InstanceCacheData) -> bool {
        // Check version compatibility
        if cache_data.version != 1 {
            return false;
        }
        
        // Check timestamp is reasonable (not too old)
        let now = chrono::Utc::now();
        let age = now.signed_duration_since(cache_data.timestamp);
        
        if age.num_hours() > 24 {
            debug!("Instance cache too old: {} hours", age.num_hours());
            return false;
        }
        
        true
    }
    
    /// Perform the actual merge operation
    async fn perform_merge(&self, main_cache: &BootstrapCache, instance_caches: Vec<InstanceCacheData>) -> Result<MergeResult> {
        let mut result = MergeResult::empty();
        result.instances_processed = instance_caches.len();
        
        // Get current main cache contacts
        let mut merged_contacts = main_cache.get_all_contacts().await;
        
        // Process each instance cache
        for instance_cache in instance_caches {
            let instance_result = self.merge_single_instance(&mut merged_contacts, instance_cache).await?;
            result.combine(instance_result);
        }
        
        // Update main cache with merged contacts
        main_cache.set_all_contacts(merged_contacts).await;
        
        // Save updated cache
        main_cache.save_to_disk().await?;
        
        Ok(result)
    }
    
    /// Merge a single instance cache
    async fn merge_single_instance(&self, main_contacts: &mut HashMap<PeerId, ContactEntry>, instance_cache: InstanceCacheData) -> Result<MergeResult> {
        let mut result = MergeResult::empty();
        
        for (peer_id, instance_contact) in instance_cache.contacts {
            match main_contacts.get(&peer_id) {
                Some(main_contact) => {
                    // Contact exists in main cache, resolve conflict
                    let resolved_contact = self.resolve_conflict(main_contact, &instance_contact)?;
                    
                    if resolved_contact.quality_metrics.quality_score != main_contact.quality_metrics.quality_score {
                        result.contacts_updated += 1;
                        result.conflicts_resolved += 1;
                    }
                    
                    main_contacts.insert(peer_id, resolved_contact);
                }
                None => {
                    // New contact, add to main cache
                    main_contacts.insert(peer_id, instance_contact);
                    result.contacts_added += 1;
                }
            }
            
            result.contacts_merged += 1;
        }
        
        Ok(result)
    }
    
    /// Resolve conflict between two contact entries
    fn resolve_conflict(&self, main_contact: &ContactEntry, instance_contact: &ContactEntry) -> Result<ContactEntry> {
        match self.merge_strategy {
            MergeStrategy::QualityBased => {
                if instance_contact.quality_metrics.quality_score > main_contact.quality_metrics.quality_score {
                    Ok(instance_contact.clone())
                } else {
                    Ok(main_contact.clone())
                }
            }
            
            MergeStrategy::TimestampBased => {
                if instance_contact.last_seen > main_contact.last_seen {
                    Ok(instance_contact.clone())
                } else {
                    Ok(main_contact.clone())
                }
            }
            
            MergeStrategy::MetricsCombined => {
                self.combine_contact_metrics(main_contact, instance_contact)
            }
            
            MergeStrategy::SuccessRateBased => {
                if instance_contact.quality_metrics.success_rate > main_contact.quality_metrics.success_rate {
                    Ok(instance_contact.clone())
                } else {
                    Ok(main_contact.clone())
                }
            }
        }
    }
    
    /// Combine metrics from two contacts
    fn combine_contact_metrics(&self, main_contact: &ContactEntry, instance_contact: &ContactEntry) -> Result<ContactEntry> {
        let mut combined_contact = main_contact.clone();
        
        // Use the most recent timestamp
        if instance_contact.last_seen > main_contact.last_seen {
            combined_contact.last_seen = instance_contact.last_seen;
        }
        
        // Combine connection history
        combined_contact.connection_history.total_attempts += 
            instance_contact.connection_history.total_attempts;
        combined_contact.connection_history.successful_connections += 
            instance_contact.connection_history.successful_connections;
        combined_contact.connection_history.failed_connections += 
            instance_contact.connection_history.failed_connections;
        
        // Update addresses (union of both sets)
        for addr in &instance_contact.addresses {
            if !combined_contact.addresses.contains(addr) {
                combined_contact.addresses.push(addr.clone());
            }
        }
        
        // Update capabilities (union of both sets)
        for capability in &instance_contact.capabilities {
            if !combined_contact.capabilities.contains(capability) {
                combined_contact.capabilities.push(capability.clone());
            }
        }
        
        // Use higher reputation score
        if instance_contact.reputation_score > combined_contact.reputation_score {
            combined_contact.reputation_score = instance_contact.reputation_score;
        }
        
        // Use verified status if either is verified
        combined_contact.ipv6_identity_verified = 
            combined_contact.ipv6_identity_verified || instance_contact.ipv6_identity_verified;
        
        // Recalculate quality metrics
        combined_contact.update_success_rate();
        combined_contact.recalculate_quality_score();
        
        Ok(combined_contact)
    }
    
    /// Cleanup processed instance caches
    async fn cleanup_processed_caches(&self) -> Result<()> {
        let cache_files = self.discover_instance_caches()?;
        let mut cleaned_count = 0;
        
        for cache_file in cache_files {
            // Check if process is still running
            if let Some(process_id) = self.extract_process_id(&cache_file) {
                if !self.is_process_running(process_id) {
                    // Process is dead, safe to remove cache
                    if let Err(e) = std::fs::remove_file(&cache_file) {
                        warn!("Failed to remove old instance cache {:?}: {}", cache_file, e);
                    } else {
                        cleaned_count += 1;
                    }
                }
            }
        }
        
        if cleaned_count > 0 {
            debug!("Cleaned up {} old instance cache files", cleaned_count);
        }
        
        Ok(())
    }
    
    /// Extract process ID from cache file name
    fn extract_process_id(&self, cache_file: &PathBuf) -> Option<u32> {
        cache_file
            .file_stem()
            .and_then(|name| name.to_str())
            .and_then(|name| {
                let parts: Vec<&str> = name.split('_').collect();
                if parts.len() >= 2 {
                    parts[0].parse().ok()
                } else {
                    None
                }
            })
    }
    
    /// Check if a process is still running
    fn is_process_running(&self, process_id: u32) -> bool {
        #[cfg(unix)]
        {
            use std::process::Command;
            Command::new("kill")
                .args(&["-0", &process_id.to_string()])
                .output()
                .map(|output| output.status.success())
                .unwrap_or(false)
        }
        
        #[cfg(windows)]
        {
            use std::process::Command;
            Command::new("tasklist")
                .args(&["/FI", &format!("PID eq {}", process_id)])
                .output()
                .map(|output| {
                    String::from_utf8_lossy(&output.stdout)
                        .contains(&process_id.to_string())
                })
                .unwrap_or(false)
        }
        
        #[cfg(not(any(unix, windows)))]
        {
            // Assume process is still running if we can't check
            true
        }
    }
    
    /// Get merge strategy
    pub fn get_strategy(&self) -> &MergeStrategy {
        &self.merge_strategy
    }
    
    /// Set merge strategy
    pub fn set_strategy(&mut self, strategy: MergeStrategy) {
        self.merge_strategy = strategy;
    }
}

impl MergeResult {
    /// Create empty merge result
    fn empty() -> Self {
        Self {
            contacts_merged: 0,
            contacts_updated: 0,
            contacts_added: 0,
            conflicts_resolved: 0,
            instances_processed: 0,
            merge_duration_ms: 0,
        }
    }
    
    /// Combine with another merge result
    fn combine(&mut self, other: MergeResult) {
        self.contacts_merged += other.contacts_merged;
        self.contacts_updated += other.contacts_updated;
        self.contacts_added += other.contacts_added;
        self.conflicts_resolved += other.conflicts_resolved;
    }
}

impl std::fmt::Display for MergeResult {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "MergeResult {{ merged: {}, updated: {}, added: {}, conflicts: {}, instances: {}, duration: {}ms }}",
            self.contacts_merged,
            self.contacts_updated,
            self.contacts_added,
            self.conflicts_resolved,
            self.instances_processed,
            self.merge_duration_ms
        )
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use tempfile::TempDir;

    #[tokio::test]
    async fn test_merge_coordinator_creation() {
        let temp_dir = TempDir::new().unwrap();
        let coordinator = MergeCoordinator::new(temp_dir.path().to_path_buf());
        assert!(coordinator.is_ok());
    }
    
    #[tokio::test]
    async fn test_conflict_resolution_quality_based() {
        let temp_dir = TempDir::new().unwrap();
        let coordinator = MergeCoordinator::with_strategy(
            temp_dir.path().to_path_buf(),
            MergeStrategy::QualityBased
        ).unwrap();
        
        let mut main_contact = ContactEntry::new(
            PeerId::from("test-peer"),
            vec!["/ip4/127.0.0.1/tcp/9000".to_string()]
        );
        main_contact.quality_metrics.quality_score = 0.5;
        
        let mut instance_contact = ContactEntry::new(
            PeerId::from("test-peer"),
            vec!["/ip4/127.0.0.1/tcp/9001".to_string()]
        );
        instance_contact.quality_metrics.quality_score = 0.8;
        
        let resolved = coordinator.resolve_conflict(&main_contact, &instance_contact).unwrap();
        assert_eq!(resolved.quality_metrics.quality_score, 0.8);
    }
    
    #[tokio::test]
    async fn test_metrics_combination() {
        let temp_dir = TempDir::new().unwrap();
        let coordinator = MergeCoordinator::with_strategy(
            temp_dir.path().to_path_buf(),
            MergeStrategy::MetricsCombined
        ).unwrap();
        
        let mut main_contact = ContactEntry::new(
            PeerId::from("test-peer"),
            vec!["/ip4/127.0.0.1/tcp/9000".to_string()]
        );
        main_contact.connection_history.total_attempts = 10;
        main_contact.connection_history.successful_connections = 8;
        
        let mut instance_contact = ContactEntry::new(
            PeerId::from("test-peer"),
            vec!["/ip4/127.0.0.1/tcp/9001".to_string()]
        );
        instance_contact.connection_history.total_attempts = 5;
        instance_contact.connection_history.successful_connections = 4;
        
        let combined = coordinator.combine_contact_metrics(&main_contact, &instance_contact).unwrap();
        assert_eq!(combined.connection_history.total_attempts, 15);
        assert_eq!(combined.connection_history.successful_connections, 12);
    }
    
    #[test]
    fn test_process_id_extraction() {
        let temp_dir = TempDir::new().unwrap();
        let coordinator = MergeCoordinator::new(temp_dir.path().to_path_buf()).unwrap();
        
        let cache_file = PathBuf::from("12345_1234567890.cache");
        let process_id = coordinator.extract_process_id(&cache_file);
        assert_eq!(process_id, Some(12345));
    }
}