pe_tasks/

dependency.rs

//! # Task dependency DAG — blocks/requires/related with cycle detection.
//!
//! Tasks can depend on other tasks. Dependencies form a directed acyclic graph (DAG).
//! Cycle detection prevents deadlocks where A blocks B blocks A.

use std::collections::{HashMap, HashSet, VecDeque};

use chrono::{DateTime, Utc};
use pe_core::PeError;
use serde::{Deserialize, Serialize};

use crate::task::{Task, TaskId, TaskStatus};

/// A dependency edge: `task_id` depends on `depends_on_id`.
///
/// # Example
///
/// ```
/// use pe_tasks::dependency::{TaskDependency, DependencyType};
///
/// let dep = TaskDependency::new("deploy", "test", DependencyType::Blocks);
/// assert_eq!(dep.dependency_type, DependencyType::Blocks);
/// ```
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct TaskDependency {
    /// The task that is blocked/waiting.
    pub task_id: TaskId,
    /// The task it depends on (must complete first for Blocks).
    pub depends_on_id: TaskId,
    /// Nature of the dependency.
    pub dependency_type: DependencyType,
    /// When this dependency was created.
    pub created_at: DateTime<Utc>,
}

impl TaskDependency {
    /// Create a new dependency.
    #[must_use]
    pub fn new(
        task_id: impl Into<TaskId>,
        depends_on_id: impl Into<TaskId>,
        dependency_type: DependencyType,
    ) -> Self {
        Self {
            task_id: task_id.into(),
            depends_on_id: depends_on_id.into(),
            dependency_type,
            created_at: Utc::now(),
        }
    }
}

/// The nature of a dependency relationship.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[non_exhaustive]
pub enum DependencyType {
    /// Hard dependency: `depends_on` must complete before task can start.
    Blocks,
    /// Soft dependency: `depends_on` should complete, but task can proceed.
    Requires,
    /// Informational link: no execution constraint, just a relationship.
    Related,
}

/// Check whether adding `(task_id → depends_on_id)` would create a cycle.
///
/// Uses BFS upstream traversal: starting from `depends_on_id`, follow
/// all outgoing dependencies. If we reach `task_id`, there's a cycle.
///
/// # Arguments
///
/// * `task_id` — the task that would be blocked
/// * `depends_on_id` — the task it would depend on
/// * `existing_deps` — all current dependencies in the system
///
/// Returns `true` if adding this edge would create a cycle.
#[must_use]
pub fn would_create_cycle(
    task_id: &str,
    depends_on_id: &str,
    existing_deps: &[TaskDependency],
) -> bool {
    if task_id == depends_on_id {
        return true; // self-loop
    }

    // We want to add: task_id depends_on depends_on_id
    // Cycle exists if depends_on_id already (transitively) depends on task_id.
    // Build "depends on" adjacency: task → what it depends on
    let mut depends_on: HashMap<&str, Vec<&str>> = HashMap::new();
    for dep in existing_deps {
        depends_on
            .entry(dep.task_id.as_str())
            .or_default()
            .push(dep.depends_on_id.as_str());
    }

    // BFS from depends_on_id, following "depends on" edges.
    // If we reach task_id, then depends_on_id transitively depends on task_id,
    // so adding task_id → depends_on_id closes the loop.
    let mut visited = HashSet::new();
    let mut queue = VecDeque::new();
    queue.push_back(depends_on_id);
    visited.insert(depends_on_id);

    while let Some(current) = queue.pop_front() {
        if let Some(parents) = depends_on.get(current) {
            for &parent in parents {
                if parent == task_id {
                    return true; // found cycle
                }
                if visited.insert(parent) {
                    queue.push_back(parent);
                }
            }
        }
    }

    false
}

/// Check which tasks in `task_ids` are "ready" — no incomplete Blocks dependencies.
///
/// A task is ready if all tasks it depends on (with `DependencyType::Blocks`)
/// have status `Completed`.
pub(crate) fn validate_parent_link(candidate: &Task, tasks: &[Task]) -> Result<(), PeError> {
    let Some(parent_id) = candidate.parent_task_id.as_deref() else {
        return Ok(());
    };

    if parent_id == candidate.id {
        return Err(PeError::InvalidUpdate {
            details: format!("Task {} cannot be its own parent", candidate.id),
        });
    }

    active_task(parent_id, tasks).ok_or_else(|| PeError::NodeNotFound {
        node: parent_id.to_string(),
    })?;

    let mut seen = HashSet::new();
    let mut current_id = Some(parent_id);
    while let Some(task_id) = current_id {
        if task_id == candidate.id {
            return Err(PeError::InvalidUpdate {
                details: format!(
                    "Parent link {} -> {} would create a task hierarchy cycle",
                    candidate.id, parent_id
                ),
            });
        }
        if !seen.insert(task_id.to_string()) {
            return Err(PeError::InvalidUpdate {
                details: format!(
                    "Parent chain for task {} already contains a cycle",
                    candidate.id
                ),
            });
        }
        current_id = active_task(task_id, tasks).and_then(|task| task.parent_task_id.as_deref());
    }

    Ok(())
}

pub(crate) fn validate_dependency_endpoints(
    dep: &TaskDependency,
    tasks: &[Task],
) -> Result<(), PeError> {
    active_task(&dep.task_id, tasks).ok_or_else(|| PeError::NodeNotFound {
        node: dep.task_id.clone(),
    })?;
    active_task(&dep.depends_on_id, tasks).ok_or_else(|| PeError::NodeNotFound {
        node: dep.depends_on_id.clone(),
    })?;
    Ok(())
}

pub(crate) fn validate_dependency_unique(
    dep: &TaskDependency,
    existing_deps: &[TaskDependency],
) -> Result<(), PeError> {
    if existing_deps.iter().any(|existing| {
        existing.task_id == dep.task_id && existing.depends_on_id == dep.depends_on_id
    }) {
        return Err(PeError::InvalidUpdate {
            details: format!(
                "Dependency {} -> {} already exists",
                dep.task_id, dep.depends_on_id
            ),
        });
    }
    Ok(())
}

fn active_task<'a>(task_id: &str, tasks: &'a [Task]) -> Option<&'a Task> {
    tasks
        .iter()
        .find(|task| task.id == task_id && task.deleted_at.is_none())
}

#[must_use]
pub fn find_ready_tasks(
    task_ids: &[TaskId],
    deps: &[TaskDependency],
    task_statuses: &HashMap<TaskId, TaskStatus>,
) -> Vec<TaskId> {
    task_ids
        .iter()
        .filter(|tid| {
            let blockers: Vec<&TaskDependency> = deps
                .iter()
                .filter(|d| d.task_id == **tid && d.dependency_type == DependencyType::Blocks)
                .collect();

            blockers.iter().all(|b| {
                task_statuses
                    .get(&b.depends_on_id)
                    .is_some_and(|s| *s == TaskStatus::Completed)
            })
        })
        .cloned()
        .collect()
}

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

    #[test]
    fn test_no_cycle_simple() {
        let deps = vec![TaskDependency::new("B", "A", DependencyType::Blocks)];
        // Adding C→B: A←B←C — no cycle
        assert!(!would_create_cycle("C", "B", &deps));
    }

    #[test]
    fn test_self_loop_detected() {
        assert!(would_create_cycle("A", "A", &[]));
    }

    #[test]
    fn test_direct_cycle_detected() {
        let deps = vec![TaskDependency::new("B", "A", DependencyType::Blocks)];
        // Adding A→B when B→A exists: cycle!
        assert!(would_create_cycle("A", "B", &deps));
    }

    #[test]
    fn test_transitive_cycle_detected() {
        let deps = vec![
            TaskDependency::new("B", "A", DependencyType::Blocks),
            TaskDependency::new("C", "B", DependencyType::Blocks),
        ];
        // Adding A→C when C→B→A exists: cycle!
        assert!(would_create_cycle("A", "C", &deps));
    }

    #[test]
    fn test_no_cycle_parallel() {
        let deps = vec![
            TaskDependency::new("C", "A", DependencyType::Blocks),
            TaskDependency::new("C", "B", DependencyType::Blocks),
        ];
        // A and B are independent — adding B→A is fine
        assert!(!would_create_cycle("B", "A", &deps));
    }

    #[test]
    fn test_ready_tasks() {
        let deps = vec![
            TaskDependency::new("B", "A", DependencyType::Blocks),
            TaskDependency::new("C", "B", DependencyType::Blocks),
        ];
        let mut statuses = HashMap::new();
        statuses.insert("A".into(), TaskStatus::Completed);
        statuses.insert("B".into(), TaskStatus::Pending);
        statuses.insert("C".into(), TaskStatus::Pending);

        let ready = find_ready_tasks(&["A".into(), "B".into(), "C".into()], &deps, &statuses);
        // A is completed (always ready), B is ready (A is done), C is NOT ready (B not done)
        assert!(ready.contains(&"A".to_string()));
        assert!(ready.contains(&"B".to_string()));
        assert!(!ready.contains(&"C".to_string()));
    }

    #[test]
    fn test_related_deps_dont_block() {
        let deps = vec![TaskDependency::new("B", "A", DependencyType::Related)];
        let mut statuses = HashMap::new();
        statuses.insert("A".into(), TaskStatus::Pending); // A not done
        statuses.insert("B".into(), TaskStatus::Pending);

        let ready = find_ready_tasks(&["B".into()], &deps, &statuses);
        // B is ready because Related deps don't block
        assert!(ready.contains(&"B".to_string()));
    }
}