wfe_core/builder/
step_builder.rs

1use crate::models::{ErrorBehavior, ExecutionResult};
2use crate::primitives;
3use crate::traits::step::{StepBody, WorkflowData};
4
5use super::inline_step::InlineStep;
6use super::workflow_builder::WorkflowBuilder;
7
8/// Builder for configuring a single step in the workflow.
9///
10/// Owns the WorkflowBuilder, consuming self on each method call.
11/// This avoids all lifetime/borrow issues.
12pub struct StepBuilder<D: WorkflowData> {
13    builder: WorkflowBuilder<D>,
14    step_id: usize,
15}
16
17/// Builder for parallel branches.
18pub struct ParallelBuilder<D: WorkflowData> {
19    builder: WorkflowBuilder<D>,
20    container_id: usize,
21}
22
23impl<D: WorkflowData> StepBuilder<D> {
24    pub(crate) fn new(builder: WorkflowBuilder<D>, step_id: usize) -> Self {
25        Self { builder, step_id }
26    }
27
28    /// Set the display name of the current step.
29    pub fn name(mut self, name: &str) -> Self {
30        self.builder.steps[self.step_id].name = Some(name.to_string());
31        self
32    }
33
34    /// Set an external ID for forward references.
35    pub fn id(mut self, external_id: &str) -> Self {
36        self.builder.steps[self.step_id].external_id = Some(external_id.to_string());
37        self
38    }
39
40    /// Set the error handling behavior for this step.
41    pub fn on_error(mut self, behavior: ErrorBehavior) -> Self {
42        self.builder.steps[self.step_id].error_behavior = Some(behavior);
43        self
44    }
45
46    /// Add a compensation step for saga rollback.
47    pub fn compensate_with<C: StepBody + Default + 'static>(mut self) -> Self {
48        let comp_id = self.builder.add_step(std::any::type_name::<C>());
49        self.builder.steps[self.step_id].compensation_step_id = Some(comp_id);
50        self
51    }
52
53    /// Chain the next step. Wires an outcome from the current step to the new one.
54    pub fn then<S: StepBody + Default + 'static>(mut self) -> StepBuilder<D> {
55        let next_id = self.builder.add_step(std::any::type_name::<S>());
56        self.builder.wire_outcome(self.step_id, next_id, None);
57        self.builder.last_step = Some(next_id);
58        StepBuilder::new(self.builder, next_id)
59    }
60
61    /// Chain an inline function step.
62    pub fn then_fn(mut self, f: impl Fn() -> ExecutionResult + Send + Sync + 'static) -> StepBuilder<D> {
63        let next_id = self.builder.add_step(std::any::type_name::<InlineStep>());
64        self.builder.wire_outcome(self.step_id, next_id, None);
65        self.builder.last_step = Some(next_id);
66        self.builder.inline_closures.insert(next_id, Box::new(f));
67        StepBuilder::new(self.builder, next_id)
68    }
69
70    /// Insert a WaitFor step.
71    pub fn wait_for(mut self, event_name: &str, event_key: &str) -> StepBuilder<D> {
72        let next_id = self.builder.add_step(std::any::type_name::<primitives::wait_for::WaitForStep>());
73        self.builder.wire_outcome(self.step_id, next_id, None);
74        self.builder.last_step = Some(next_id);
75        self.builder.steps[next_id].step_config = Some(serde_json::json!({
76            "event_name": event_name,
77            "event_key": event_key,
78        }));
79        StepBuilder::new(self.builder, next_id)
80    }
81
82    /// Insert a Delay step.
83    pub fn delay(mut self, duration: std::time::Duration) -> StepBuilder<D> {
84        let next_id = self.builder.add_step(std::any::type_name::<primitives::delay::DelayStep>());
85        self.builder.wire_outcome(self.step_id, next_id, None);
86        self.builder.last_step = Some(next_id);
87        self.builder.steps[next_id].step_config = Some(serde_json::json!({
88            "duration_millis": duration.as_millis() as u64,
89        }));
90        StepBuilder::new(self.builder, next_id)
91    }
92
93    /// Insert an If container step with child steps built by the closure.
94    /// The type parameter S is the step type used for the If condition evaluation.
95    pub fn if_do<S: StepBody + Default + 'static>(
96        mut self,
97        build_children: impl FnOnce(&mut WorkflowBuilder<D>),
98    ) -> StepBuilder<D> {
99        let if_id = self.builder.add_step(std::any::type_name::<primitives::if_step::IfStep>());
100        self.builder.wire_outcome(self.step_id, if_id, None);
101
102        // Build children
103        let before_count = self.builder.steps.len();
104        build_children(&mut self.builder);
105        let after_count = self.builder.steps.len();
106
107        // Register children with the If step
108        for child_id in before_count..after_count {
109            self.builder.add_child(if_id, child_id);
110        }
111
112        self.builder.last_step = Some(if_id);
113        StepBuilder::new(self.builder, if_id)
114    }
115
116    /// Insert a While container step.
117    pub fn while_do<S: StepBody + Default + 'static>(
118        mut self,
119        build_children: impl FnOnce(&mut WorkflowBuilder<D>),
120    ) -> StepBuilder<D> {
121        let while_id = self.builder.add_step(std::any::type_name::<primitives::while_step::WhileStep>());
122        self.builder.wire_outcome(self.step_id, while_id, None);
123
124        let before_count = self.builder.steps.len();
125        build_children(&mut self.builder);
126        let after_count = self.builder.steps.len();
127
128        for child_id in before_count..after_count {
129            self.builder.add_child(while_id, child_id);
130        }
131
132        self.builder.last_step = Some(while_id);
133        StepBuilder::new(self.builder, while_id)
134    }
135
136    /// Insert a ForEach container step.
137    pub fn for_each<S: StepBody + Default + 'static>(
138        mut self,
139        build_children: impl FnOnce(&mut WorkflowBuilder<D>),
140    ) -> StepBuilder<D> {
141        let fe_id = self.builder.add_step(std::any::type_name::<primitives::foreach_step::ForEachStep>());
142        self.builder.wire_outcome(self.step_id, fe_id, None);
143
144        let before_count = self.builder.steps.len();
145        build_children(&mut self.builder);
146        let after_count = self.builder.steps.len();
147
148        for child_id in before_count..after_count {
149            self.builder.add_child(fe_id, child_id);
150        }
151
152        self.builder.last_step = Some(fe_id);
153        StepBuilder::new(self.builder, fe_id)
154    }
155
156    /// Insert a Saga container step with child steps.
157    pub fn saga(
158        mut self,
159        build_children: impl FnOnce(&mut WorkflowBuilder<D>),
160    ) -> StepBuilder<D> {
161        let saga_id = self.builder.add_step(std::any::type_name::<primitives::saga_container::SagaContainerStep>());
162        self.builder.steps[saga_id].saga = true;
163        self.builder.wire_outcome(self.step_id, saga_id, None);
164
165        let before_count = self.builder.steps.len();
166        build_children(&mut self.builder);
167        let after_count = self.builder.steps.len();
168
169        for child_id in before_count..after_count {
170            self.builder.add_child(saga_id, child_id);
171        }
172
173        self.builder.last_step = Some(saga_id);
174        StepBuilder::new(self.builder, saga_id)
175    }
176
177    /// Start a parallel block.
178    pub fn parallel(
179        mut self,
180        build_branches: impl FnOnce(ParallelBuilder<D>) -> ParallelBuilder<D>,
181    ) -> StepBuilder<D> {
182        let seq_id = self.builder.add_step(std::any::type_name::<primitives::sequence::SequenceStep>());
183        self.builder.wire_outcome(self.step_id, seq_id, None);
184
185        let pb = ParallelBuilder {
186            builder: self.builder,
187            container_id: seq_id,
188        };
189        let pb = build_branches(pb);
190        let mut builder = pb.builder;
191        builder.last_step = Some(seq_id);
192        StepBuilder::new(builder, seq_id)
193    }
194
195    /// Mark this step as the terminal step (no further outcomes).
196    pub fn end_workflow(self) -> WorkflowBuilder<D> {
197        self.builder
198    }
199
200    /// Access the compiled definition directly (shortcut for end_workflow().build()).
201    pub fn build(self, id: impl Into<String>, version: u32) -> crate::models::WorkflowDefinition {
202        self.builder.build(id, version)
203    }
204}
205
206impl<D: WorkflowData> ParallelBuilder<D> {
207    /// Add a parallel branch.
208    pub fn branch(
209        mut self,
210        build_branch: impl FnOnce(&mut WorkflowBuilder<D>),
211    ) -> Self {
212        let before_count = self.builder.steps.len();
213        build_branch(&mut self.builder);
214        let after_count = self.builder.steps.len();
215
216        for child_id in before_count..after_count {
217            self.builder.add_child(self.container_id, child_id);
218        }
219        self
220    }
221}
wfe_core/builder/step_builder.rs

wfe_core/builder/
step_builder.rs
