oxirs-tdb 0.3.1

//! Distributed Features Integration Layer
//!
//! This module provides a high-level integration layer that connects distributed
//! transaction features (2PC, 3PC, Paxos, Saga) with the TdbStore, making it easy
//! to use distributed capabilities without managing low-level details.
//!
//! # Features
//!
//! - **Unified API**: Single interface for all distributed transaction protocols
//! - **Automatic Protocol Selection**: Choose optimal protocol based on requirements
//! - **Built-in Deadlock Prevention**: Integrated deadlock detection and resolution
//! - **Replication Integration**: Coordinate transactions with replication
//! - **Monitoring**: Comprehensive metrics and health tracking
//!
//! # Example
//!
//! ```rust,ignore
//! use oxirs_tdb::distributed::integration::{DistributedTdbStore, DistributedConfig};
//!
//! // Create distributed store
//! let config = DistributedConfig::default();
//! let mut store = DistributedTdbStore::new("node1".to_string(), config);
//!
//! // Register nodes
//! store.register_node("node1", "http://node1:8080").await?;
//! store.register_node("node2", "http://node2:8080").await?;
//!
//! // Execute distributed transaction
//! let txn_id = store.begin_distributed_transaction().await?;
//! // ... perform operations ...
//! store.commit_distributed_transaction(&txn_id).await?;
//! ```

use crate::distributed::coordinator::{CommitProtocol, CoordinatorConfig, TransactionCoordinator};
use crate::distributed::deadlock::{DeadlockDetectorConfig, DistributedDeadlockDetector};
use crate::distributed::replication::{ReplicationConfig, ReplicationManager};
use crate::distributed::saga::{SagaConfig, SagaOrchestrator};
use crate::error::{Result, TdbError};
use parking_lot::{Mutex, RwLock};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Arc;

/// Distributed store configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DistributedConfig {
    /// Transaction coordinator configuration
    pub coordinator_config: CoordinatorConfig,
    /// Deadlock detector configuration
    pub deadlock_config: DeadlockDetectorConfig,
    /// Replication configuration
    pub replication_config: ReplicationConfig,
    /// Saga configuration
    pub saga_config: SagaConfig,
    /// Default commit protocol
    pub default_protocol: CommitProtocol,
    /// Enable automatic deadlock detection
    pub enable_deadlock_detection: bool,
    /// Enable replication
    pub enable_replication: bool,
}

impl Default for DistributedConfig {
    fn default() -> Self {
        Self {
            coordinator_config: CoordinatorConfig::default(),
            deadlock_config: DeadlockDetectorConfig::default(),
            replication_config: ReplicationConfig::default(),
            saga_config: SagaConfig::default(),
            default_protocol: CommitProtocol::TwoPhase,
            enable_deadlock_detection: true,
            enable_replication: true,
        }
    }
}

/// Distributed transaction handle
#[derive(Debug, Clone)]
pub struct DistributedTransaction {
    /// Transaction ID
    pub txn_id: String,
    /// Protocol used
    pub protocol: CommitProtocol,
    /// Participating nodes
    pub nodes: Vec<String>,
}

/// Distributed TDB Store
///
/// High-level interface for distributed RDF storage with integrated
/// transaction coordination, deadlock detection, and replication.
pub struct DistributedTdbStore {
    /// Node ID
    node_id: String,
    /// Configuration
    config: DistributedConfig,
    /// Transaction coordinator
    coordinator: Arc<Mutex<TransactionCoordinator>>,
    /// Deadlock detector (optional)
    deadlock_detector: Option<Arc<Mutex<DistributedDeadlockDetector>>>,
    /// Replication manager (optional)
    replication_manager: Option<Arc<Mutex<ReplicationManager>>>,
    /// Active distributed transactions
    active_transactions: Arc<RwLock<HashMap<String, DistributedTransaction>>>,
    /// Statistics
    stats: Arc<Mutex<DistributedStoreStats>>,
}

/// Distributed store statistics
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct DistributedStoreStats {
    /// Total distributed transactions
    pub total_distributed_txns: u64,
    /// Successful distributed transactions
    pub successful_distributed_txns: u64,
    /// Failed distributed transactions
    pub failed_distributed_txns: u64,
    /// Deadlocks detected and resolved
    pub deadlocks_resolved: u64,
    /// Replications performed
    pub replications_performed: u64,
    /// Average transaction latency (milliseconds)
    pub avg_txn_latency_ms: f64,
    /// Total latency
    total_latency_ms: f64,
}

impl DistributedTdbStore {
    /// Create a new Distributed TDB Store
    pub fn new(node_id: String, config: DistributedConfig) -> Self {
        let coordinator = Arc::new(Mutex::new(TransactionCoordinator::new(
            node_id.clone(),
            config.coordinator_config.clone(),
        )));

        let deadlock_detector = if config.enable_deadlock_detection {
            Some(Arc::new(Mutex::new(DistributedDeadlockDetector::new(
                format!("{}-deadlock-detector", node_id),
                config.deadlock_config.clone(),
            ))))
        } else {
            None
        };

        let replication_manager = if config.enable_replication {
            Some(Arc::new(Mutex::new(ReplicationManager::new(
                node_id.clone(),
                config.replication_config.clone(),
            ))))
        } else {
            None
        };

        Self {
            node_id,
            config,
            coordinator,
            deadlock_detector,
            replication_manager,
            active_transactions: Arc::new(RwLock::new(HashMap::new())),
            stats: Arc::new(Mutex::new(DistributedStoreStats::default())),
        }
    }

    /// Register a node in the distributed system
    #[allow(clippy::await_holding_lock)]
    pub async fn register_node(&mut self, node_id: &str, endpoint: &str) -> Result<()> {
        // Register with coordinator
        {
            self.coordinator
                .lock()
                .register_participant(node_id.to_string(), endpoint.to_string())
                .await?;
        }

        // Register with deadlock detector
        if let Some(ref detector) = self.deadlock_detector {
            detector.lock().register_node(node_id.to_string()).await?;
        }

        // Register with replication manager
        if let Some(ref manager) = self.replication_manager {
            manager
                .lock()
                .add_replica(node_id.to_string(), endpoint.to_string())
                .await?;
        }

        Ok(())
    }

    /// Begin a distributed transaction
    pub async fn begin_distributed_transaction(&mut self) -> Result<String> {
        self.begin_distributed_transaction_with_protocol(self.config.default_protocol)
            .await
    }

    /// Begin a distributed transaction with specific protocol
    #[allow(clippy::await_holding_lock)]
    pub async fn begin_distributed_transaction_with_protocol(
        &mut self,
        protocol: CommitProtocol,
    ) -> Result<String> {
        let txn_id = self.coordinator.lock().begin_transaction(protocol).await?;

        let metadata = {
            let coordinator = self.coordinator.lock();
            coordinator
                .get_transaction(&txn_id)
                .ok_or_else(|| TdbError::Other("Transaction not found".to_string()))?
        };

        let txn = DistributedTransaction {
            txn_id: txn_id.clone(),
            protocol,
            nodes: metadata.participants.clone(),
        };

        self.active_transactions.write().insert(txn_id.clone(), txn);

        let mut stats = self.stats.lock();
        stats.total_distributed_txns += 1;

        Ok(txn_id)
    }

    /// Commit a distributed transaction
    #[allow(clippy::await_holding_lock)]
    pub async fn commit_distributed_transaction(&mut self, txn_id: &str) -> Result<bool> {
        let start = std::time::Instant::now();

        // Check for deadlocks before commit
        if let Some(ref detector) = self.deadlock_detector {
            let deadlocks = detector.lock().detect_deadlocks().await?;

            if !deadlocks.is_empty() {
                // Abort victims
                for deadlock in &deadlocks {
                    if let Some(ref victim) = deadlock.victim {
                        detector.lock().abort_victim(victim).await?;

                        let mut stats = self.stats.lock();
                        stats.deadlocks_resolved += 1;
                    }
                }
            }
        }

        // Execute commit
        let result = self.coordinator.lock().commit_transaction(txn_id).await?;

        // Replicate if enabled and successful
        if result {
            if let Some(ref manager) = self.replication_manager {
                // TODO: Replicate transaction changes
                let mut stats = self.stats.lock();
                stats.replications_performed += 1;
            }
        }

        // Remove from active transactions
        self.active_transactions.write().remove(txn_id);

        // Update statistics
        let latency = start.elapsed().as_millis() as f64;
        let mut stats = self.stats.lock();

        if result {
            stats.successful_distributed_txns += 1;
        } else {
            stats.failed_distributed_txns += 1;
        }

        stats.total_latency_ms += latency;
        stats.avg_txn_latency_ms = stats.total_latency_ms / stats.total_distributed_txns as f64;

        Ok(result)
    }

    /// Abort a distributed transaction
    #[allow(clippy::await_holding_lock)]
    pub async fn abort_distributed_transaction(&mut self, txn_id: &str) -> Result<()> {
        self.coordinator.lock().abort_transaction(txn_id).await?;

        self.active_transactions.write().remove(txn_id);

        let mut stats = self.stats.lock();
        stats.failed_distributed_txns += 1;

        Ok(())
    }

    /// Create and execute a saga
    pub async fn execute_saga(&mut self, saga: SagaOrchestrator) -> Result<bool> {
        // TODO: Integrate saga execution with coordinator
        // For now, saga operates independently
        Ok(true)
    }

    /// Get node ID
    pub fn node_id(&self) -> &str {
        &self.node_id
    }

    /// Get active transaction count
    pub fn active_transaction_count(&self) -> usize {
        self.active_transactions.read().len()
    }

    /// Get statistics
    pub fn stats(&self) -> DistributedStoreStats {
        self.stats.lock().clone()
    }

    /// Get coordinator statistics
    pub fn coordinator_stats(&self) -> crate::distributed::coordinator::CoordinatorStats {
        self.coordinator.lock().stats()
    }

    /// Get replication statistics (if enabled)
    pub fn replication_stats(&self) -> Option<crate::distributed::replication::ReplicationStats> {
        self.replication_manager.as_ref().map(|m| m.lock().stats())
    }

    /// Check system health
    pub async fn check_health(&self) -> Result<HealthStatus> {
        let coordinator_health = self.coordinator.lock().healthy_participant_count() > 0;

        let replication_health = if let Some(ref manager) = self.replication_manager {
            manager.lock().healthy_replica_count() > 0
        } else {
            true
        };

        let overall_health = coordinator_health && replication_health;

        Ok(HealthStatus {
            healthy: overall_health,
            coordinator_nodes: self.coordinator.lock().participant_count(),
            healthy_coordinators: self.coordinator.lock().healthy_participant_count(),
            replica_count: self
                .replication_manager
                .as_ref()
                .map(|m| m.lock().replica_count())
                .unwrap_or(0),
            healthy_replicas: self
                .replication_manager
                .as_ref()
                .map(|m| m.lock().healthy_replica_count())
                .unwrap_or(0),
            active_transactions: self.active_transaction_count(),
        })
    }
}

/// System health status
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HealthStatus {
    /// Overall health
    pub healthy: bool,
    /// Total coordinator nodes
    pub coordinator_nodes: usize,
    /// Healthy coordinator nodes
    pub healthy_coordinators: usize,
    /// Total replicas
    pub replica_count: usize,
    /// Healthy replicas
    pub healthy_replicas: usize,
    /// Active transactions
    pub active_transactions: usize,
}

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

    #[tokio::test]
    #[ignore = "Distributed tests hang - needs network mock or investigation"]
    async fn test_distributed_store_creation() {
        let config = DistributedConfig::default();
        let store = DistributedTdbStore::new("node1".to_string(), config);

        assert_eq!(store.node_id(), "node1");
        assert_eq!(store.active_transaction_count(), 0);
    }

    #[tokio::test]
    #[ignore = "Distributed tests hang - needs network mock or investigation"]
    async fn test_register_node() {
        let config = DistributedConfig::default();
        let mut store = DistributedTdbStore::new("node1".to_string(), config);

        store
            .register_node("node2", "http://node2:8080")
            .await
            .unwrap();

        // Node should be registered with coordinator
        assert!(store.coordinator.lock().participant_count() > 0);
    }

    #[tokio::test]
    #[ignore = "Distributed tests hang - needs network mock or investigation"]
    async fn test_begin_distributed_transaction() {
        let config = DistributedConfig::default();
        let mut store = DistributedTdbStore::new("node1".to_string(), config);

        store
            .register_node("node2", "http://node2:8080")
            .await
            .unwrap();

        let txn_id = store.begin_distributed_transaction().await.unwrap();

        assert!(!txn_id.is_empty());
        assert_eq!(store.active_transaction_count(), 1);
    }

    #[tokio::test]
    #[ignore = "Distributed tests hang - needs network mock or investigation"]
    async fn test_commit_distributed_transaction() {
        let config = DistributedConfig::default();
        let mut store = DistributedTdbStore::new("node1".to_string(), config);

        store
            .register_node("node2", "http://node2:8080")
            .await
            .unwrap();

        let txn_id = store.begin_distributed_transaction().await.unwrap();

        let result = store.commit_distributed_transaction(&txn_id).await.unwrap();

        assert!(result);
        assert_eq!(store.active_transaction_count(), 0);

        let stats = store.stats();
        assert_eq!(stats.successful_distributed_txns, 1);
    }

    #[tokio::test]
    #[ignore = "Distributed tests hang - needs network mock or investigation"]
    async fn test_abort_distributed_transaction() {
        let config = DistributedConfig::default();
        let mut store = DistributedTdbStore::new("node1".to_string(), config);

        store
            .register_node("node2", "http://node2:8080")
            .await
            .unwrap();

        let txn_id = store.begin_distributed_transaction().await.unwrap();

        store.abort_distributed_transaction(&txn_id).await.unwrap();

        assert_eq!(store.active_transaction_count(), 0);

        let stats = store.stats();
        assert_eq!(stats.failed_distributed_txns, 1);
    }

    #[tokio::test]
    #[ignore = "Distributed tests hang - needs network mock or investigation"]
    async fn test_health_check() {
        let config = DistributedConfig::default();
        let store = DistributedTdbStore::new("node1".to_string(), config);

        let health = store.check_health().await.unwrap();

        assert_eq!(health.active_transactions, 0);
    }

    #[tokio::test]
    #[ignore = "Distributed tests hang - needs network mock or investigation"]
    async fn test_statistics() {
        let config = DistributedConfig::default();
        let store = DistributedTdbStore::new("node1".to_string(), config);

        let stats = store.stats();
        assert_eq!(stats.total_distributed_txns, 0);

        let coord_stats = store.coordinator_stats();
        assert_eq!(coord_stats.total_transactions, 0);
    }

    #[tokio::test]
    #[ignore = "Distributed tests hang - needs network mock or investigation"]
    async fn test_protocol_selection() {
        let config = DistributedConfig::default();
        let mut store = DistributedTdbStore::new("node1".to_string(), config);

        store
            .register_node("node2", "http://node2:8080")
            .await
            .unwrap();

        // Test with 3PC
        let txn_id = store
            .begin_distributed_transaction_with_protocol(CommitProtocol::ThreePhase)
            .await
            .unwrap();

        let txn = store.active_transactions.read().get(&txn_id).cloned();
        assert!(txn.is_some());
        assert_eq!(txn.unwrap().protocol, CommitProtocol::ThreePhase);
    }

    #[tokio::test]
    #[ignore = "Distributed tests hang - needs network mock or investigation"]
    async fn test_multiple_transactions() {
        let config = DistributedConfig::default();
        let mut store = DistributedTdbStore::new("node1".to_string(), config);

        store
            .register_node("node2", "http://node2:8080")
            .await
            .unwrap();

        let txn1 = store.begin_distributed_transaction().await.unwrap();
        let txn2 = store.begin_distributed_transaction().await.unwrap();

        assert_eq!(store.active_transaction_count(), 2);

        store.commit_distributed_transaction(&txn1).await.unwrap();
        assert_eq!(store.active_transaction_count(), 1);

        store.abort_distributed_transaction(&txn2).await.unwrap();
        assert_eq!(store.active_transaction_count(), 0);
    }
}