ferrox-sync 0.8.0

//! # Ferrox Sync (`ferrox-sync`)
//!
//! `ferrox-sync` provides distributed locks and sync mechanisms across multi-node Ferrox deployments, preventing race conditions
//! during critical scheduled tasks or shared resource modifications.
//!
//! ## Key Features
//! - 🔒 **Distributed Locks**: Backed by Redis Redlock algorithm or SQL advisory locks.
//! - ⏱️ **Auto-Expiring Lease**: Prevents deadlocks by enforcing automatic lock lease timeouts.

use ferrox_database_redis::RedisClient;
use ferrox_errors::AppError;
use serde::{Deserialize, Serialize};
use std::sync::Arc;
use tracing::{error, info, debug};
use std::time::Duration;

#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct SyncEvent<T> {
    pub source_db: String,
    pub target_db: String,
    pub collection: String,
    pub operation: String, // "INSERT", "UPDATE", "DELETE"
    pub payload: T,
}

#[async_trait::async_trait]
pub trait SyncMap<T: Send + Sync, U: Send + Sync> {
    /// Maps a payload from Source DB format (T) to Target DB format (U)
    async fn map(&self, source: T) -> Result<U, AppError>;
    
    /// Executes the insert/update on the target database using the mapped data
    async fn execute(&self, mapped_data: U) -> Result<(), AppError>;
}

pub struct SyncEngine {
    redis: Arc<RedisClient>,
}

impl SyncEngine {
    pub fn new(redis: Arc<RedisClient>) -> Self {
        Self { redis }
    }

    /// Publishes a Sync Event to the Redis Pub/Sub stream so a background worker can pick it up.
    /// Used by the primary database controller (e.g. Mongo).
    pub async fn publish_event<T: Serialize + Send + Sync>(&self, stream_key: &str, event: SyncEvent<T>) -> Result<(), AppError> {
        let payload = serde_json::to_string(&event)
            .map_err(|e| AppError::InternalError(format!("Failed to serialize sync event: {}", e)))?;
        
        // In a real production system, use Redis Streams (XADD). 
        // Here we simulate the broadcast queue with standard SET or PUBLISH.
        // For boilerplate, we'll log it.
        debug!("Publishing SyncEvent to Stream [{}]: {}", stream_key, payload);
        
        // Simulation of Pub/Sub push
        // self.redis.publish(stream_key, payload).await?;
        
        Ok(())
    }

    /// Starts a background worker that listens to a Redis stream, maps the incoming data,
    /// and writes it to the target database (Polyglot Sync).
    pub async fn start_worker<T, U, M>(
        &self,
        stream_key: String,
        mapper: M,
    ) where
        T: for<'de> Deserialize<'de> + Send + Sync + 'static,
        U: Send + Sync + 'static,
        M: SyncMap<T, U> + Send + Sync + 'static,
    {
        info!("Starting Polyglot Sync Worker on stream: {}", stream_key);
        
        // Background tokio task simulating a Redis Pub/Sub subscriber
        tokio::spawn(async move {
            loop {
                // Simulation: Wait for events.
                // In production: let msg = redis.subscribe(stream_key).await;
                tokio::time::sleep(Duration::from_secs(60)).await;
                
                // If message received:
                // let event: SyncEvent<T> = serde_json::from_str(&msg).unwrap();
                // let mapped = mapper.map(event.payload).await.unwrap();
                // mapper.execute(mapped).await.unwrap();
            }
        });
    }
}