dex_sdk/
step.rs

1// Copyright (c) 2026 Super Durable, Inc.
2//
3// Licensed under the Super Durable Source License 1.0.
4// You may not use this file except in compliance with the License.
5// See the LICENSE file in the repository root.
6//
7// SPDX-License-Identifier: LicenseRef-Super-Durable-1.0
8
9use std::marker::PhantomData;
10
11use dex_protocol::dex::Value as ProtoValue;
12
13use crate::channel::ChannelGuard;
14use crate::step_options::ErasedStepOptions;
15use crate::value_mapper;
16use crate::{Context, HandlerResult, SdkResult, StepOptions, Value, Wait, short_type_name};
17
18/// Defines one durable state transition in a Flow.
19///
20/// Dex calls [`Self::wait_for`] to obtain durable conditions, then calls [`Self::execute`] after the
21/// wait is satisfied. The default `wait_for` skips immediately, making execute-only Steps concise.
22/// Handler code may read and stage persistence changes through [`Context`].
23///
24/// # Examples
25///
26/// ```
27/// use dex_sdk::{Context, HandlerResult, Step, StepDecision};
28///
29/// struct CompleteOrder;
30///
31/// impl Step for CompleteOrder {
32///     type Input = String;
33///
34///     fn execute(
35///         &self,
36///         _context: &mut Context,
37///         order_id: Self::Input,
38///     ) -> HandlerResult<StepDecision> {
39///         Ok(StepDecision::graceful_complete(order_id))
40///     }
41/// }
42/// ```
43pub trait Step: Send + Sync + 'static {
44    /// Value decoded for both `wait_for` and `execute`.
45    type Input: Value;
46
47    /// Produces the next movement or terminal Flow decision.
48    ///
49    /// `context` represents one isolated invocation and `input` is a newly decoded owned value.
50    /// Returning [`HandlerError`](crate::HandlerError) activates the configured execute retry and
51    /// failure behavior.
52    fn execute(&self, context: &mut Context, input: Self::Input) -> HandlerResult<StepDecision>;
53
54    /// Returns the durable conditions Dex must satisfy before `execute`.
55    ///
56    /// The default returns [`Wait::skip_immediately`]. Returning an error activates the configured
57    /// `wait_for` retry and failure behavior.
58    fn wait_for(&self, _context: &mut Context, _input: Self::Input) -> HandlerResult<Wait> {
59        Ok(Wait::skip_immediately())
60    }
61
62    /// Returns the stable Step type sent to Dex.
63    ///
64    /// The default is the final Rust type-name segment.
65    fn step_type(&self) -> &'static str {
66        short_type_name::<Self>()
67    }
68
69    /// Returns timeouts, retries, locks, durability, and failure behavior for this Step.
70    ///
71    /// The default preserves server defaults and fails the Flow when `wait_for` exhausts retries.
72    fn options(&self) -> StepOptions<Self::Input> {
73        StepOptions::new()
74    }
75}
76
77/// Collects a Flow's starting Step and remaining Step definitions.
78///
79/// The lifetime ties registered definitions to Step values owned by the Flow. A list may contain at
80/// most one starting Step; [`Self::and`] adds non-starting Steps with arbitrary input types.
81pub struct StepList<'a, StartInput> {
82    definitions: Vec<StepDefinition<'a>>,
83    marker: PhantomData<fn(StartInput)>,
84}
85
86impl<'a, StartInput: Value> StepList<'a, StartInput> {
87    /// Creates a list without a starting Step.
88    pub fn empty() -> Self {
89        Self {
90            definitions: Vec::new(),
91            marker: PhantomData,
92        }
93    }
94
95    /// Creates a list whose starting Step accepts the Flow's start input.
96    pub fn start<StartStep>(step: &'a StartStep) -> Self
97    where
98        StartStep: Step<Input = StartInput>,
99    {
100        Self {
101            definitions: vec![StepDefinition::new(step, true)],
102            marker: PhantomData,
103        }
104    }
105
106    /// Adds a non-starting Step and returns the updated list.
107    pub fn and<OtherStep>(mut self, step: &'a OtherStep) -> Self
108    where
109        OtherStep: Step,
110    {
111        self.definitions.push(StepDefinition::new(step, false));
112        self
113    }
114
115    pub(crate) fn into_definitions(self) -> Vec<RegisteredStep> {
116        self.definitions
117            .into_iter()
118            .map(|definition| RegisteredStep {
119                name: definition.name,
120                starting: definition.starting,
121            })
122            .collect()
123    }
124
125    pub(crate) fn find(&self, name: &str) -> Option<&dyn ErasedStep> {
126        self.definitions
127            .iter()
128            .find(|definition| definition.name == name)
129            .map(|definition| definition.handler)
130    }
131}
132
133impl<StartInput: Value> Default for StepList<'_, StartInput> {
134    fn default() -> Self {
135        Self::empty()
136    }
137}
138
139/// Targets a Step with encoded input and an optional per-movement options override.
140///
141/// Create movements with [`Self::to`] or [`Self::to_with_options`], then pass them to
142/// [`StepDecision::go_to_many`], [`crate::RpcResult::then`], or conditional completion.
143pub struct StepMovement {
144    pub(crate) target_step_type: &'static str,
145    input: Box<dyn ErasedValue>,
146    pub(crate) options_override: Option<ErasedStepOptions>,
147}
148
149impl StepMovement {
150    /// Targets `step` with typed `input` and the Step's registered options.
151    pub fn to<TargetStep>(step: &TargetStep, input: TargetStep::Input) -> Self
152    where
153        TargetStep: Step,
154    {
155        Self {
156            target_step_type: step.step_type(),
157            input: Box::new(TypedValue(input)),
158            options_override: None,
159        }
160    }
161
162    /// Targets `step` with typed input and options used only for this movement.
163    pub fn to_with_options<TargetStep>(
164        step: &TargetStep,
165        input: TargetStep::Input,
166        options: StepOptions<TargetStep::Input>,
167    ) -> Self
168    where
169        TargetStep: Step,
170    {
171        Self {
172            target_step_type: step.step_type(),
173            input: Box::new(TypedValue(input)),
174            options_override: Some(options.into()),
175        }
176    }
177
178    pub(crate) fn encode_input(&self) -> SdkResult<ProtoValue> {
179        self.input.encode()
180    }
181}
182
183/// Describes the durable outcome of one successful Step execution.
184///
185/// A decision moves to one or more Steps, completes or fails the Flow, waits for Channels to empty,
186/// or deliberately leaves the Flow at a dead end.
187pub struct StepDecision {
188    pub(crate) kind: StepDecisionKind,
189}
190
191pub(crate) enum StepDecisionKind {
192    Next(Vec<StepMovement>),
193    GracefulComplete(Box<dyn ErasedValue>),
194    ForceComplete(Box<dyn ErasedValue>),
195    ForceCompleteIfChannelsEmpty {
196        output: Box<dyn ErasedValue>,
197        fallback: Box<StepMovement>,
198        channels: Vec<ChannelGuard>,
199    },
200    ForceFail(String),
201    DeadEnd,
202}
203
204impl StepDecision {
205    /// Moves to one typed target Step.
206    pub fn go_to<TargetStep>(step: &TargetStep, input: TargetStep::Input) -> Self
207    where
208        TargetStep: Step,
209    {
210        Self::go_to_many([StepMovement::to(step, input)])
211    }
212
213    /// Moves to all targets, enabling concurrent active Steps.
214    pub fn go_to_many(movements: impl IntoIterator<Item = StepMovement>) -> Self {
215        Self {
216            kind: StepDecisionKind::Next(movements.into_iter().collect()),
217        }
218    }
219
220    /// Completes only after all other active Steps finish, returning `output`.
221    pub fn graceful_complete<Output: Value>(output: Output) -> Self {
222        Self {
223            kind: StepDecisionKind::GracefulComplete(Box::new(TypedValue(output))),
224        }
225    }
226
227    /// Completes immediately, abandoning other active Steps and returning `output`.
228    pub fn force_complete<Output: Value>(output: Output) -> Self {
229        Self {
230            kind: StepDecisionKind::ForceComplete(Box::new(TypedValue(output))),
231        }
232    }
233
234    /// Completes with `output` when every guard is empty; otherwise follows `fallback`.
235    pub fn force_complete_if_channels_empty<Output: Value>(
236        output: Output,
237        fallback: StepMovement,
238        channels: impl IntoIterator<Item = ChannelGuard>,
239    ) -> Self {
240        Self {
241            kind: StepDecisionKind::ForceCompleteIfChannelsEmpty {
242                output: Box::new(TypedValue(output)),
243                fallback: Box::new(fallback),
244                channels: channels.into_iter().collect(),
245            },
246        }
247    }
248
249    /// Fails the Flow immediately with an application-provided reason.
250    pub fn force_fail(reason: impl Into<String>) -> Self {
251        Self {
252            kind: StepDecisionKind::ForceFail(reason.into()),
253        }
254    }
255
256    /// Leaves the Flow running with no next Step.
257    ///
258    /// Use this only when an RPC or external operation will resume the Flow later.
259    pub fn dead_end() -> Self {
260        Self {
261            kind: StepDecisionKind::DeadEnd,
262        }
263    }
264}
265
266#[derive(Clone)]
267pub(crate) struct RegisteredStep {
268    pub(crate) name: &'static str,
269    pub(crate) starting: bool,
270}
271
272struct StepDefinition<'a> {
273    name: &'static str,
274    starting: bool,
275    handler: &'a dyn ErasedStep,
276}
277
278impl<'a> StepDefinition<'a> {
279    fn new<SomeStep>(step: &'a SomeStep, starting: bool) -> Self
280    where
281        SomeStep: Step,
282    {
283        let name = step.step_type();
284        Self {
285            name,
286            starting,
287            handler: step,
288        }
289    }
290}
291
292pub(crate) trait ErasedStep: Send + Sync {
293    fn wait_for(&self, context: &mut Context, input: &ProtoValue) -> HandlerResult<Wait>;
294    fn execute(&self, context: &mut Context, input: &ProtoValue) -> HandlerResult<StepDecision>;
295    fn options(&self) -> ErasedStepOptions;
296}
297
298impl<SomeStep> ErasedStep for SomeStep
299where
300    SomeStep: Step,
301{
302    fn wait_for(&self, context: &mut Context, input: &ProtoValue) -> HandlerResult<Wait> {
303        Step::wait_for(self, context, value_mapper::decode_handler(input)?)
304    }
305
306    fn execute(&self, context: &mut Context, input: &ProtoValue) -> HandlerResult<StepDecision> {
307        Step::execute(self, context, value_mapper::decode_handler(input)?)
308    }
309
310    fn options(&self) -> ErasedStepOptions {
311        Step::options(self).into()
312    }
313}
314
315pub(crate) trait ErasedValue: Send + Sync {
316    fn encode(&self) -> SdkResult<ProtoValue>;
317}
318
319pub(crate) struct TypedValue<T>(pub(crate) T);
320
321impl<T: Value> ErasedValue for TypedValue<T> {
322    fn encode(&self) -> SdkResult<ProtoValue> {
323        value_mapper::encode(&self.0)
324    }
325}
dex_sdk/step.rs

dex_sdk/
step.rs
