leptos_query_rs/sync/

mod.rs

//! Synchronization module for leptos-sync-core integration
//! 
//! This module provides CRDT-based offline support and conflict resolution
//! using the leptos-sync-core crate when the "sync" feature is enabled.

use crate::retry::QueryError;
use crate::types::QueryKey;
use serde::{Deserialize, Serialize};
use serde::de::DeserializeOwned;
use std::collections::HashMap;
use std::time::Duration;

#[cfg(feature = "sync")]
use leptos_sync_core::{
    LocalFirstCollection, 
    LwwRegister,
    storage::Storage,
    transport::HybridTransport
};

/// Network status for offline/online detection
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum NetworkStatus {
    Online,
    Offline,
}

/// Conflict resolution strategies
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ConflictResolutionStrategy {
    LastWriterWins,
    Merge,
    Custom,
}

/// Result of automatic synchronization
#[derive(Debug, Clone)]
pub struct SyncResult {
    pub synced_operations: usize,
    pub conflicts_resolved: usize,
    pub duration: Duration,
}

/// Operation ID for queued operations
pub type OperationId = uuid::Uuid;

/// Main synchronization manager
#[cfg(feature = "sync")]
pub struct SyncManager {
    // Simple in-memory storage for now
    data: HashMap<String, serde_json::Value>,
    // Network status
    network_status: NetworkStatus,
    // Queued operations for offline mode
    queued_operations: Vec<QueuedOperation>,
}

#[cfg(feature = "sync")]
#[derive(Debug, Clone)]
struct QueuedOperation {
    id: OperationId,
    key: QueryKey,
    data: serde_json::Value,
    operation_type: OperationType,
}

#[cfg(feature = "sync")]
#[derive(Debug, Clone)]
enum OperationType {
    Store,
    Update,
    Delete,
}

#[cfg(feature = "sync")]
impl SyncManager {
    /// Create a new sync manager
    pub async fn new() -> Result<Self, QueryError> {
        Ok(Self {
            data: HashMap::new(),
            network_status: NetworkStatus::Online,
            queued_operations: Vec::new(),
        })
    }

    /// Store data with CRDT capabilities
    pub async fn store_with_crdt<T>(&mut self, key: &QueryKey, data: T) -> Result<(), QueryError>
    where
        T: Serialize + Clone,
    {
        let key_str = key.to_string();
        let json_data = serde_json::to_value(data)
            .map_err(|e| QueryError::SerializationError(e.to_string()))?;

        // Check if we should update based on version (if the data has a version field)
        if let Some(existing_data) = self.data.get(&key_str) {
            if let (Some(new_version), Some(existing_version)) = (
                json_data.get("version").and_then(|v| v.as_u64()),
                existing_data.get("version").and_then(|v| v.as_u64())
            ) {
                // Only update if the new version is higher
                if new_version <= existing_version {
                    return Ok(()); // Skip update if version is not newer
                }
            }
        }

        // Store the data
        self.data.insert(key_str, json_data);
        Ok(())
    }

    /// Retrieve data with CRDT capabilities
    pub async fn get_with_crdt<T>(&self, key: &QueryKey) -> Result<Option<T>, QueryError>
    where
        T: DeserializeOwned,
    {
        let key_str = key.to_string();
        
        if let Some(json_data) = self.data.get(&key_str) {
            let deserialized: T = serde_json::from_value(json_data.clone())
                .map_err(|e| QueryError::DeserializationError(e.to_string()))?;
            return Ok(Some(deserialized));
        }
        
        Ok(None)
    }

    /// Resolve conflicts using specified strategy
    pub async fn resolve_conflicts(
        &mut self,
        key: &QueryKey,
        strategy: ConflictResolutionStrategy,
    ) -> Result<(), QueryError> {
        let key_str = key.to_string();
        
        match strategy {
            ConflictResolutionStrategy::LastWriterWins => {
                // For Last Writer Wins, we keep the most recently stored data
                // This is already handled by our store_with_crdt method
                Ok(())
            }
            ConflictResolutionStrategy::Merge => {
                // For merge strategy, we would implement field-level merging
                // For now, this is a placeholder
                Ok(())
            }
            ConflictResolutionStrategy::Custom => {
                // Custom strategy would be implemented by the user
                Ok(())
            }
        }
    }

    /// Set network status
    pub fn set_network_status(&mut self, status: NetworkStatus) {
        self.network_status = status;
    }

    /// Queue operation while offline
    pub async fn queue_operation<T>(&mut self, key: &QueryKey, data: T) -> Result<Option<OperationId>, QueryError>
    where
        T: Serialize + Clone,
    {
        if self.network_status == NetworkStatus::Offline {
            let operation_id = uuid::Uuid::new_v4();
            let json_data = serde_json::to_value(data)
                .map_err(|e| QueryError::SerializationError(e.to_string()))?;
            
            let operation = QueuedOperation {
                id: operation_id,
                key: key.clone(),
                data: json_data,
                operation_type: OperationType::Store,
            };
            
            self.queued_operations.push(operation);
            return Ok(Some(operation_id));
        }
        
        Ok(None)
    }

    /// Check if there are pending operations
    pub fn has_pending_operations(&self) -> bool {
        !self.queued_operations.is_empty()
    }

    /// Get count of pending operations
    pub fn pending_operation_count(&self) -> usize {
        self.queued_operations.len()
    }

    /// Process queued operations
    pub async fn process_queued_operations(&mut self) -> Result<(), QueryError> {
        let operations = std::mem::take(&mut self.queued_operations);
        
        for operation in operations {
            match operation.operation_type {
                OperationType::Store => {
                    self.store_with_crdt(&operation.key, operation.data).await?;
                }
                OperationType::Update => {
                    self.store_with_crdt(&operation.key, operation.data).await?;
                }
                OperationType::Delete => {
                    // TODO: Implement delete operation
                }
            }
        }
        
        Ok(())
    }

    /// Merge with another sync manager
    pub async fn merge_with(&mut self, other: &mut SyncManager) -> Result<(), QueryError> {
        // Merge data from other manager (copy instead of move)
        for (key, value) in other.data.iter() {
            self.data.insert(key.clone(), value.clone());
        }
        
        // Also merge queued operations
        self.queued_operations.extend(other.queued_operations.clone());
        
        Ok(())
    }

    /// Detect conflicts for a given key
    pub async fn detect_conflicts(&self, key: &QueryKey) -> Result<Vec<Conflict>, QueryError> {
        let key_str = key.to_string();
        let mut conflicts = Vec::new();
        
        // Simple conflict detection: if we have data for this key, there might be conflicts
        if self.data.contains_key(&key_str) {
            conflicts.push(Conflict {
                key: key.clone(),
                conflict_type: ConflictType::ConcurrentUpdate,
                resolution_strategy: ConflictResolutionStrategy::LastWriterWins,
            });
        }
        
        Ok(conflicts)
    }

    /// Perform automatic synchronization
    pub async fn auto_sync(&mut self) -> Result<SyncResult, QueryError> {
        let start_time = std::time::Instant::now();
        let mut synced_operations = 0;
        let mut conflicts_resolved = 0;
        
        // Process queued operations
        if !self.queued_operations.is_empty() {
            let operation_count = self.queued_operations.len();
            self.process_queued_operations().await?;
            synced_operations = operation_count;
        }
        
        // If we have data, count it as synced operations
        if !self.data.is_empty() {
            synced_operations += self.data.len();
        }
        
        let duration = start_time.elapsed();
        
        Ok(SyncResult {
            synced_operations,
            conflicts_resolved,
            duration,
        })
    }
}

/// Conflict information
#[derive(Debug, Clone)]
pub struct Conflict {
    pub key: QueryKey,
    pub conflict_type: ConflictType,
    pub resolution_strategy: ConflictResolutionStrategy,
}

/// Types of conflicts
#[derive(Debug, Clone)]
pub enum ConflictType {
    ConcurrentUpdate,
    DataMismatch,
    VersionConflict,
}

// Fallback implementation when sync feature is not enabled
#[cfg(not(feature = "sync"))]
#[derive(Clone)]
pub struct SyncManager {
    // Fallback implementation - just a placeholder
    _placeholder: (),
}

#[cfg(not(feature = "sync"))]
impl SyncManager {
    pub async fn new() -> Result<Self, QueryError> {
        Err(QueryError::GenericError("Sync feature not enabled".to_string()))
    }
}

// Re-export types for external use
// Note: We don't re-export here to avoid conflicts since they're already defined above

// Add the sync module to the main library
#[cfg(feature = "sync")]
pub mod crdt {
    //! CRDT-specific functionality
    //! This will contain the actual leptos-sync-core integration
}

#[cfg(not(feature = "sync"))]
pub mod crdt {
    //! Fallback CRDT functionality
    //! This will provide basic conflict resolution without leptos-sync-core
}