oximedia_workflow/
workflow.rs

1//! Workflow definition and management.
2
3use crate::error::{Result, WorkflowError};
4use crate::task::{Task, TaskId};
5use serde::{Deserialize, Serialize};
6use std::collections::{HashMap, HashSet};
7use uuid::Uuid;
8
9/// Unique workflow identifier.
10#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
11pub struct WorkflowId(Uuid);
12
13impl WorkflowId {
14    /// Create a new random workflow ID.
15    #[must_use]
16    pub fn new() -> Self {
17        Self(Uuid::new_v4())
18    }
19
20    /// Get the underlying UUID.
21    #[must_use]
22    pub const fn as_uuid(&self) -> &Uuid {
23        &self.0
24    }
25}
26
27impl Default for WorkflowId {
28    fn default() -> Self {
29        Self::new()
30    }
31}
32
33impl std::fmt::Display for WorkflowId {
34    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
35        write!(f, "{}", self.0)
36    }
37}
38
39impl From<Uuid> for WorkflowId {
40    fn from(uuid: Uuid) -> Self {
41        Self(uuid)
42    }
43}
44
45/// Workflow state.
46#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
47pub enum WorkflowState {
48    /// Workflow is created but not started.
49    Created,
50    /// Workflow is scheduled to run.
51    Scheduled,
52    /// Workflow is running.
53    Running,
54    /// Workflow is paused.
55    Paused,
56    /// Workflow completed successfully.
57    Completed,
58    /// Workflow failed.
59    Failed,
60    /// Workflow was cancelled.
61    Cancelled,
62}
63
64impl WorkflowState {
65    /// Check if workflow is in a terminal state.
66    #[must_use]
67    pub const fn is_terminal(&self) -> bool {
68        matches!(self, Self::Completed | Self::Failed | Self::Cancelled)
69    }
70
71    /// Check if workflow is active.
72    #[must_use]
73    pub const fn is_active(&self) -> bool {
74        matches!(self, Self::Running)
75    }
76}
77
78/// Edge between tasks in the workflow DAG.
79#[derive(Debug, Clone, Serialize, Deserialize)]
80pub struct Edge {
81    /// Source task ID.
82    pub from: TaskId,
83    /// Target task ID.
84    pub to: TaskId,
85    /// Edge condition (optional).
86    pub condition: Option<String>,
87}
88
89impl Edge {
90    /// Create a new edge.
91    #[must_use]
92    pub const fn new(from: TaskId, to: TaskId) -> Self {
93        Self {
94            from,
95            to,
96            condition: None,
97        }
98    }
99
100    /// Create an edge with a condition.
101    #[must_use]
102    pub fn with_condition(from: TaskId, to: TaskId, condition: String) -> Self {
103        Self {
104            from,
105            to,
106            condition: Some(condition),
107        }
108    }
109}
110
111/// Workflow configuration.
112#[derive(Debug, Clone, Serialize, Deserialize)]
113pub struct WorkflowConfig {
114    /// Maximum concurrent tasks.
115    #[serde(default = "default_max_concurrent")]
116    pub max_concurrent_tasks: usize,
117    /// Global timeout for workflow.
118    #[serde(default)]
119    pub global_timeout: Option<std::time::Duration>,
120    /// Whether to stop on first error.
121    #[serde(default)]
122    pub fail_fast: bool,
123    /// Whether to continue on task failure.
124    #[serde(default)]
125    pub continue_on_error: bool,
126    /// Variables available to all tasks.
127    #[serde(default)]
128    pub variables: HashMap<String, serde_json::Value>,
129}
130
131fn default_max_concurrent() -> usize {
132    4
133}
134
135impl Default for WorkflowConfig {
136    fn default() -> Self {
137        Self {
138            max_concurrent_tasks: default_max_concurrent(),
139            global_timeout: None,
140            fail_fast: false,
141            continue_on_error: false,
142            variables: HashMap::new(),
143        }
144    }
145}
146
147/// Workflow definition.
148#[derive(Debug, Clone, Serialize, Deserialize)]
149pub struct Workflow {
150    /// Unique workflow identifier.
151    pub id: WorkflowId,
152    /// Workflow name.
153    pub name: String,
154    /// Workflow description.
155    #[serde(default)]
156    pub description: String,
157    /// Tasks in the workflow.
158    pub tasks: HashMap<TaskId, Task>,
159    /// Edges defining task dependencies.
160    pub edges: Vec<Edge>,
161    /// Workflow configuration.
162    #[serde(default)]
163    pub config: WorkflowConfig,
164    /// Current state.
165    #[serde(default = "default_workflow_state")]
166    pub state: WorkflowState,
167    /// Workflow metadata.
168    #[serde(default)]
169    pub metadata: HashMap<String, String>,
170}
171
172fn default_workflow_state() -> WorkflowState {
173    WorkflowState::Created
174}
175
176impl Workflow {
177    /// Create a new workflow.
178    #[must_use]
179    pub fn new(name: impl Into<String>) -> Self {
180        Self {
181            id: WorkflowId::new(),
182            name: name.into(),
183            description: String::new(),
184            tasks: HashMap::new(),
185            edges: Vec::new(),
186            config: WorkflowConfig::default(),
187            state: WorkflowState::Created,
188            metadata: HashMap::new(),
189        }
190    }
191
192    /// Add a task to the workflow.
193    pub fn add_task(&mut self, task: Task) -> TaskId {
194        let task_id = task.id;
195        self.tasks.insert(task_id, task);
196        task_id
197    }
198
199    /// Add an edge between tasks.
200    pub fn add_edge(&mut self, from: TaskId, to: TaskId) -> Result<()> {
201        // Validate that both tasks exist
202        if !self.tasks.contains_key(&from) {
203            return Err(WorkflowError::TaskNotFound(from.to_string()));
204        }
205        if !self.tasks.contains_key(&to) {
206            return Err(WorkflowError::TaskNotFound(to.to_string()));
207        }
208
209        self.edges.push(Edge::new(from, to));
210        Ok(())
211    }
212
213    /// Add a conditional edge.
214    pub fn add_conditional_edge(
215        &mut self,
216        from: TaskId,
217        to: TaskId,
218        condition: String,
219    ) -> Result<()> {
220        if !self.tasks.contains_key(&from) {
221            return Err(WorkflowError::TaskNotFound(from.to_string()));
222        }
223        if !self.tasks.contains_key(&to) {
224            return Err(WorkflowError::TaskNotFound(to.to_string()));
225        }
226
227        self.edges.push(Edge::with_condition(from, to, condition));
228        Ok(())
229    }
230
231    /// Get task by ID.
232    #[must_use]
233    pub fn get_task(&self, task_id: &TaskId) -> Option<&Task> {
234        self.tasks.get(task_id)
235    }
236
237    /// Get mutable task by ID.
238    pub fn get_task_mut(&mut self, task_id: &TaskId) -> Option<&mut Task> {
239        self.tasks.get_mut(task_id)
240    }
241
242    /// Get all tasks.
243    #[must_use]
244    pub fn tasks(&self) -> impl Iterator<Item = &Task> {
245        self.tasks.values()
246    }
247
248    /// Get dependencies for a task.
249    #[must_use]
250    pub fn get_dependencies(&self, task_id: &TaskId) -> Vec<TaskId> {
251        self.edges
252            .iter()
253            .filter(|e| &e.to == task_id)
254            .map(|e| e.from)
255            .collect()
256    }
257
258    /// Get dependents for a task.
259    #[must_use]
260    pub fn get_dependents(&self, task_id: &TaskId) -> Vec<TaskId> {
261        self.edges
262            .iter()
263            .filter(|e| &e.from == task_id)
264            .map(|e| e.to)
265            .collect()
266    }
267
268    /// Get root tasks (tasks with no dependencies).
269    #[must_use]
270    pub fn get_root_tasks(&self) -> Vec<TaskId> {
271        let has_incoming: HashSet<_> = self.edges.iter().map(|e| e.to).collect();
272        self.tasks
273            .keys()
274            .filter(|id| !has_incoming.contains(id))
275            .copied()
276            .collect()
277    }
278
279    /// Get leaf tasks (tasks with no dependents).
280    #[must_use]
281    pub fn get_leaf_tasks(&self) -> Vec<TaskId> {
282        let has_outgoing: HashSet<_> = self.edges.iter().map(|e| e.from).collect();
283        self.tasks
284            .keys()
285            .filter(|id| !has_outgoing.contains(id))
286            .copied()
287            .collect()
288    }
289
290    /// Validate workflow structure.
291    pub fn validate(&self) -> Result<()> {
292        // Check for cycles
293        if self.has_cycle() {
294            return Err(WorkflowError::CycleDetected);
295        }
296
297        // Validate all edge references
298        for edge in &self.edges {
299            if !self.tasks.contains_key(&edge.from) {
300                return Err(WorkflowError::TaskNotFound(edge.from.to_string()));
301            }
302            if !self.tasks.contains_key(&edge.to) {
303                return Err(WorkflowError::TaskNotFound(edge.to.to_string()));
304            }
305        }
306
307        Ok(())
308    }
309
310    /// Check if workflow has a cycle.
311    #[must_use]
312    pub fn has_cycle(&self) -> bool {
313        let mut visited = HashSet::new();
314        let mut rec_stack = HashSet::new();
315
316        for task_id in self.tasks.keys() {
317            if self.has_cycle_util(*task_id, &mut visited, &mut rec_stack) {
318                return true;
319            }
320        }
321
322        false
323    }
324
325    fn has_cycle_util(
326        &self,
327        task_id: TaskId,
328        visited: &mut HashSet<TaskId>,
329        rec_stack: &mut HashSet<TaskId>,
330    ) -> bool {
331        if rec_stack.contains(&task_id) {
332            return true;
333        }
334
335        if visited.contains(&task_id) {
336            return false;
337        }
338
339        visited.insert(task_id);
340        rec_stack.insert(task_id);
341
342        for dependent in self.get_dependents(&task_id) {
343            if self.has_cycle_util(dependent, visited, rec_stack) {
344                return true;
345            }
346        }
347
348        rec_stack.remove(&task_id);
349        false
350    }
351
352    /// Get topological order of tasks.
353    pub fn topological_sort(&self) -> Result<Vec<TaskId>> {
354        if self.has_cycle() {
355            return Err(WorkflowError::CycleDetected);
356        }
357
358        let mut result = Vec::new();
359        let mut visited = HashSet::new();
360
361        for task_id in self.tasks.keys() {
362            self.topological_sort_util(*task_id, &mut visited, &mut result);
363        }
364
365        result.reverse();
366        Ok(result)
367    }
368
369    fn topological_sort_util(
370        &self,
371        task_id: TaskId,
372        visited: &mut HashSet<TaskId>,
373        result: &mut Vec<TaskId>,
374    ) {
375        if visited.contains(&task_id) {
376            return;
377        }
378
379        visited.insert(task_id);
380
381        for dependent in self.get_dependents(&task_id) {
382            self.topological_sort_util(dependent, visited, result);
383        }
384
385        result.push(task_id);
386    }
387
388    /// Set workflow description.
389    pub fn with_description(mut self, description: impl Into<String>) -> Self {
390        self.description = description.into();
391        self
392    }
393
394    /// Set workflow configuration.
395    #[must_use]
396    pub fn with_config(mut self, config: WorkflowConfig) -> Self {
397        self.config = config;
398        self
399    }
400
401    /// Add metadata.
402    pub fn with_metadata(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
403        self.metadata.insert(key.into(), value.into());
404        self
405    }
406}
407
408#[cfg(test)]
409mod tests {
410    use super::*;
411    use crate::task::TaskType;
412    use std::time::Duration;
413
414    fn create_test_task(name: &str) -> Task {
415        Task::new(
416            name,
417            TaskType::Wait {
418                duration: Duration::from_secs(1),
419            },
420        )
421    }
422
423    #[test]
424    fn test_workflow_creation() {
425        let workflow = Workflow::new("test-workflow");
426        assert_eq!(workflow.name, "test-workflow");
427        assert_eq!(workflow.state, WorkflowState::Created);
428        assert!(workflow.tasks.is_empty());
429    }
430
431    #[test]
432    fn test_add_task() {
433        let mut workflow = Workflow::new("test");
434        let task = create_test_task("task1");
435        let task_id = workflow.add_task(task);
436        assert_eq!(workflow.tasks.len(), 1);
437        assert!(workflow.get_task(&task_id).is_some());
438    }
439
440    #[test]
441    fn test_add_edge() {
442        let mut workflow = Workflow::new("test");
443        let task1 = create_test_task("task1");
444        let task2 = create_test_task("task2");
445        let id1 = workflow.add_task(task1);
446        let id2 = workflow.add_task(task2);
447
448        assert!(workflow.add_edge(id1, id2).is_ok());
449        assert_eq!(workflow.edges.len(), 1);
450    }
451
452    #[test]
453    fn test_add_edge_invalid_task() {
454        let mut workflow = Workflow::new("test");
455        let task1 = create_test_task("task1");
456        let id1 = workflow.add_task(task1);
457        let invalid_id = TaskId::new();
458
459        assert!(workflow.add_edge(id1, invalid_id).is_err());
460    }
461
462    #[test]
463    fn test_get_dependencies() {
464        let mut workflow = Workflow::new("test");
465        let task1 = create_test_task("task1");
466        let task2 = create_test_task("task2");
467        let task3 = create_test_task("task3");
468
469        let id1 = workflow.add_task(task1);
470        let id2 = workflow.add_task(task2);
471        let id3 = workflow.add_task(task3);
472
473        workflow.add_edge(id1, id3).expect("should succeed in test");
474        workflow.add_edge(id2, id3).expect("should succeed in test");
475
476        let deps = workflow.get_dependencies(&id3);
477        assert_eq!(deps.len(), 2);
478        assert!(deps.contains(&id1));
479        assert!(deps.contains(&id2));
480    }
481
482    #[test]
483    fn test_get_root_tasks() {
484        let mut workflow = Workflow::new("test");
485        let task1 = create_test_task("task1");
486        let task2 = create_test_task("task2");
487        let task3 = create_test_task("task3");
488
489        let id1 = workflow.add_task(task1);
490        let id2 = workflow.add_task(task2);
491        let id3 = workflow.add_task(task3);
492
493        workflow.add_edge(id1, id3).expect("should succeed in test");
494        workflow.add_edge(id2, id3).expect("should succeed in test");
495
496        let roots = workflow.get_root_tasks();
497        assert_eq!(roots.len(), 2);
498        assert!(roots.contains(&id1));
499        assert!(roots.contains(&id2));
500    }
501
502    #[test]
503    fn test_get_leaf_tasks() {
504        let mut workflow = Workflow::new("test");
505        let task1 = create_test_task("task1");
506        let task2 = create_test_task("task2");
507        let task3 = create_test_task("task3");
508
509        let id1 = workflow.add_task(task1);
510        let id2 = workflow.add_task(task2);
511        let id3 = workflow.add_task(task3);
512
513        workflow.add_edge(id1, id2).expect("should succeed in test");
514        workflow.add_edge(id1, id3).expect("should succeed in test");
515
516        let leaves = workflow.get_leaf_tasks();
517        assert_eq!(leaves.len(), 2);
518        assert!(leaves.contains(&id2));
519        assert!(leaves.contains(&id3));
520    }
521
522    #[test]
523    fn test_has_cycle_no_cycle() {
524        let mut workflow = Workflow::new("test");
525        let task1 = create_test_task("task1");
526        let task2 = create_test_task("task2");
527        let task3 = create_test_task("task3");
528
529        let id1 = workflow.add_task(task1);
530        let id2 = workflow.add_task(task2);
531        let id3 = workflow.add_task(task3);
532
533        workflow.add_edge(id1, id2).expect("should succeed in test");
534        workflow.add_edge(id2, id3).expect("should succeed in test");
535
536        assert!(!workflow.has_cycle());
537    }
538
539    #[test]
540    fn test_has_cycle_with_cycle() {
541        let mut workflow = Workflow::new("test");
542        let task1 = create_test_task("task1");
543        let task2 = create_test_task("task2");
544
545        let id1 = workflow.add_task(task1);
546        let id2 = workflow.add_task(task2);
547
548        workflow.add_edge(id1, id2).expect("should succeed in test");
549        workflow.add_edge(id2, id1).expect("should succeed in test");
550
551        assert!(workflow.has_cycle());
552    }
553
554    #[test]
555    fn test_validate_success() {
556        let mut workflow = Workflow::new("test");
557        let task1 = create_test_task("task1");
558        let task2 = create_test_task("task2");
559
560        let id1 = workflow.add_task(task1);
561        let id2 = workflow.add_task(task2);
562        workflow.add_edge(id1, id2).expect("should succeed in test");
563
564        assert!(workflow.validate().is_ok());
565    }
566
567    #[test]
568    fn test_validate_cycle() {
569        let mut workflow = Workflow::new("test");
570        let task1 = create_test_task("task1");
571        let task2 = create_test_task("task2");
572
573        let id1 = workflow.add_task(task1);
574        let id2 = workflow.add_task(task2);
575        workflow.add_edge(id1, id2).expect("should succeed in test");
576        workflow.add_edge(id2, id1).expect("should succeed in test");
577
578        assert!(workflow.validate().is_err());
579    }
580
581    #[test]
582    fn test_topological_sort() {
583        let mut workflow = Workflow::new("test");
584        let task1 = create_test_task("task1");
585        let task2 = create_test_task("task2");
586        let task3 = create_test_task("task3");
587
588        let id1 = workflow.add_task(task1);
589        let id2 = workflow.add_task(task2);
590        let id3 = workflow.add_task(task3);
591
592        workflow.add_edge(id1, id2).expect("should succeed in test");
593        workflow.add_edge(id2, id3).expect("should succeed in test");
594
595        let sorted = workflow.topological_sort().expect("should succeed in test");
596        assert_eq!(sorted.len(), 3);
597
598        let pos1 = sorted
599            .iter()
600            .position(|&x| x == id1)
601            .expect("should succeed in test");
602        let pos2 = sorted
603            .iter()
604            .position(|&x| x == id2)
605            .expect("should succeed in test");
606        let pos3 = sorted
607            .iter()
608            .position(|&x| x == id3)
609            .expect("should succeed in test");
610
611        assert!(pos1 < pos2);
612        assert!(pos2 < pos3);
613    }
614
615    #[test]
616    fn test_workflow_state_terminal() {
617        assert!(WorkflowState::Completed.is_terminal());
618        assert!(WorkflowState::Failed.is_terminal());
619        assert!(!WorkflowState::Running.is_terminal());
620    }
621
622    #[test]
623    fn test_conditional_edge() {
624        let mut workflow = Workflow::new("test");
625        let task1 = create_test_task("task1");
626        let task2 = create_test_task("task2");
627
628        let id1 = workflow.add_task(task1);
629        let id2 = workflow.add_task(task2);
630
631        assert!(workflow
632            .add_conditional_edge(id1, id2, "result == true".to_string())
633            .is_ok());
634        assert_eq!(workflow.edges.len(), 1);
635        assert!(workflow.edges[0].condition.is_some());
636    }
637}
oximedia_workflow/workflow.rs

oximedia_workflow/
workflow.rs
