dex_sdk/
start_flow_options.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::time::Duration;
10
11use std::sync::Arc;
12
13use dex_protocol::dex::{AttributeSyncConfig, IndexConfig};
14
15use crate::registry::physical_name;
16use crate::step::{ErasedValue, TypedValue};
17use crate::{Attribute, AttributeMap, FlowConfig, RetryPolicy, Value};
18
19#[derive(Clone, Copy, Debug, Eq, PartialEq)]
20/// Controls whether [`crate::Client::start_flow_with_options`] may reuse a Flow ID.
21pub enum IdReusePolicy {
22    /// Uses the Dex server's configured reuse policy.
23    Default,
24    /// Allows reuse only when the previous run failed.
25    AllowIfPreviousFailed,
26    /// Allows reuse when no run with the ID is currently active.
27    AllowIfNotRunning,
28    /// Rejects reuse regardless of the previous run status.
29    Disallow,
30    /// Terminates an active run before starting the replacement.
31    TerminateIfRunning,
32}
33
34#[derive(Clone)]
35/// Configures a new Flow execution.
36///
37/// All optional fields are absent by default. Initial Attribute values are encoded when the start
38/// request is sent. Use a stable request ID to make retries of the same logical request idempotent.
39///
40/// # Examples
41///
42/// ```
43/// use dex_sdk::{Attribute, IdReusePolicy, StartFlowOptions};
44/// use std::time::Duration;
45///
46/// let customer = Attribute::<String>::new("customer");
47/// let options = StartFlowOptions::new()
48///     .timeout(Duration::from_secs(3_600))
49///     .id_reuse_policy(IdReusePolicy::AllowIfNotRunning)
50///     .initial_attribute(&customer, "customer-42".to_owned())
51///     .request_id("create-order-42");
52/// ```
53pub struct StartFlowOptions {
54    pub(crate) timeout: Option<Duration>,
55    pub(crate) start_delay: Option<Duration>,
56    pub(crate) id_reuse_policy: IdReusePolicy,
57    pub(crate) cron_schedule: Option<String>,
58    pub(crate) retry_policy: Option<RetryPolicy>,
59    pub(crate) config_override: Option<FlowConfig>,
60    pub(crate) ignore_already_started: bool,
61    pub(crate) request_id: Option<String>,
62    pub(crate) attributes: Vec<InitialAttribute>,
63}
64
65#[derive(Clone)]
66pub(crate) struct InitialAttribute {
67    pub(crate) key: String,
68    pub(crate) value: Arc<dyn ErasedValue>,
69    pub(crate) index_config: Option<IndexConfig>,
70    pub(crate) sync_config: Option<AttributeSyncConfig>,
71}
72
73impl StartFlowOptions {
74    /// Creates start options that preserve all server defaults.
75    pub fn new() -> Self {
76        Self {
77            timeout: None,
78            start_delay: None,
79            id_reuse_policy: IdReusePolicy::Default,
80            cron_schedule: None,
81            retry_policy: None,
82            config_override: None,
83            ignore_already_started: false,
84            request_id: None,
85            attributes: Vec::new(),
86        }
87    }
88
89    /// Sets the maximum total Flow execution duration.
90    pub fn timeout(mut self, value: Duration) -> Self {
91        self.timeout = Some(value);
92        self
93    }
94
95    /// Delays the first Step after the server accepts the Flow.
96    pub fn start_delay(mut self, value: Duration) -> Self {
97        self.start_delay = Some(value);
98        self
99    }
100
101    /// Sets how the server handles an existing Flow with the same ID.
102    pub fn id_reuse_policy(mut self, value: IdReusePolicy) -> Self {
103        self.id_reuse_policy = value;
104        self
105    }
106
107    /// Sets the server cron expression used to schedule recurring runs.
108    pub fn cron_schedule(mut self, value: impl Into<String>) -> Self {
109        self.cron_schedule = Some(value.into());
110        self
111    }
112
113    /// Sets the whole-Flow retry policy after a terminal failure.
114    pub fn retry_policy(mut self, value: RetryPolicy) -> Self {
115        self.retry_policy = Some(value);
116        self
117    }
118
119    /// Adds an initial value for a declared Attribute.
120    pub fn initial_attribute<T: Value>(mut self, attribute: &Attribute<T>, value: T) -> Self {
121        self.attributes.push(InitialAttribute {
122            key: attribute.name().to_string(),
123            value: Arc::new(TypedValue(value)),
124            index_config: attribute.index().map(|index| index.proto_config(false)),
125            sync_config: attribute.sync_config(),
126        });
127        self
128    }
129
130    /// Adds an initial value for one Attribute-map instance.
131    pub fn initial_attribute_map<T: Value>(
132        mut self,
133        attribute: &AttributeMap<T>,
134        instance: &str,
135        value: T,
136    ) -> Self {
137        self.attributes.push(InitialAttribute {
138            key: physical_name(attribute.name(), instance),
139            value: Arc::new(TypedValue(value)),
140            index_config: attribute.index().map(|index| index.proto_config(true)),
141            sync_config: attribute.sync_config(),
142        });
143        self
144    }
145
146    /// Overrides the registered Flow configuration for this execution.
147    pub fn config_override(mut self, value: FlowConfig) -> Self {
148        self.config_override = Some(value);
149        self
150    }
151
152    /// Returns the existing run ID instead of an error when the Flow already exists.
153    pub fn ignore_already_started(mut self, value: bool) -> Self {
154        self.ignore_already_started = value;
155        self
156    }
157
158    /// Sets the idempotency key for this start request.
159    pub fn request_id(mut self, value: impl Into<String>) -> Self {
160        self.request_id = Some(value.into());
161        self
162    }
163}
164
165impl Default for StartFlowOptions {
166    fn default() -> Self {
167        Self::new()
168    }
169}
170
171#[cfg(test)]
172mod tests {
173    use super::StartFlowOptions;
174    use crate::{Attribute, AttributeMap};
175
176    #[test]
177    fn initial_attributes_retain_sync_configuration() {
178        let plain = Attribute::<String>::new("plain");
179        let synced = Attribute::<String>::new("synced").sync_to_attribute_store();
180        let synced_map = AttributeMap::<String>::new("map").sync_to_attribute_store();
181        let options = StartFlowOptions::new()
182            .initial_attribute(&plain, "plain".to_string())
183            .initial_attribute(&synced, "synced".to_string())
184            .initial_attribute_map(&synced_map, "tenant-1", "mapped".to_string());
185
186        assert!(options.attributes[0].sync_config.is_none());
187        assert_eq!(
188            options.attributes[1]
189                .sync_config
190                .as_ref()
191                .map(|config| config.enabled),
192            Some(true)
193        );
194        assert_eq!(
195            options.attributes[2]
196                .sync_config
197                .as_ref()
198                .map(|config| config.enabled),
199            Some(true)
200        );
201    }
202}
dex_sdk/start_flow_options.rs

dex_sdk/
start_flow_options.rs
