forge_runtime/workflow/
state.rs

1use chrono::{DateTime, Utc};
2use forge_core::workflow::{StepStatus, WorkflowStatus};
3use uuid::Uuid;
4
5/// A workflow run record in the database.
6#[derive(Debug, Clone)]
7pub struct WorkflowRecord {
8    /// Unique workflow run ID.
9    pub id: Uuid,
10    /// Workflow name.
11    pub workflow_name: String,
12    /// Workflow version.
13    pub version: u32,
14    /// Input data as JSON.
15    pub input: serde_json::Value,
16    /// Output data as JSON (if completed).
17    pub output: Option<serde_json::Value>,
18    /// Current status.
19    pub status: WorkflowStatus,
20    /// Current step name.
21    pub current_step: Option<String>,
22    /// Step results as JSON map.
23    pub step_results: serde_json::Value,
24    /// When the workflow started.
25    pub started_at: DateTime<Utc>,
26    /// When the workflow completed.
27    pub completed_at: Option<DateTime<Utc>>,
28    /// Error message if failed.
29    pub error: Option<String>,
30    /// Trace ID for distributed tracing.
31    pub trace_id: Option<String>,
32}
33
34impl WorkflowRecord {
35    /// Create a new workflow record.
36    pub fn new(workflow_name: impl Into<String>, version: u32, input: serde_json::Value) -> Self {
37        Self {
38            id: Uuid::new_v4(),
39            workflow_name: workflow_name.into(),
40            version,
41            input,
42            output: None,
43            status: WorkflowStatus::Created,
44            current_step: None,
45            step_results: serde_json::json!({}),
46            started_at: Utc::now(),
47            completed_at: None,
48            error: None,
49            trace_id: None,
50        }
51    }
52
53    /// Set trace ID.
54    pub fn with_trace_id(mut self, trace_id: impl Into<String>) -> Self {
55        self.trace_id = Some(trace_id.into());
56        self
57    }
58
59    /// Mark as running.
60    pub fn start(&mut self) {
61        self.status = WorkflowStatus::Running;
62    }
63
64    /// Mark as completed.
65    pub fn complete(&mut self, output: serde_json::Value) {
66        self.status = WorkflowStatus::Completed;
67        self.output = Some(output);
68        self.completed_at = Some(Utc::now());
69    }
70
71    /// Mark as failed.
72    pub fn fail(&mut self, error: impl Into<String>) {
73        self.status = WorkflowStatus::Failed;
74        self.error = Some(error.into());
75        self.completed_at = Some(Utc::now());
76    }
77
78    /// Mark as compensating.
79    pub fn compensating(&mut self) {
80        self.status = WorkflowStatus::Compensating;
81    }
82
83    /// Mark as compensated.
84    pub fn compensated(&mut self) {
85        self.status = WorkflowStatus::Compensated;
86        self.completed_at = Some(Utc::now());
87    }
88
89    /// Update current step.
90    pub fn set_current_step(&mut self, step: impl Into<String>) {
91        self.current_step = Some(step.into());
92    }
93
94    /// Add step result.
95    pub fn add_step_result(&mut self, step_name: &str, result: serde_json::Value) {
96        if let Some(obj) = self.step_results.as_object_mut() {
97            obj.insert(step_name.to_string(), result);
98        }
99    }
100}
101
102/// A workflow step record in the database.
103#[derive(Debug, Clone)]
104pub struct WorkflowStepRecord {
105    /// Step record ID.
106    pub id: Uuid,
107    /// Parent workflow run ID.
108    pub workflow_run_id: Uuid,
109    /// Step name.
110    pub step_name: String,
111    /// Step status.
112    pub status: StepStatus,
113    /// Step result as JSON.
114    pub result: Option<serde_json::Value>,
115    /// Error message if failed.
116    pub error: Option<String>,
117    /// When the step started.
118    pub started_at: Option<DateTime<Utc>>,
119    /// When the step completed.
120    pub completed_at: Option<DateTime<Utc>>,
121}
122
123impl WorkflowStepRecord {
124    /// Create a new step record.
125    pub fn new(workflow_run_id: Uuid, step_name: impl Into<String>) -> Self {
126        Self {
127            id: Uuid::new_v4(),
128            workflow_run_id,
129            step_name: step_name.into(),
130            status: StepStatus::Pending,
131            result: None,
132            error: None,
133            started_at: None,
134            completed_at: None,
135        }
136    }
137
138    /// Mark as running.
139    pub fn start(&mut self) {
140        self.status = StepStatus::Running;
141        self.started_at = Some(Utc::now());
142    }
143
144    /// Mark as completed.
145    pub fn complete(&mut self, result: serde_json::Value) {
146        self.status = StepStatus::Completed;
147        self.result = Some(result);
148        self.completed_at = Some(Utc::now());
149    }
150
151    /// Mark as failed.
152    pub fn fail(&mut self, error: impl Into<String>) {
153        self.status = StepStatus::Failed;
154        self.error = Some(error.into());
155        self.completed_at = Some(Utc::now());
156    }
157
158    /// Mark as compensated.
159    pub fn compensate(&mut self) {
160        self.status = StepStatus::Compensated;
161    }
162}
163
164#[cfg(test)]
165mod tests {
166    use super::*;
167
168    #[test]
169    fn test_workflow_record_creation() {
170        let record = WorkflowRecord::new("test_workflow", 1, serde_json::json!({}));
171        assert_eq!(record.workflow_name, "test_workflow");
172        assert_eq!(record.version, 1);
173        assert_eq!(record.status, WorkflowStatus::Created);
174    }
175
176    #[test]
177    fn test_workflow_record_transitions() {
178        let mut record = WorkflowRecord::new("test", 1, serde_json::json!({}));
179
180        record.start();
181        assert_eq!(record.status, WorkflowStatus::Running);
182
183        record.complete(serde_json::json!({"result": "ok"}));
184        assert_eq!(record.status, WorkflowStatus::Completed);
185        assert!(record.completed_at.is_some());
186    }
187
188    #[test]
189    fn test_workflow_step_record() {
190        let workflow_id = Uuid::new_v4();
191        let mut step = WorkflowStepRecord::new(workflow_id, "step1");
192
193        assert_eq!(step.step_name, "step1");
194        assert_eq!(step.status, StepStatus::Pending);
195
196        step.start();
197        assert_eq!(step.status, StepStatus::Running);
198
199        step.complete(serde_json::json!({}));
200        assert_eq!(step.status, StepStatus::Completed);
201    }
202}
forge_runtime/workflow/state.rs

forge_runtime/workflow/
state.rs
