taskflow-rs 0.1.1

use crate::error::{Result, TaskFlowError};
use crate::storage::TaskStorage;
use crate::task::{Task, TaskStatus};
use std::collections::HashSet;
use std::pin::Pin;
use std::sync::Arc;

pub struct DependencyResolver {
    storage: Arc<dyn TaskStorage>,
}

impl DependencyResolver {
    pub fn new(storage: Arc<dyn TaskStorage>) -> Self {
        Self { storage }
    }

    pub async fn validate_dependencies(&self, task: &Task) -> Result<()> {
        for dep_id in &task.definition.dependencies {
            if self.storage.get_task(dep_id).await?.is_none() {
                return Err(TaskFlowError::InvalidConfiguration(format!(
                    "Dependency task not found: {}",
                    dep_id
                )));
            }
        }

        if self
            .has_circular_dependency(task, &mut HashSet::new())
            .await?
        {
            return Err(TaskFlowError::InvalidConfiguration(
                "Circular dependency detected".to_string(),
            ));
        }

        Ok(())
    }

    pub async fn are_dependencies_satisfied(&self, task: &Task) -> Result<bool> {
        for dep_id in &task.definition.dependencies {
            if let Some(dep_task) = self.storage.get_task(dep_id).await? {
                if !matches!(dep_task.status, TaskStatus::Completed) {
                    return Ok(false);
                }
            } else {
                return Err(TaskFlowError::TaskNotFound(dep_id.clone()));
            }
        }
        Ok(true)
    }

    fn has_circular_dependency<'a>(
        &'a self,
        task: &'a Task,
        visited: &'a mut HashSet<String>,
    ) -> Pin<Box<dyn std::future::Future<Output = Result<bool>> + Send + 'a>> {
        Box::pin(async move {
            if visited.contains(&task.definition.id) {
                return Ok(true);
            }

            visited.insert(task.definition.id.clone());

            for dep_id in &task.definition.dependencies {
                if let Some(dep_task) = self.storage.get_task(dep_id).await? {
                    if self.has_circular_dependency(&dep_task, visited).await? {
                        return Ok(true);
                    }
                }
            }

            visited.remove(&task.definition.id);
            Ok(false)
        })
    }
}