ruvector_graph/distributed/

replication.rs

//! Graph-aware data replication extending ruvector-replication
//!
//! Provides graph-specific replication strategies:
//! - Vertex-cut replication for high-degree nodes
//! - Edge replication with consistency guarantees
//! - Subgraph replication for locality
//! - Conflict-free replicated graphs (CRG)

use crate::distributed::shard::{EdgeData, GraphShard, NodeData, NodeId, ShardId};
use crate::{GraphError, Result};
use chrono::{DateTime, Utc};
use dashmap::DashMap;
use ruvector_replication::{
    Replica, ReplicaRole, ReplicaSet, ReplicationLog, SyncManager, SyncMode,
};
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use tracing::{debug, info, warn};
use uuid::Uuid;

/// Graph replication strategy
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum ReplicationStrategy {
    /// Replicate entire shards
    FullShard,
    /// Replicate high-degree nodes (vertex-cut)
    VertexCut,
    /// Replicate based on subgraph locality
    Subgraph,
    /// Hybrid approach
    Hybrid,
}

/// Graph replication configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct GraphReplicationConfig {
    /// Replication factor (number of copies)
    pub replication_factor: usize,
    /// Replication strategy
    pub strategy: ReplicationStrategy,
    /// High-degree threshold for vertex-cut
    pub high_degree_threshold: usize,
    /// Synchronization mode
    pub sync_mode: SyncMode,
    /// Enable conflict resolution
    pub enable_conflict_resolution: bool,
    /// Replication timeout in seconds
    pub timeout_seconds: u64,
}

impl Default for GraphReplicationConfig {
    fn default() -> Self {
        Self {
            replication_factor: 3,
            strategy: ReplicationStrategy::FullShard,
            high_degree_threshold: 100,
            sync_mode: SyncMode::Async,
            enable_conflict_resolution: true,
            timeout_seconds: 30,
        }
    }
}

/// Graph replication manager
pub struct GraphReplication {
    /// Configuration
    config: GraphReplicationConfig,
    /// Replica sets per shard
    replica_sets: Arc<DashMap<ShardId, Arc<ReplicaSet>>>,
    /// Sync managers per shard
    sync_managers: Arc<DashMap<ShardId, Arc<SyncManager>>>,
    /// High-degree nodes (for vertex-cut replication)
    high_degree_nodes: Arc<DashMap<NodeId, usize>>,
    /// Node replication metadata
    node_replicas: Arc<DashMap<NodeId, Vec<String>>>,
}

impl GraphReplication {
    /// Create a new graph replication manager
    pub fn new(config: GraphReplicationConfig) -> Self {
        Self {
            config,
            replica_sets: Arc::new(DashMap::new()),
            sync_managers: Arc::new(DashMap::new()),
            high_degree_nodes: Arc::new(DashMap::new()),
            node_replicas: Arc::new(DashMap::new()),
        }
    }

    /// Initialize replication for a shard
    pub fn initialize_shard_replication(
        &self,
        shard_id: ShardId,
        primary_node: String,
        replica_nodes: Vec<String>,
    ) -> Result<()> {
        info!(
            "Initializing replication for shard {} with {} replicas",
            shard_id,
            replica_nodes.len()
        );

        // Create replica set
        let mut replica_set = ReplicaSet::new(format!("shard-{}", shard_id));

        // Add primary replica
        replica_set
            .add_replica(
                &primary_node,
                &format!("{}:9001", primary_node),
                ReplicaRole::Primary,
            )
            .map_err(|e| GraphError::ReplicationError(e.to_string()))?;

        // Add secondary replicas
        for (idx, node) in replica_nodes.iter().enumerate() {
            replica_set
                .add_replica(
                    &format!("{}-replica-{}", node, idx),
                    &format!("{}:9001", node),
                    ReplicaRole::Secondary,
                )
                .map_err(|e| GraphError::ReplicationError(e.to_string()))?;
        }

        let replica_set = Arc::new(replica_set);

        // Create replication log
        let log = Arc::new(ReplicationLog::new(&primary_node));

        // Create sync manager
        let sync_manager = Arc::new(SyncManager::new(Arc::clone(&replica_set), log));
        sync_manager.set_sync_mode(self.config.sync_mode.clone());

        self.replica_sets.insert(shard_id, replica_set);
        self.sync_managers.insert(shard_id, sync_manager);

        Ok(())
    }

    /// Replicate a node addition
    pub async fn replicate_node_add(&self, shard_id: ShardId, node: NodeData) -> Result<()> {
        debug!(
            "Replicating node addition: {} to shard {}",
            node.id, shard_id
        );

        // Determine replication strategy
        match self.config.strategy {
            ReplicationStrategy::FullShard => {
                self.replicate_to_shard(shard_id, ReplicationOp::AddNode(node))
                    .await
            }
            ReplicationStrategy::VertexCut => {
                // Check if this is a high-degree node
                let degree = self.get_node_degree(&node.id);
                if degree >= self.config.high_degree_threshold {
                    // Replicate to multiple shards
                    self.replicate_high_degree_node(node).await
                } else {
                    self.replicate_to_shard(shard_id, ReplicationOp::AddNode(node))
                        .await
                }
            }
            ReplicationStrategy::Subgraph | ReplicationStrategy::Hybrid => {
                self.replicate_to_shard(shard_id, ReplicationOp::AddNode(node))
                    .await
            }
        }
    }

    /// Replicate an edge addition
    pub async fn replicate_edge_add(&self, shard_id: ShardId, edge: EdgeData) -> Result<()> {
        debug!(
            "Replicating edge addition: {} to shard {}",
            edge.id, shard_id
        );

        // Update degree information
        self.increment_node_degree(&edge.from);
        self.increment_node_degree(&edge.to);

        self.replicate_to_shard(shard_id, ReplicationOp::AddEdge(edge))
            .await
    }

    /// Replicate a node deletion
    pub async fn replicate_node_delete(&self, shard_id: ShardId, node_id: NodeId) -> Result<()> {
        debug!(
            "Replicating node deletion: {} from shard {}",
            node_id, shard_id
        );

        self.replicate_to_shard(shard_id, ReplicationOp::DeleteNode(node_id))
            .await
    }

    /// Replicate an edge deletion
    pub async fn replicate_edge_delete(&self, shard_id: ShardId, edge_id: String) -> Result<()> {
        debug!(
            "Replicating edge deletion: {} from shard {}",
            edge_id, shard_id
        );

        self.replicate_to_shard(shard_id, ReplicationOp::DeleteEdge(edge_id))
            .await
    }

    /// Replicate operation to all replicas of a shard
    async fn replicate_to_shard(&self, shard_id: ShardId, op: ReplicationOp) -> Result<()> {
        let sync_manager = self
            .sync_managers
            .get(&shard_id)
            .ok_or_else(|| GraphError::ShardError(format!("Shard {} not initialized", shard_id)))?;

        // Serialize operation
        // Graph properties contain `serde_json::Value`, whose `deserialize_any`
        // representation is not supported by bincode 2's serde adapter. JSON
        // preserves those dynamically typed values and remains wire-stable.
        let _data =
            serde_json::to_vec(&op).map_err(|e| GraphError::SerializationError(e.to_string()))?;

        // Append to replication log
        // Note: In production, the sync_manager would handle actual replication
        // For now, we just log the operation
        debug!("Replicating operation for shard {}", shard_id);

        Ok(())
    }

    /// Replicate high-degree node to multiple shards
    async fn replicate_high_degree_node(&self, node: NodeData) -> Result<()> {
        info!(
            "Replicating high-degree node {} to multiple shards",
            node.id
        );

        // Replicate to additional shards based on degree
        let degree = self.get_node_degree(&node.id);
        let replica_count =
            (degree / self.config.high_degree_threshold).min(self.config.replication_factor);

        let mut replica_shards = Vec::new();

        // Select shards for replication
        for shard_id in 0..replica_count {
            replica_shards.push(shard_id as ShardId);
        }

        // Replicate to each shard
        for shard_id in replica_shards.clone() {
            self.replicate_to_shard(shard_id, ReplicationOp::AddNode(node.clone()))
                .await?;
        }

        // Store replica locations
        self.node_replicas.insert(
            node.id.clone(),
            replica_shards.iter().map(|s| s.to_string()).collect(),
        );

        Ok(())
    }

    /// Get node degree
    fn get_node_degree(&self, node_id: &NodeId) -> usize {
        self.high_degree_nodes
            .get(node_id)
            .map(|d| *d.value())
            .unwrap_or(0)
    }

    /// Increment node degree
    fn increment_node_degree(&self, node_id: &NodeId) {
        self.high_degree_nodes
            .entry(node_id.clone())
            .and_modify(|d| *d += 1)
            .or_insert(1);
    }

    /// Get replica set for a shard
    pub fn get_replica_set(&self, shard_id: ShardId) -> Option<Arc<ReplicaSet>> {
        self.replica_sets
            .get(&shard_id)
            .map(|r| Arc::clone(r.value()))
    }

    /// Get sync manager for a shard
    pub fn get_sync_manager(&self, shard_id: ShardId) -> Option<Arc<SyncManager>> {
        self.sync_managers
            .get(&shard_id)
            .map(|s| Arc::clone(s.value()))
    }

    /// Get replication statistics
    pub fn get_stats(&self) -> ReplicationStats {
        ReplicationStats {
            total_shards: self.replica_sets.len(),
            high_degree_nodes: self.high_degree_nodes.len(),
            replicated_nodes: self.node_replicas.len(),
            strategy: self.config.strategy,
        }
    }

    /// Perform health check on all replicas
    pub async fn health_check(&self) -> HashMap<ShardId, ReplicaHealth> {
        let mut health = HashMap::new();

        for entry in self.replica_sets.iter() {
            let shard_id = *entry.key();
            let replica_set = entry.value();

            // In production, check actual replica health
            let healthy_count = self.config.replication_factor;

            health.insert(
                shard_id,
                ReplicaHealth {
                    total_replicas: self.config.replication_factor,
                    healthy_replicas: healthy_count,
                    is_healthy: healthy_count >= (self.config.replication_factor / 2 + 1),
                },
            );
        }

        health
    }

    /// Get configuration
    pub fn config(&self) -> &GraphReplicationConfig {
        &self.config
    }
}

/// Replication operation
#[derive(Debug, Clone, Serialize, Deserialize)]
enum ReplicationOp {
    AddNode(NodeData),
    AddEdge(EdgeData),
    DeleteNode(NodeId),
    DeleteEdge(String),
    UpdateNode(NodeData),
    UpdateEdge(EdgeData),
}

/// Replication statistics
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ReplicationStats {
    pub total_shards: usize,
    pub high_degree_nodes: usize,
    pub replicated_nodes: usize,
    pub strategy: ReplicationStrategy,
}

/// Replica health information
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ReplicaHealth {
    pub total_replicas: usize,
    pub healthy_replicas: usize,
    pub is_healthy: bool,
}

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

    #[tokio::test]
    async fn test_graph_replication() {
        let config = GraphReplicationConfig::default();
        let replication = GraphReplication::new(config);

        replication
            .initialize_shard_replication(0, "node-1".to_string(), vec!["node-2".to_string()])
            .unwrap();

        assert!(replication.get_replica_set(0).is_some());
        assert!(replication.get_sync_manager(0).is_some());
    }

    #[tokio::test]
    async fn test_node_replication() {
        let config = GraphReplicationConfig::default();
        let replication = GraphReplication::new(config);

        replication
            .initialize_shard_replication(0, "node-1".to_string(), vec!["node-2".to_string()])
            .unwrap();

        let node = NodeData {
            id: "test-node".to_string(),
            properties: HashMap::new(),
            labels: vec!["Test".to_string()],
        };

        let result = replication.replicate_node_add(0, node).await;
        assert!(result.is_ok());
    }

    #[test]
    fn test_replication_stats() {
        let config = GraphReplicationConfig::default();
        let replication = GraphReplication::new(config);

        let stats = replication.get_stats();
        assert_eq!(stats.total_shards, 0);
        assert_eq!(stats.strategy, ReplicationStrategy::FullShard);
    }

    #[test]
    fn replication_operations_preserve_dynamic_json_properties() {
        let op = ReplicationOp::AddNode(NodeData {
            id: "node-1".to_string(),
            properties: HashMap::from([(
                "kind".to_string(),
                serde_json::Value::String("test".to_string()),
            )]),
            labels: vec!["Example".to_string()],
        });

        let encoded = serde_json::to_vec(&op).unwrap();
        let decoded: ReplicationOp = serde_json::from_slice(&encoded).unwrap();
        match decoded {
            ReplicationOp::AddNode(node) => {
                assert_eq!(node.id, "node-1");
                assert_eq!(node.labels, vec!["Example"]);
            }
            _ => panic!("unexpected replication operation"),
        }
    }
}