wfe-core 1.0.0

Core traits, models, builder, and executor for the WFE workflow engine
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141

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

use super::execution_pointer::ExecutionPointer;
use super::status::{PointerStatus, WorkflowStatus};

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WorkflowInstance {
    pub id: String,
    pub workflow_definition_id: String,
    pub version: u32,
    pub description: Option<String>,
    pub reference: Option<String>,
    pub execution_pointers: Vec<ExecutionPointer>,
    pub next_execution: Option<i64>,
    pub status: WorkflowStatus,
    pub data: serde_json::Value,
    pub create_time: DateTime<Utc>,
    pub complete_time: Option<DateTime<Utc>>,
}

impl WorkflowInstance {
    pub fn new(workflow_definition_id: impl Into<String>, version: u32, data: serde_json::Value) -> Self {
        Self {
            id: uuid::Uuid::new_v4().to_string(),
            workflow_definition_id: workflow_definition_id.into(),
            version,
            description: None,
            reference: None,
            execution_pointers: Vec::new(),
            next_execution: Some(0),
            status: WorkflowStatus::Runnable,
            data,
            create_time: Utc::now(),
            complete_time: None,
        }
    }

    /// Check if all execution pointers in a given scope have completed.
    pub fn is_branch_complete(&self, scope: &[String]) -> bool {
        self.execution_pointers
            .iter()
            .filter(|p| p.scope == scope)
            .all(|p| {
                matches!(
                    p.status,
                    PointerStatus::Complete
                        | PointerStatus::Compensated
                        | PointerStatus::Cancelled
                        | PointerStatus::Failed
                )
            })
    }
}

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

    #[test]
    fn new_instance_defaults() {
        let instance = WorkflowInstance::new("test-workflow", 1, serde_json::json!({}));
        assert_eq!(instance.workflow_definition_id, "test-workflow");
        assert_eq!(instance.version, 1);
        assert_eq!(instance.status, WorkflowStatus::Runnable);
        assert_eq!(instance.next_execution, Some(0));
        assert!(instance.execution_pointers.is_empty());
        assert!(instance.complete_time.is_none());
    }

    #[test]
    fn is_branch_complete_empty_scope_returns_true() {
        let instance = WorkflowInstance::new("test", 1, serde_json::json!({}));
        assert!(instance.is_branch_complete(&[]));
    }

    #[test]
    fn is_branch_complete_all_complete() {
        let scope = vec!["parent-1".to_string()];
        let mut instance = WorkflowInstance::new("test", 1, serde_json::json!({}));

        let mut p1 = ExecutionPointer::new(0);
        p1.scope = scope.clone();
        p1.status = PointerStatus::Complete;

        let mut p2 = ExecutionPointer::new(1);
        p2.scope = scope.clone();
        p2.status = PointerStatus::Compensated;

        instance.execution_pointers = vec![p1, p2];
        assert!(instance.is_branch_complete(&scope));
    }

    #[test]
    fn is_branch_complete_with_active_pointer() {
        let scope = vec!["parent-1".to_string()];
        let mut instance = WorkflowInstance::new("test", 1, serde_json::json!({}));

        let mut p1 = ExecutionPointer::new(0);
        p1.scope = scope.clone();
        p1.status = PointerStatus::Complete;

        let mut p2 = ExecutionPointer::new(1);
        p2.scope = scope.clone();
        p2.status = PointerStatus::Running;

        instance.execution_pointers = vec![p1, p2];
        assert!(!instance.is_branch_complete(&scope));
    }

    #[test]
    fn is_branch_complete_ignores_different_scope() {
        let scope_a = vec!["parent-a".to_string()];
        let scope_b = vec!["parent-b".to_string()];
        let mut instance = WorkflowInstance::new("test", 1, serde_json::json!({}));

        let mut p1 = ExecutionPointer::new(0);
        p1.scope = scope_a.clone();
        p1.status = PointerStatus::Complete;

        let mut p2 = ExecutionPointer::new(1);
        p2.scope = scope_b.clone();
        p2.status = PointerStatus::Running;

        instance.execution_pointers = vec![p1, p2];
        assert!(instance.is_branch_complete(&scope_a));
    }

    #[test]
    fn serde_round_trip() {
        let instance = WorkflowInstance::new("my-workflow", 2, serde_json::json!({"key": "value"}));
        let json = serde_json::to_string(&instance).unwrap();
        let deserialized: WorkflowInstance = serde_json::from_str(&json).unwrap();
        assert_eq!(instance.id, deserialized.id);
        assert_eq!(instance.workflow_definition_id, deserialized.workflow_definition_id);
        assert_eq!(instance.version, deserialized.version);
        assert_eq!(instance.status, deserialized.status);
        assert_eq!(instance.data, deserialized.data);
    }
}