oxirs_stream/

multi_region_replication.rs

//! # Multi-Region Replication
//!
//! Advanced multi-region replication system for OxiRS Stream providing global data consistency,
//! failover capabilities, and optimized cross-region communication.

use crate::StreamEvent;
use anyhow::{anyhow, Result};
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, HashSet, VecDeque};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::{Mutex, RwLock, Semaphore};
use tokio::time;
use tracing::{error, info, warn};
use uuid::Uuid;

/// Region configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RegionConfig {
    /// Region identifier
    pub region_id: String,
    /// Region name (human-readable)
    pub region_name: String,
    /// Geographic location
    pub location: GeographicLocation,
    /// Network endpoints for this region
    pub endpoints: Vec<RegionEndpoint>,
    /// Replication priority (higher is more preferred)
    pub priority: u8,
    /// Whether this region is active for writes
    pub is_write_active: bool,
    /// Whether this region is active for reads
    pub is_read_active: bool,
    /// Replication mode for this region
    pub replication_mode: ReplicationMode,
    /// Network latency to other regions (ms)
    pub latency_map: HashMap<String, u64>,
}

/// Geographic location information
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct GeographicLocation {
    /// Country code (ISO 3166-1 alpha-2)
    pub country: String,
    /// Region/state/province
    pub region: String,
    /// City
    pub city: String,
    /// Latitude
    pub latitude: f64,
    /// Longitude
    pub longitude: f64,
    /// Availability zone (if applicable)
    pub availability_zone: Option<String>,
}

/// Region endpoint configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RegionEndpoint {
    /// Endpoint URL
    pub url: String,
    /// Endpoint type
    pub endpoint_type: EndpointType,
    /// Whether this endpoint is currently healthy
    pub is_healthy: bool,
    /// Last health check timestamp
    pub last_health_check: Option<DateTime<Utc>>,
    /// Authentication configuration
    pub auth: Option<EndpointAuth>,
}

/// Endpoint type
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum EndpointType {
    /// Primary streaming endpoint
    Primary,
    /// Secondary/backup endpoint
    Secondary,
    /// Administrative endpoint
    Admin,
    /// Health check endpoint
    HealthCheck,
}

/// Endpoint authentication
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EndpointAuth {
    /// Authentication type
    pub auth_type: String,
    /// Credentials (encrypted)
    pub credentials: HashMap<String, String>,
}

/// Replication mode
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum ReplicationMode {
    /// Synchronous replication (wait for all regions)
    Synchronous,
    /// Asynchronous replication (fire and forget)
    Asynchronous,
    /// Semi-synchronous (wait for majority)
    SemiSynchronous { min_replicas: usize },
    /// Leader-follower (one primary region)
    LeaderFollower { leader_region: String },
    /// Active-active (all regions can write)
    ActiveActive,
}

/// Replication configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ReplicationConfig {
    /// Replication strategy
    pub strategy: ReplicationStrategy,
    /// Conflict resolution strategy
    pub conflict_resolution: ConflictResolution,
    /// Maximum replication lag tolerance
    pub max_lag_ms: u64,
    /// Replication timeout
    pub replication_timeout: Duration,
    /// Enable compression for cross-region traffic
    pub enable_compression: bool,
    /// Batch size for replication
    pub batch_size: usize,
    /// Health check interval
    pub health_check_interval: Duration,
    /// Failover timeout
    pub failover_timeout: Duration,
}

/// Replication strategy
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum ReplicationStrategy {
    /// Replicate all events to all regions
    FullReplication,
    /// Replicate only specific event types
    SelectiveReplication { event_types: HashSet<String> },
    /// Partition-based replication
    PartitionBased {
        partition_strategy: PartitionStrategy,
    },
    /// Geography-based replication
    GeographyBased {
        region_groups: HashMap<String, Vec<String>>,
    },
}

/// Partition strategy for replication
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum PartitionStrategy {
    /// Hash-based partitioning
    Hash { hash_key: String },
    /// Range-based partitioning
    Range { ranges: Vec<PartitionRange> },
    /// Custom partitioning logic
    Custom { strategy_name: String },
}

/// Partition range definition
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PartitionRange {
    pub start: String,
    pub end: String,
    pub regions: Vec<String>,
}

/// Conflict resolution strategy
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum ConflictResolution {
    /// Last write wins (timestamp-based)
    LastWriteWins,
    /// First write wins
    FirstWriteWins,
    /// Region priority based
    RegionPriority { priority_order: Vec<String> },
    /// Custom conflict resolution
    Custom { resolver_name: String },
    /// Manual resolution (queue conflicts)
    Manual,
}

/// Replicated event with replication metadata
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ReplicatedEvent {
    /// Original event
    pub event: StreamEvent,
    /// Replication metadata
    pub replication_metadata: ReplicationMetadata,
}

/// Replication metadata
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ReplicationMetadata {
    /// Unique replication ID
    pub replication_id: Uuid,
    /// Source region
    pub source_region: String,
    /// Target regions
    pub target_regions: Vec<String>,
    /// Replication timestamp
    pub replication_timestamp: DateTime<Utc>,
    /// Replication status per region
    pub region_status: HashMap<String, ReplicationStatus>,
    /// Vector clock for ordering
    pub vector_clock: VectorClock,
    /// Conflict resolution information
    pub conflict_info: Option<ConflictInfo>,
}

/// Replication status for a region
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum ReplicationStatus {
    /// Pending replication
    Pending,
    /// Successfully replicated
    Success { timestamp: DateTime<Utc> },
    /// Replication failed
    Failed {
        error: String,
        timestamp: DateTime<Utc>,
    },
    /// Replication in progress
    InProgress { started_at: DateTime<Utc> },
}

/// Vector clock for event ordering
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct VectorClock {
    /// Clock values per region
    pub clocks: HashMap<String, u64>,
}

impl Default for VectorClock {
    fn default() -> Self {
        Self::new()
    }
}

impl VectorClock {
    /// Create a new vector clock
    pub fn new() -> Self {
        Self {
            clocks: HashMap::new(),
        }
    }

    /// Increment clock for a region
    pub fn increment(&mut self, region: &str) {
        let current = self.clocks.get(region).unwrap_or(&0);
        self.clocks.insert(region.to_string(), current + 1);
    }

    /// Update clock with another vector clock
    pub fn update(&mut self, other: &VectorClock) {
        for (region, other_clock) in &other.clocks {
            let current = self.clocks.get(region).unwrap_or(&0);
            self.clocks
                .insert(region.clone(), (*current).max(*other_clock));
        }
    }

    /// Check if this clock happens before another
    pub fn happens_before(&self, other: &VectorClock) -> bool {
        let mut strictly_less = false;

        for region in self.clocks.keys().chain(other.clocks.keys()) {
            let self_clock = self.clocks.get(region).unwrap_or(&0);
            let other_clock = other.clocks.get(region).unwrap_or(&0);

            if self_clock > other_clock {
                return false; // Not happens-before
            } else if self_clock < other_clock {
                strictly_less = true;
            }
        }

        strictly_less
    }

    /// Check if clocks are concurrent (neither happens before the other)
    pub fn is_concurrent(&self, other: &VectorClock) -> bool {
        !self.happens_before(other) && !other.happens_before(self)
    }
}

/// Conflict information for resolution
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ConflictInfo {
    /// Conflict type
    pub conflict_type: ConflictType,
    /// Conflicting events
    pub conflicting_events: Vec<StreamEvent>,
    /// Resolution strategy used
    pub resolution_strategy: ConflictResolution,
    /// Resolution timestamp
    pub resolved_at: Option<DateTime<Utc>>,
    /// Resolution result
    pub resolution_result: Option<StreamEvent>,
}

/// Type of conflict detected
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum ConflictType {
    /// Write-write conflict (concurrent writes to same resource)
    WriteWrite,
    /// Write-read conflict
    WriteRead,
    /// Schema conflict
    Schema,
    /// Ordering conflict
    Ordering,
}

/// Multi-region replication manager
pub struct MultiRegionReplicationManager {
    /// Replication configuration
    config: ReplicationConfig,
    /// Configured regions
    regions: Arc<RwLock<HashMap<String, RegionConfig>>>,
    /// Current region ID
    current_region: String,
    /// Replication statistics
    stats: Arc<ReplicationStats>,
    /// Conflict resolution queue
    conflict_queue: Arc<Mutex<VecDeque<ConflictInfo>>>,
    /// Vector clock for this region
    vector_clock: Arc<Mutex<VectorClock>>,
    /// Health monitoring
    health_monitor: Arc<RegionHealthMonitor>,
    /// Replication semaphore
    replication_semaphore: Arc<Semaphore>,
}

/// Replication statistics
#[derive(Debug, Default)]
pub struct ReplicationStats {
    pub total_events_replicated: AtomicU64,
    pub successful_replications: AtomicU64,
    pub failed_replications: AtomicU64,
    pub conflicts_detected: AtomicU64,
    pub conflicts_resolved: AtomicU64,
    pub average_replication_latency_ms: AtomicU64,
    pub cross_region_bandwidth_bytes: AtomicU64,
    pub region_failures: AtomicU64,
    pub failover_events: AtomicU64,
}

/// Region health monitor
pub struct RegionHealthMonitor {
    /// Health status per region
    health_status: Arc<RwLock<HashMap<String, RegionHealth>>>,
    /// Health check interval
    check_interval: Duration,
    /// Statistics
    stats: Arc<HealthStats>,
}

/// Health status for a region
#[derive(Debug, Clone)]
pub struct RegionHealth {
    /// Whether region is healthy
    pub is_healthy: bool,
    /// Last successful health check
    pub last_success: Option<DateTime<Utc>>,
    /// Last health check attempt
    pub last_check: DateTime<Utc>,
    /// Current latency to region
    pub latency_ms: Option<u64>,
    /// Error count in recent window
    pub recent_errors: u32,
    /// Health score (0.0 to 1.0)
    pub health_score: f64,
}

/// Health monitoring statistics
#[derive(Debug, Default)]
pub struct HealthStats {
    pub total_health_checks: AtomicU64,
    pub failed_health_checks: AtomicU64,
    pub average_latency_ms: AtomicU64,
    pub regions_down: AtomicU64,
}

impl MultiRegionReplicationManager {
    /// Create a new multi-region replication manager
    pub fn new(config: ReplicationConfig, current_region: String) -> Self {
        let health_monitor = Arc::new(RegionHealthMonitor::new(config.health_check_interval));

        Self {
            config,
            current_region,
            regions: Arc::new(RwLock::new(HashMap::new())),
            stats: Arc::new(ReplicationStats::default()),
            conflict_queue: Arc::new(Mutex::new(VecDeque::new())),
            vector_clock: Arc::new(Mutex::new(VectorClock::new())),
            health_monitor,
            replication_semaphore: Arc::new(Semaphore::new(100)), // Max 100 concurrent replications
        }
    }

    /// Add a region to the replication topology
    pub async fn add_region(&self, region_config: RegionConfig) -> Result<()> {
        let region_id = region_config.region_id.clone();
        let mut regions = self.regions.write().await;
        regions.insert(region_id.clone(), region_config);

        // Initialize health monitoring for this region
        self.health_monitor.add_region(region_id.clone()).await;

        info!("Added region {} to replication topology", region_id);
        Ok(())
    }

    /// Remove a region from the replication topology
    pub async fn remove_region(&self, region_id: &str) -> Result<()> {
        let mut regions = self.regions.write().await;
        if regions.remove(region_id).is_some() {
            self.health_monitor.remove_region(region_id).await;
            info!("Removed region {} from replication topology", region_id);
            Ok(())
        } else {
            Err(anyhow!("Region {} not found", region_id))
        }
    }

    /// Replicate an event to other regions
    pub async fn replicate_event(&self, event: StreamEvent) -> Result<ReplicatedEvent> {
        let _permit = self.replication_semaphore.acquire().await?;
        let start_time = Instant::now();

        // Generate replication metadata
        let mut vector_clock = self.vector_clock.lock().await;
        vector_clock.increment(&self.current_region);
        let replication_metadata = ReplicationMetadata {
            replication_id: Uuid::new_v4(),
            source_region: self.current_region.clone(),
            target_regions: self.get_target_regions(&event).await?,
            replication_timestamp: Utc::now(),
            region_status: HashMap::new(),
            vector_clock: vector_clock.clone(),
            conflict_info: None,
        };
        drop(vector_clock);

        let replicated_event = ReplicatedEvent {
            event,
            replication_metadata,
        };

        // Perform replication based on strategy
        match self.config.strategy {
            ReplicationStrategy::FullReplication => {
                self.replicate_to_all_regions(&replicated_event).await?;
            }
            ReplicationStrategy::SelectiveReplication { ref event_types } => {
                if self.should_replicate_event(&replicated_event.event, event_types) {
                    self.replicate_to_all_regions(&replicated_event).await?;
                }
            }
            ReplicationStrategy::PartitionBased {
                ref partition_strategy,
            } => {
                self.replicate_partitioned(&replicated_event, partition_strategy)
                    .await?;
            }
            ReplicationStrategy::GeographyBased { ref region_groups } => {
                self.replicate_by_geography(&replicated_event, region_groups)
                    .await?;
            }
        }

        // Record statistics
        let replication_latency = start_time.elapsed();
        self.stats
            .total_events_replicated
            .fetch_add(1, Ordering::Relaxed);
        self.stats
            .average_replication_latency_ms
            .store(replication_latency.as_millis() as u64, Ordering::Relaxed);

        info!(
            "Replicated event {} to {} regions in {:?}",
            replicated_event.replication_metadata.replication_id,
            replicated_event.replication_metadata.target_regions.len(),
            replication_latency
        );

        Ok(replicated_event)
    }

    /// Handle an incoming replicated event
    pub async fn handle_replicated_event(&self, replicated_event: ReplicatedEvent) -> Result<()> {
        // Check for conflicts
        if let Some(conflict) = self.detect_conflict(&replicated_event).await? {
            self.handle_conflict(conflict).await?;
            return Ok(());
        }

        // Update vector clock
        let mut vector_clock = self.vector_clock.lock().await;
        vector_clock.update(&replicated_event.replication_metadata.vector_clock);
        drop(vector_clock);

        // Process the event locally
        self.process_replicated_event(replicated_event).await?;

        Ok(())
    }

    /// Detect conflicts in replicated events
    async fn detect_conflict(
        &self,
        replicated_event: &ReplicatedEvent,
    ) -> Result<Option<ConflictInfo>> {
        // Simple conflict detection based on vector clocks
        let vector_clock = self.vector_clock.lock().await;

        if vector_clock.is_concurrent(&replicated_event.replication_metadata.vector_clock) {
            // Potential conflict detected
            self.stats
                .conflicts_detected
                .fetch_add(1, Ordering::Relaxed);

            let conflict_info = ConflictInfo {
                conflict_type: ConflictType::WriteWrite,
                conflicting_events: vec![replicated_event.event.clone()],
                resolution_strategy: self.config.conflict_resolution.clone(),
                resolved_at: None,
                resolution_result: None,
            };

            warn!(
                "Conflict detected for event {}",
                replicated_event.replication_metadata.replication_id
            );
            return Ok(Some(conflict_info));
        }

        Ok(None)
    }

    /// Handle a detected conflict.
    ///
    /// Every strategy either produces a `resolution_result` or explicitly
    /// queues the conflict for manual resolution — a conflict must never be
    /// silently dropped on the floor.
    async fn handle_conflict(&self, mut conflict_info: ConflictInfo) -> Result<()> {
        match &self.config.conflict_resolution {
            ConflictResolution::LastWriteWins => {
                // Resolve by latest timestamp
                conflict_info.resolution_result = Some(
                    conflict_info
                        .conflicting_events
                        .iter()
                        .max_by_key(|e| e.metadata().timestamp)
                        .expect("conflicting_events should not be empty")
                        .clone(),
                );
                conflict_info.resolved_at = Some(Utc::now());
                self.stats
                    .conflicts_resolved
                    .fetch_add(1, Ordering::Relaxed);
            }
            ConflictResolution::FirstWriteWins => {
                // Resolve by earliest timestamp
                conflict_info.resolution_result = Some(
                    conflict_info
                        .conflicting_events
                        .iter()
                        .min_by_key(|e| e.metadata().timestamp)
                        .expect("conflicting_events should not be empty")
                        .clone(),
                );
                conflict_info.resolved_at = Some(Utc::now());
                self.stats
                    .conflicts_resolved
                    .fetch_add(1, Ordering::Relaxed);
            }
            ConflictResolution::RegionPriority { priority_order } => {
                // Pick the event whose originating region (carried in
                // `EventMetadata::source`) ranks earliest in `priority_order`.
                // Events from regions not listed are treated as lowest
                // priority; ties (including "not listed" ties) are broken by
                // latest timestamp (Last-Write-Wins as a secondary rule).
                let rank = |event: &StreamEvent| -> usize {
                    let source = &event.metadata().source;
                    priority_order
                        .iter()
                        .position(|region| region == source)
                        .unwrap_or(priority_order.len())
                };
                let resolved = conflict_info
                    .conflicting_events
                    .iter()
                    .min_by(|a, b| {
                        rank(a)
                            .cmp(&rank(b))
                            .then_with(|| b.metadata().timestamp.cmp(&a.metadata().timestamp))
                    })
                    .expect("conflicting_events should not be empty")
                    .clone();
                conflict_info.resolution_result = Some(resolved);
                conflict_info.resolved_at = Some(Utc::now());
                self.stats
                    .conflicts_resolved
                    .fetch_add(1, Ordering::Relaxed);
            }
            ConflictResolution::Custom { resolver_name } => {
                // No pluggable custom-resolver mechanism exists yet. Rather
                // than silently discarding the conflict, route it into the
                // same manual-resolution queue as `Manual` and warn loudly
                // so operators know a resolver they configured isn't wired up.
                warn!(
                    "Custom conflict resolver '{}' is not implemented; routing conflict {:?} to \
                     the manual resolution queue instead of discarding it",
                    resolver_name, conflict_info.conflict_type
                );
                let mut queue = self.conflict_queue.lock().await;
                queue.push_back(conflict_info);
            }
            ConflictResolution::Manual => {
                // Queue for manual resolution
                let mut queue = self.conflict_queue.lock().await;
                queue.push_back(conflict_info);
            }
        }

        Ok(())
    }

    /// Get target regions for an event based on strategy
    async fn get_target_regions(&self, _event: &StreamEvent) -> Result<Vec<String>> {
        let regions = self.regions.read().await;
        let healthy_regions = self.health_monitor.get_healthy_regions().await;

        Ok(regions
            .keys()
            .filter(|region_id| {
                **region_id != self.current_region && healthy_regions.contains(*region_id)
            })
            .cloned()
            .collect())
    }

    /// Replicate to all available regions
    async fn replicate_to_all_regions(&self, replicated_event: &ReplicatedEvent) -> Result<()> {
        let regions = self.regions.read().await;
        let mut replication_tasks = Vec::new();

        for region_id in &replicated_event.replication_metadata.target_regions {
            if let Some(region_config) = regions.get(region_id) {
                let event_clone = replicated_event.clone();
                let region_config_clone = region_config.clone();
                let region_id_clone = region_id.clone();
                let stats = self.stats.clone();

                let task = tokio::spawn(async move {
                    match Self::send_to_region(event_clone, region_config_clone).await {
                        Ok(_) => {
                            stats
                                .successful_replications
                                .fetch_add(1, Ordering::Relaxed);
                        }
                        Err(e) => {
                            stats.failed_replications.fetch_add(1, Ordering::Relaxed);
                            error!("Failed to replicate to region {}: {}", region_id_clone, e);
                        }
                    }
                });

                replication_tasks.push(task);
            }
        }

        // Wait for replication based on mode
        // Wait for all replications (can be optimized based on replication mode)
        for task in replication_tasks {
            let _ = task.await;
        }

        Ok(())
    }

    /// Send event to a specific region
    async fn send_to_region(
        _replicated_event: ReplicatedEvent,
        _region_config: RegionConfig,
    ) -> Result<()> {
        // Simulate network call - in real implementation, this would use HTTP/gRPC
        time::sleep(Duration::from_millis(50)).await;

        // Simulate occasional failures
        if fastrand::f32() < 0.05 {
            // 5% failure rate
            return Err(anyhow!("Simulated network failure"));
        }

        Ok(())
    }

    /// Check if an event should be replicated based on selective replication rules
    fn should_replicate_event(&self, event: &StreamEvent, event_types: &HashSet<String>) -> bool {
        let event_type = format!("{:?}", std::mem::discriminant(event));
        event_types.contains(&event_type)
    }

    /// Replicate using partition-based strategy
    async fn replicate_partitioned(
        &self,
        _replicated_event: &ReplicatedEvent,
        _partition_strategy: &PartitionStrategy,
    ) -> Result<()> {
        // Implementation for partition-based replication
        // This would determine which regions to replicate to based on partitioning
        Ok(())
    }

    /// Replicate using geography-based strategy
    async fn replicate_by_geography(
        &self,
        _replicated_event: &ReplicatedEvent,
        _region_groups: &HashMap<String, Vec<String>>,
    ) -> Result<()> {
        // Implementation for geography-based replication
        // This would replicate to regions in the same geographic group
        Ok(())
    }

    /// Process a replicated event locally
    async fn process_replicated_event(&self, _replicated_event: ReplicatedEvent) -> Result<()> {
        // Implementation for processing replicated events locally
        // This would typically integrate with the local storage system
        Ok(())
    }

    /// Get replication statistics
    pub fn get_stats(&self) -> ReplicationStats {
        ReplicationStats {
            total_events_replicated: AtomicU64::new(
                self.stats.total_events_replicated.load(Ordering::Relaxed),
            ),
            successful_replications: AtomicU64::new(
                self.stats.successful_replications.load(Ordering::Relaxed),
            ),
            failed_replications: AtomicU64::new(
                self.stats.failed_replications.load(Ordering::Relaxed),
            ),
            conflicts_detected: AtomicU64::new(
                self.stats.conflicts_detected.load(Ordering::Relaxed),
            ),
            conflicts_resolved: AtomicU64::new(
                self.stats.conflicts_resolved.load(Ordering::Relaxed),
            ),
            average_replication_latency_ms: AtomicU64::new(
                self.stats
                    .average_replication_latency_ms
                    .load(Ordering::Relaxed),
            ),
            cross_region_bandwidth_bytes: AtomicU64::new(
                self.stats
                    .cross_region_bandwidth_bytes
                    .load(Ordering::Relaxed),
            ),
            region_failures: AtomicU64::new(self.stats.region_failures.load(Ordering::Relaxed)),
            failover_events: AtomicU64::new(self.stats.failover_events.load(Ordering::Relaxed)),
        }
    }

    /// Get conflicts waiting for manual resolution
    pub async fn get_pending_conflicts(&self) -> Vec<ConflictInfo> {
        let queue = self.conflict_queue.lock().await;
        queue.iter().cloned().collect()
    }
}

impl RegionHealthMonitor {
    /// Create a new region health monitor
    pub fn new(check_interval: Duration) -> Self {
        Self {
            health_status: Arc::new(RwLock::new(HashMap::new())),
            check_interval,
            stats: Arc::new(HealthStats::default()),
        }
    }

    /// Add a region to monitor
    pub async fn add_region(&self, region_id: String) {
        let mut health_status = self.health_status.write().await;
        health_status.insert(
            region_id,
            RegionHealth {
                is_healthy: true,
                last_success: None,
                last_check: Utc::now(),
                latency_ms: None,
                recent_errors: 0,
                health_score: 1.0,
            },
        );
    }

    /// Remove a region from monitoring
    pub async fn remove_region(&self, region_id: &str) {
        let mut health_status = self.health_status.write().await;
        health_status.remove(region_id);
    }

    /// Get list of healthy regions
    pub async fn get_healthy_regions(&self) -> Vec<String> {
        let health_status = self.health_status.read().await;
        health_status
            .iter()
            .filter(|(_, health)| health.is_healthy)
            .map(|(region_id, _)| region_id.clone())
            .collect()
    }

    /// Perform health check for all regions
    pub async fn check_all_regions(&self) -> Result<()> {
        let regions: Vec<String> = {
            let health_status = self.health_status.read().await;
            health_status.keys().cloned().collect()
        };

        for region_id in regions {
            self.check_region_health(&region_id).await?;
        }

        Ok(())
    }

    /// Check health of a specific region
    async fn check_region_health(&self, region_id: &str) -> Result<()> {
        let start_time = Instant::now();
        self.stats
            .total_health_checks
            .fetch_add(1, Ordering::Relaxed);

        // Simulate health check - in real implementation, this would ping the region
        let is_healthy = fastrand::f32() > 0.1; // 90% success rate
        let latency = start_time.elapsed();

        let mut health_status = self.health_status.write().await;
        if let Some(health) = health_status.get_mut(region_id) {
            health.last_check = Utc::now();
            health.latency_ms = Some(latency.as_millis() as u64);

            if is_healthy {
                health.is_healthy = true;
                health.last_success = Some(Utc::now());
                health.recent_errors = 0;
                health.health_score = (health.health_score + 0.1).min(1.0);
            } else {
                health.recent_errors += 1;
                health.health_score = (health.health_score - 0.2).max(0.0);

                if health.recent_errors > 3 {
                    health.is_healthy = false;
                    self.stats
                        .failed_health_checks
                        .fetch_add(1, Ordering::Relaxed);
                }
            }
        }

        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::event::EventMetadata;
    use std::collections::HashMap;

    fn create_test_region(id: &str) -> RegionConfig {
        RegionConfig {
            region_id: id.to_string(),
            region_name: format!("Region {id}"),
            location: GeographicLocation {
                country: "US".to_string(),
                region: "California".to_string(),
                city: "San Francisco".to_string(),
                latitude: 37.7749,
                longitude: -122.4194,
                availability_zone: Some("us-west-1a".to_string()),
            },
            endpoints: vec![RegionEndpoint {
                url: format!("https://{id}.example.com"),
                endpoint_type: EndpointType::Primary,
                is_healthy: true,
                last_health_check: Some(Utc::now()),
                auth: None,
            }],
            priority: 1,
            is_write_active: true,
            is_read_active: true,
            replication_mode: ReplicationMode::Asynchronous,
            latency_map: HashMap::new(),
        }
    }

    fn create_test_event() -> StreamEvent {
        StreamEvent::TripleAdded {
            subject: "http://test.org/subject".to_string(),
            predicate: "http://test.org/predicate".to_string(),
            object: "\"test_value\"".to_string(),
            graph: None,
            metadata: EventMetadata {
                event_id: Uuid::new_v4().to_string(),
                timestamp: Utc::now(),
                source: "test".to_string(),
                user: None,
                context: None,
                caused_by: None,
                version: "1.0".to_string(),
                properties: HashMap::new(),
                checksum: None,
            },
        }
    }

    #[tokio::test]
    async fn test_replication_manager_creation() {
        let config = ReplicationConfig {
            strategy: ReplicationStrategy::FullReplication,
            conflict_resolution: ConflictResolution::LastWriteWins,
            max_lag_ms: 1000,
            replication_timeout: Duration::from_secs(30),
            enable_compression: true,
            batch_size: 100,
            health_check_interval: Duration::from_secs(60),
            failover_timeout: Duration::from_secs(300),
        };

        let manager = MultiRegionReplicationManager::new(config, "us-west-1".to_string());
        assert_eq!(manager.current_region, "us-west-1");
    }

    #[tokio::test]
    async fn test_region_management() {
        let config = ReplicationConfig {
            strategy: ReplicationStrategy::FullReplication,
            conflict_resolution: ConflictResolution::LastWriteWins,
            max_lag_ms: 1000,
            replication_timeout: Duration::from_secs(30),
            enable_compression: true,
            batch_size: 100,
            health_check_interval: Duration::from_secs(60),
            failover_timeout: Duration::from_secs(300),
        };

        let manager = MultiRegionReplicationManager::new(config, "us-west-1".to_string());

        // Add regions
        manager
            .add_region(create_test_region("us-east-1"))
            .await
            .unwrap();
        manager
            .add_region(create_test_region("eu-west-1"))
            .await
            .unwrap();

        let regions = manager.regions.read().await;
        assert_eq!(regions.len(), 2);
        assert!(regions.contains_key("us-east-1"));
        assert!(regions.contains_key("eu-west-1"));
    }

    #[tokio::test]
    async fn test_vector_clock() {
        let mut clock1 = VectorClock::new();
        let mut clock2 = VectorClock::new();

        clock1.increment("region1");
        clock2.increment("region2");

        assert!(clock1.is_concurrent(&clock2));
        assert!(!clock1.happens_before(&clock2));

        clock1.update(&clock2);
        clock1.increment("region1");

        assert!(clock2.happens_before(&clock1));
        assert!(!clock1.happens_before(&clock2));
    }

    #[tokio::test]
    async fn test_health_monitor() {
        let monitor = RegionHealthMonitor::new(Duration::from_secs(60));

        monitor.add_region("us-west-1".to_string()).await;
        monitor.add_region("us-east-1".to_string()).await;

        let healthy_regions = monitor.get_healthy_regions().await;
        assert_eq!(healthy_regions.len(), 2);

        monitor.check_all_regions().await.unwrap();

        let stats = &monitor.stats;
        assert!(stats.total_health_checks.load(Ordering::Relaxed) >= 2);
    }

    #[test]
    fn test_replication_config() {
        let config = ReplicationConfig {
            strategy: ReplicationStrategy::SelectiveReplication {
                event_types: ["TripleAdded", "TripleRemoved"]
                    .iter()
                    .map(|s| s.to_string())
                    .collect(),
            },
            conflict_resolution: ConflictResolution::RegionPriority {
                priority_order: vec!["us-west-1".to_string(), "us-east-1".to_string()],
            },
            max_lag_ms: 500,
            replication_timeout: Duration::from_secs(15),
            enable_compression: false,
            batch_size: 50,
            health_check_interval: Duration::from_secs(30),
            failover_timeout: Duration::from_secs(120),
        };

        match config.strategy {
            ReplicationStrategy::SelectiveReplication { ref event_types } => {
                assert_eq!(event_types.len(), 2);
            }
            _ => panic!("Wrong strategy type"),
        }
    }

    fn make_replication_config(conflict_resolution: ConflictResolution) -> ReplicationConfig {
        ReplicationConfig {
            strategy: ReplicationStrategy::FullReplication,
            conflict_resolution,
            max_lag_ms: 1000,
            replication_timeout: Duration::from_secs(30),
            enable_compression: true,
            batch_size: 100,
            health_check_interval: Duration::from_secs(60),
            failover_timeout: Duration::from_secs(300),
        }
    }

    fn event_with(source: &str, timestamp: DateTime<Utc>) -> StreamEvent {
        let mut event = create_test_event();
        if let StreamEvent::TripleAdded { metadata, .. } = &mut event {
            metadata.source = source.to_string();
            metadata.timestamp = timestamp;
        }
        event
    }

    /// Regression test for multi_region_replication.rs:561 — `FirstWriteWins`
    /// used to fall into the catch-all `_ => warn!(...)` arm and never
    /// resolve or count the conflict at all. It must now actually resolve
    /// (reflected by `conflicts_resolved`) instead of being silently dropped.
    #[tokio::test]
    async fn test_handle_conflict_first_write_wins_resolves() {
        let manager = MultiRegionReplicationManager::new(
            make_replication_config(ConflictResolution::FirstWriteWins),
            "us-west-1".to_string(),
        );

        let now = Utc::now();
        let conflict = ConflictInfo {
            conflict_type: ConflictType::WriteWrite,
            conflicting_events: vec![
                event_with("us-east-1", now),
                event_with("us-west-1", now - chrono::Duration::seconds(10)),
            ],
            resolution_strategy: ConflictResolution::FirstWriteWins,
            resolved_at: None,
            resolution_result: None,
        };

        manager.handle_conflict(conflict).await.unwrap();

        assert_eq!(
            manager.stats.conflicts_resolved.load(Ordering::Relaxed),
            1,
            "FirstWriteWins must actually resolve the conflict, not silently drop it"
        );
        assert!(
            manager.get_pending_conflicts().await.is_empty(),
            "a resolved conflict must not also sit in the manual-resolution queue"
        );
    }

    /// Regression test for multi_region_replication.rs:561 — `RegionPriority`
    /// used to be silently dropped by the same catch-all arm. It must now
    /// pick the event from the highest-priority listed region and count as
    /// resolved.
    #[tokio::test]
    async fn test_handle_conflict_region_priority_resolves() {
        let manager = MultiRegionReplicationManager::new(
            make_replication_config(ConflictResolution::RegionPriority {
                priority_order: vec!["us-west-1".to_string(), "us-east-1".to_string()],
            }),
            "us-west-1".to_string(),
        );

        let now = Utc::now();
        // The lower-priority region's event is newer, so a naive
        // Last-Write-Wins fallback would pick it; RegionPriority must still
        // prefer "us-west-1" because it's first in `priority_order`.
        let conflict = ConflictInfo {
            conflict_type: ConflictType::WriteWrite,
            conflicting_events: vec![
                event_with("us-east-1", now),
                event_with("us-west-1", now - chrono::Duration::seconds(10)),
            ],
            resolution_strategy: ConflictResolution::RegionPriority {
                priority_order: vec!["us-west-1".to_string(), "us-east-1".to_string()],
            },
            resolved_at: None,
            resolution_result: None,
        };

        manager.handle_conflict(conflict).await.unwrap();

        assert_eq!(
            manager.stats.conflicts_resolved.load(Ordering::Relaxed),
            1,
            "RegionPriority must actually resolve the conflict, not silently drop it"
        );
    }

    /// Regression test for multi_region_replication.rs:561 — `Custom` used to
    /// be silently dropped (no queueing, no resolution). It must now be
    /// routed into the manual-resolution queue instead of being discarded.
    #[tokio::test]
    async fn test_handle_conflict_custom_routes_to_manual_queue() {
        let manager = MultiRegionReplicationManager::new(
            make_replication_config(ConflictResolution::Custom {
                resolver_name: "not-yet-implemented".to_string(),
            }),
            "us-west-1".to_string(),
        );

        let conflict = ConflictInfo {
            conflict_type: ConflictType::WriteWrite,
            conflicting_events: vec![event_with("us-east-1", Utc::now())],
            resolution_strategy: ConflictResolution::Custom {
                resolver_name: "not-yet-implemented".to_string(),
            },
            resolved_at: None,
            resolution_result: None,
        };

        manager.handle_conflict(conflict).await.unwrap();

        let pending = manager.get_pending_conflicts().await;
        assert_eq!(
            pending.len(),
            1,
            "an unimplemented Custom resolver must route the conflict to the manual queue, \
             not discard it"
        );
    }
}