dex_sdk/
wait.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::any::TypeId;
10use std::sync::Arc;
11use std::time::Duration;
12
13use dex_protocol::dex::Value as ProtoValue;
14
15use crate::SubFlowOptions;
16
17/// Describes when Dex may invoke a Step's `execute` handler.
18///
19/// Build waits from durable timer and Channel [`Condition`] values. Combinators consume their
20/// inputs and preserve order, which determines timer indexes used by [`crate::TimerId`].
21///
22/// # Examples
23///
24/// ```
25/// use dex_sdk::{Channel, Timer, Wait};
26/// use std::time::Duration;
27///
28/// let approvals = Channel::<String>::new("approvals");
29/// let wait = Wait::any_of([
30///     approvals.for_one(),
31///     Timer::by_duration(Duration::from_secs(300)),
32/// ]);
33/// ```
34pub struct Wait {
35    pub(crate) kind: WaitKind,
36}
37
38pub(crate) enum WaitKind {
39    SkipImmediately,
40    AllOf(Vec<Condition>),
41    AnyOf(Vec<Condition>),
42    AnyCombinationOf(Vec<ConditionCombination>),
43}
44
45impl Wait {
46    /// Skips `wait_for` and asks Dex to invoke `execute` immediately.
47    pub fn skip_immediately() -> Self {
48        Self {
49            kind: WaitKind::SkipImmediately,
50        }
51    }
52
53    /// Waits until one condition is satisfied. The Condition does not need an ID.
54    pub fn until(condition: Condition) -> Self {
55        Self::all_of([condition])
56    }
57
58    /// Waits until every condition is satisfied. Conditions do not need IDs.
59    pub fn all_of(conditions: impl IntoIterator<Item = Condition>) -> Self {
60        Self {
61            kind: WaitKind::AllOf(conditions.into_iter().collect()),
62        }
63    }
64
65    /// Waits until at least one condition is satisfied. Conditions do not need IDs.
66    ///
67    /// Channel consumption is not greedy across alternatives. Only the selected Channel condition
68    /// consumes messages; other ready Channel conditions consume nothing.
69    pub fn any_of(conditions: impl IntoIterator<Item = Condition>) -> Self {
70        Self {
71            kind: WaitKind::AnyOf(conditions.into_iter().collect()),
72        }
73    }
74
75    /// Waits for every Condition in any combination.
76    ///
77    /// Channel consumption is limited to Channel conditions in the selected combination.
78    /// Conditions belonging only to other ready combinations consume nothing.
79    ///
80    /// Every Condition needs a non-empty user ID. Clones of the same Condition
81    /// may be reused across combinations; distinct Conditions need unique IDs.
82    pub fn any_combination_of(
83        combinations: impl IntoIterator<Item = ConditionCombination>,
84    ) -> Self {
85        Self {
86            kind: WaitKind::AnyCombinationOf(combinations.into_iter().collect()),
87        }
88    }
89}
90
91/// Represents one durable timer, Channel, or SubFlow predicate.
92///
93/// Conditions are created by [`crate::Timer`], [`crate::Channel`], or [`crate::ChannelMap`]. Assign
94/// an ID when identifying a timer independently or using [`Wait::any_combination_of`].
95#[derive(Clone)]
96pub struct Condition {
97    pub(crate) identity: Arc<ConditionIdentity>,
98    pub(crate) id: Option<String>,
99    pub(crate) kind: ConditionKind,
100}
101
102pub(crate) struct ConditionIdentity;
103
104#[derive(Clone)]
105pub(crate) enum ConditionKind {
106    Timer(Duration),
107    Channel {
108        name: String,
109        instance: Option<String>,
110        at_least: Option<usize>,
111        at_most: Option<usize>,
112    },
113    SubFlow(Box<SubFlowDefinition>),
114}
115
116#[derive(Clone)]
117pub(crate) struct SubFlowDefinition {
118    pub(crate) flow_type: &'static str,
119    pub(crate) type_id: TypeId,
120    pub(crate) input: ProtoValue,
121    pub(crate) options: SubFlowOptions,
122}
123
124impl Condition {
125    pub(crate) fn timer(duration: Duration) -> Self {
126        Self {
127            identity: Arc::new(ConditionIdentity),
128            id: None,
129            kind: ConditionKind::Timer(duration),
130        }
131    }
132
133    pub(crate) fn channel(
134        name: String,
135        instance: Option<String>,
136        at_least: Option<usize>,
137        at_most: Option<usize>,
138    ) -> Self {
139        Self {
140            identity: Arc::new(ConditionIdentity),
141            id: None,
142            kind: ConditionKind::Channel {
143                name,
144                instance,
145                at_least,
146                at_most,
147            },
148        }
149    }
150
151    pub(crate) fn sub_flow(definition: SubFlowDefinition) -> Self {
152        let id = definition.options.condition_id.clone();
153        Self {
154            identity: Arc::new(ConditionIdentity),
155            id,
156            kind: ConditionKind::SubFlow(Box::new(definition)),
157        }
158    }
159
160    /// Assigns a stable ID used by [`crate::TimerId::by_condition_id`].
161    pub fn with_id(mut self, id: impl Into<String>) -> Self {
162        self.id = Some(id.into());
163        self
164    }
165}
166
167/// Groups conditions that must all succeed as one branch of [`Wait::any_combination_of`].
168pub struct ConditionCombination {
169    pub(crate) conditions: Vec<Condition>,
170}
171
172impl ConditionCombination {
173    /// Creates a branch satisfied only when all `conditions` are satisfied.
174    pub fn all_of(conditions: impl IntoIterator<Item = Condition>) -> Self {
175        Self {
176            conditions: conditions.into_iter().collect(),
177        }
178    }
179}
dex_sdk/wait.rs

dex_sdk/
wait.rs
