dex_sdk/
wait.rs

1// Copyright (c) 2026 Super Durable, Inc.
2//
3// Licensed under the Sustainable Use 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-Sustainable-Use-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    /// Among ready conditions, Dex selects the first in canonical order: Timers, Channels, then
68    /// SubFlows, preserving iterator order within each kind. An unready condition does not block a
69    /// later ready condition. To wait for strict priority, return only the higher-priority
70    /// condition.
71    ///
72    /// Channel consumption is not greedy across alternatives. Only the selected Channel condition
73    /// consumes messages; other ready Channel conditions consume nothing.
74    pub fn any_of(conditions: impl IntoIterator<Item = Condition>) -> Self {
75        Self {
76            kind: WaitKind::AnyOf(conditions.into_iter().collect()),
77        }
78    }
79
80    /// Waits for every Condition in any combination.
81    ///
82    /// Channel consumption is limited to Channel conditions in the selected combination.
83    /// Conditions belonging only to other ready combinations consume nothing.
84    ///
85    /// Every Condition needs a non-empty user ID. Clones of the same Condition
86    /// may be reused across combinations; distinct Conditions need unique IDs.
87    pub fn any_combination_of(
88        combinations: impl IntoIterator<Item = ConditionCombination>,
89    ) -> Self {
90        Self {
91            kind: WaitKind::AnyCombinationOf(combinations.into_iter().collect()),
92        }
93    }
94}
95
96/// Represents one durable timer, Channel, or SubFlow predicate.
97///
98/// Conditions are created by [`crate::Timer`], [`crate::Channel`], or [`crate::ChannelMap`]. Assign
99/// an ID when identifying a timer independently or using [`Wait::any_combination_of`].
100#[derive(Clone)]
101pub struct Condition {
102    pub(crate) identity: Arc<ConditionIdentity>,
103    pub(crate) id: Option<String>,
104    pub(crate) kind: ConditionKind,
105}
106
107pub(crate) struct ConditionIdentity;
108
109#[derive(Clone)]
110pub(crate) enum ConditionKind {
111    Timer(Duration),
112    Channel {
113        name: String,
114        instance: Option<String>,
115        at_least: Option<usize>,
116        at_most: Option<usize>,
117    },
118    SubFlow(Box<SubFlowDefinition>),
119}
120
121#[derive(Clone)]
122pub(crate) struct SubFlowDefinition {
123    pub(crate) flow_type: &'static str,
124    pub(crate) type_id: TypeId,
125    pub(crate) input: ProtoValue,
126    pub(crate) options: SubFlowOptions,
127}
128
129impl Condition {
130    pub(crate) fn timer(duration: Duration) -> Self {
131        Self {
132            identity: Arc::new(ConditionIdentity),
133            id: None,
134            kind: ConditionKind::Timer(duration),
135        }
136    }
137
138    pub(crate) fn channel(
139        name: String,
140        instance: Option<String>,
141        at_least: Option<usize>,
142        at_most: Option<usize>,
143    ) -> Self {
144        Self {
145            identity: Arc::new(ConditionIdentity),
146            id: None,
147            kind: ConditionKind::Channel {
148                name,
149                instance,
150                at_least,
151                at_most,
152            },
153        }
154    }
155
156    pub(crate) fn sub_flow(definition: SubFlowDefinition) -> Self {
157        let id = definition.options.condition_id.clone();
158        Self {
159            identity: Arc::new(ConditionIdentity),
160            id,
161            kind: ConditionKind::SubFlow(Box::new(definition)),
162        }
163    }
164
165    /// Assigns a stable ID used by [`crate::TimerId::by_condition_id`].
166    pub fn with_id(mut self, id: impl Into<String>) -> Self {
167        self.id = Some(id.into());
168        self
169    }
170}
171
172/// Groups conditions that must all succeed as one branch of [`Wait::any_combination_of`].
173pub struct ConditionCombination {
174    pub(crate) conditions: Vec<Condition>,
175}
176
177impl ConditionCombination {
178    /// Creates a branch satisfied only when all `conditions` are satisfied.
179    pub fn all_of(conditions: impl IntoIterator<Item = Condition>) -> Self {
180        Self {
181            conditions: conditions.into_iter().collect(),
182        }
183    }
184}
dex_sdk/wait.rs

dex_sdk/
wait.rs
