ironflow_engine/
run_creator.rs

1//! [`RunCreator`] trait and [`CreateRunOpts`] builder -- centralised run creation.
2//!
3//! [`RunCreator`] is a minimal trait with a single method: create a run from
4//! a [`NewRun`]. It is intentionally thinner than [`RunStore`] so that any
5//! store implementation can be used as a run creator through the blanket impl.
6//!
7//! [`CreateRunOpts`] is a builder for the optional fields of a run creation
8//! request. Combined with [`WorkflowHandler::create_run`](crate::handler::WorkflowHandler::create_run), it assembles a
9//! [`NewRun`] from the handler's own metadata, removing duplication across
10//! call sites.
11//!
12//! # Examples
13//!
14//! ```no_run
15//! use ironflow_engine::run_creator::{CreateRunOpts, RunCreator};
16//! use ironflow_store::entities::TriggerKind;
17//! use ironflow_store::memory::InMemoryStore;
18//!
19//! # async fn example() -> Result<(), ironflow_engine::error::EngineError> {
20//! let store = InMemoryStore::new();
21//! let creator: &dyn RunCreator = &store;
22//!
23//! let new_run = CreateRunOpts::new()
24//!     .trigger(TriggerKind::Api)
25//!     .build("deploy", Some("1.0.0"), None);
26//!
27//! let creation = creator.create_run(new_run).await?;
28//! # Ok(())
29//! # }
30//! ```
31
32use std::collections::HashMap;
33use std::future::Future;
34use std::pin::Pin;
35
36use chrono::{DateTime, Utc};
37use rust_decimal::Decimal;
38use serde_json::Value;
39
40use ironflow_store::entities::{NewRun, RunActor, RunCreation, TriggerKind, normalize_worker_tags};
41use ironflow_store::store::RunStore;
42
43use crate::error::EngineError;
44
45/// Future returned by [`RunCreator::create_run`].
46pub type RunCreatorFuture<'a> =
47    Pin<Box<dyn Future<Output = Result<RunCreation, EngineError>> + Send + 'a>>;
48
49/// Minimal trait for creating workflow runs.
50///
51/// Automatically implemented for every [`RunStore`] via a blanket impl,
52/// so any store (InMemory, Postgres, ApiRunStore) is a valid [`RunCreator`].
53///
54/// # Examples
55///
56/// ```no_run
57/// use ironflow_engine::run_creator::RunCreator;
58/// use ironflow_store::entities::{NewRun, TriggerKind};
59///
60/// # async fn example(creator: &dyn RunCreator) -> Result<(), ironflow_engine::error::EngineError> {
61/// let new_run = NewRun {
62///     workflow_name: "deploy".to_string(),
63///     trigger: TriggerKind::Manual,
64///     payload: serde_json::json!({}),
65///     max_retries: 0,
66///     handler_version: None,
67///     labels: Default::default(),
68///     scheduled_at: None,
69///     created_by: None,
70///     idempotency_key: None,
71///     concurrency_key: None,
72///     concurrency_limits: Vec::new(),
73///     max_cost_usd: None,
74///     worker_tags: Vec::new(),
75/// };
76/// let creation = creator.create_run(new_run).await?;
77/// # Ok(())
78/// # }
79/// ```
80pub trait RunCreator: Send + Sync {
81    /// Create a new workflow run.
82    ///
83    /// # Errors
84    ///
85    /// Returns [`EngineError`] if the run could not be created.
86    fn create_run(&self, req: NewRun) -> RunCreatorFuture<'_>;
87}
88
89impl<T: RunStore + ?Sized> RunCreator for T {
90    fn create_run(&self, req: NewRun) -> RunCreatorFuture<'_> {
91        Box::pin(async move {
92            RunStore::create_run(self, req)
93                .await
94                .map_err(EngineError::from)
95        })
96    }
97}
98
99/// Builder for optional run creation fields.
100///
101/// Builds a [`NewRun`] from handler metadata plus user-supplied overrides.
102/// Use with [`WorkflowHandler::create_run`] to avoid duplicating workflow
103/// name, version, and cost cap at every call site.
104///
105/// [`WorkflowHandler::create_run`]: crate::handler::WorkflowHandler::create_run
106///
107/// # Examples
108///
109/// ```
110/// use ironflow_engine::run_creator::CreateRunOpts;
111/// use ironflow_store::entities::TriggerKind;
112/// use serde_json::json;
113///
114/// let new_run = CreateRunOpts::new()
115///     .trigger(TriggerKind::Webhook { path: "/hooks/gh".to_string() })
116///     .payload(json!({"ref": "main"}))
117///     .max_retries(3)
118///     .build("deploy", Some("2.0.0"), None);
119///
120/// assert_eq!(new_run.workflow_name, "deploy");
121/// assert_eq!(new_run.max_retries, 3);
122/// assert_eq!(new_run.handler_version, Some("2.0.0".to_string()));
123/// ```
124#[derive(Debug, Clone, Default)]
125pub struct CreateRunOpts {
126    trigger: Option<TriggerKind>,
127    payload: Option<Value>,
128    max_retries: Option<u32>,
129    scheduled_at: Option<DateTime<Utc>>,
130    created_by: Option<RunActor>,
131    idempotency_key: Option<String>,
132    concurrency_key: Option<String>,
133    labels: Option<HashMap<String, String>>,
134    max_cost_usd: Option<Decimal>,
135    worker_tags: Vec<String>,
136}
137
138impl CreateRunOpts {
139    /// Create a new builder with all fields unset.
140    ///
141    /// # Examples
142    ///
143    /// ```
144    /// use ironflow_engine::run_creator::CreateRunOpts;
145    ///
146    /// let opts = CreateRunOpts::new();
147    /// let new_run = opts.build("my-workflow", None, None);
148    /// assert_eq!(new_run.workflow_name, "my-workflow");
149    /// ```
150    pub fn new() -> Self {
151        Self::default()
152    }
153
154    /// Set how the run was triggered.
155    ///
156    /// Defaults to [`TriggerKind::Manual`] if not set.
157    pub fn trigger(mut self, trigger: TriggerKind) -> Self {
158        self.trigger = Some(trigger);
159        self
160    }
161
162    /// Set the trigger-specific payload.
163    ///
164    /// Defaults to `json!({})` if not set.
165    pub fn payload(mut self, payload: Value) -> Self {
166        self.payload = Some(payload);
167        self
168    }
169
170    /// Set the maximum retry attempts.
171    ///
172    /// Defaults to `0` if not set.
173    pub fn max_retries(mut self, max_retries: u32) -> Self {
174        self.max_retries = Some(max_retries);
175        self
176    }
177
178    /// Schedule the run for later execution.
179    pub fn scheduled_at(mut self, at: DateTime<Utc>) -> Self {
180        self.scheduled_at = Some(at);
181        self
182    }
183
184    /// Set the authenticated principal creating this run.
185    pub fn created_by(mut self, actor: RunActor) -> Self {
186        self.created_by = Some(actor);
187        self
188    }
189
190    /// Set an idempotency key to prevent duplicate runs.
191    pub fn idempotency_key(mut self, key: impl Into<String>) -> Self {
192        self.idempotency_key = Some(key.into());
193        self
194    }
195
196    /// Set a concurrency key: the store refuses the run while another
197    /// non-terminal run holds the same key.
198    ///
199    /// # Examples
200    ///
201    /// ```
202    /// use ironflow_engine::run_creator::CreateRunOpts;
203    ///
204    /// let new_run = CreateRunOpts::new()
205    ///     .concurrency_key("issue:12")
206    ///     .build("deploy", None, None);
207    /// assert_eq!(new_run.concurrency_key.as_deref(), Some("issue:12"));
208    /// ```
209    pub fn concurrency_key(mut self, key: impl Into<String>) -> Self {
210        self.concurrency_key = Some(key.into());
211        self
212    }
213
214    /// Set user-defined labels for categorization.
215    pub fn labels(mut self, labels: HashMap<String, String>) -> Self {
216        self.labels = Some(labels);
217        self
218    }
219
220    /// Set the maximum cumulative cost allowed for this run.
221    pub fn max_cost_usd(mut self, cap: Decimal) -> Self {
222        self.max_cost_usd = Some(cap);
223        self
224    }
225
226    /// Add worker tags the run requires. Extends the tags already set.
227    ///
228    /// Tags are trimmed, sorted and deduplicated by [`build`](Self::build).
229    /// The store refuses invalid ones when the run is created.
230    ///
231    /// # Examples
232    ///
233    /// ```
234    /// use ironflow_engine::run_creator::CreateRunOpts;
235    ///
236    /// let new_run = CreateRunOpts::new()
237    ///     .worker_tags(["region:eu", "gpu"])
238    ///     .worker_tags(["gpu"])
239    ///     .build("transcode", None, None);
240    /// assert_eq!(new_run.worker_tags, vec!["gpu".to_string(), "region:eu".to_string()]);
241    /// ```
242    pub fn worker_tags<I, S>(mut self, tags: I) -> Self
243    where
244        I: IntoIterator<Item = S>,
245        S: Into<String>,
246    {
247        self.worker_tags.extend(tags.into_iter().map(Into::into));
248        self
249    }
250
251    /// Assemble a [`NewRun`] from these options and handler metadata.
252    ///
253    /// * `workflow_name` -- typically from [`WorkflowHandler::name`].
254    /// * `handler_version` -- typically from [`WorkflowHandler::version`].
255    /// * `default_max_cost_usd` -- typically from [`WorkflowHandler::default_max_cost_usd`].
256    ///   Applied only when [`max_cost_usd`](Self::max_cost_usd) was not set.
257    ///
258    /// [`WorkflowHandler::name`]: crate::handler::WorkflowHandler::name
259    /// [`WorkflowHandler::version`]: crate::handler::WorkflowHandler::version
260    /// [`WorkflowHandler::default_max_cost_usd`]: crate::handler::WorkflowHandler::default_max_cost_usd
261    ///
262    /// # Examples
263    ///
264    /// ```
265    /// use ironflow_engine::run_creator::CreateRunOpts;
266    /// use rust_decimal::Decimal;
267    ///
268    /// let new_run = CreateRunOpts::new()
269    ///     .build("my-handler", Some("3.0.0"), Some(Decimal::new(1000, 2)));
270    ///
271    /// assert_eq!(new_run.workflow_name, "my-handler");
272    /// assert_eq!(new_run.handler_version, Some("3.0.0".to_string()));
273    /// assert_eq!(new_run.max_cost_usd, Some(Decimal::new(1000, 2)));
274    /// ```
275    pub fn build(
276        self,
277        workflow_name: &str,
278        handler_version: Option<&str>,
279        default_max_cost_usd: Option<Decimal>,
280    ) -> NewRun {
281        NewRun {
282            workflow_name: workflow_name.to_string(),
283            trigger: self.trigger.unwrap_or(TriggerKind::Manual),
284            payload: self.payload.unwrap_or_else(|| serde_json::json!({})),
285            max_retries: self.max_retries.unwrap_or(0),
286            handler_version: handler_version.map(str::to_string),
287            labels: self.labels.unwrap_or_default(),
288            scheduled_at: self.scheduled_at,
289            created_by: self.created_by,
290            idempotency_key: self.idempotency_key,
291            concurrency_key: self.concurrency_key,
292            concurrency_limits: Vec::new(),
293            max_cost_usd: self.max_cost_usd.or(default_max_cost_usd),
294            worker_tags: normalize_worker_tags(self.worker_tags),
295        }
296    }
297}
298
299#[cfg(test)]
300mod tests {
301    use super::*;
302    use serde_json::json;
303
304    #[test]
305    fn create_run_opts_default_produces_correct_defaults() {
306        let opts = CreateRunOpts::new();
307        let new_run = opts.build("test-workflow", None, None);
308
309        assert_eq!(new_run.workflow_name, "test-workflow");
310        assert_eq!(new_run.trigger, TriggerKind::Manual);
311        assert_eq!(new_run.payload, json!({}));
312        assert_eq!(new_run.max_retries, 0);
313        assert_eq!(new_run.handler_version, None);
314        assert!(new_run.labels.is_empty());
315        assert_eq!(new_run.scheduled_at, None);
316        assert_eq!(new_run.created_by, None);
317        assert_eq!(new_run.idempotency_key, None);
318        assert_eq!(new_run.concurrency_key, None);
319        assert_eq!(new_run.max_cost_usd, None);
320    }
321
322    #[test]
323    fn create_run_opts_without_worker_tags_requires_none() {
324        let new_run = CreateRunOpts::new().build("test-workflow", None, None);
325        assert!(new_run.worker_tags.is_empty());
326    }
327
328    #[test]
329    fn create_run_opts_worker_tags_are_merged_and_normalized() {
330        let new_run = CreateRunOpts::new()
331            .worker_tags(["region:eu", " gpu "])
332            .worker_tags(vec!["gpu".to_string()])
333            .build("test-workflow", None, None);
334        assert_eq!(
335            new_run.worker_tags,
336            vec!["gpu".to_string(), "region:eu".to_string()]
337        );
338    }
339
340    #[test]
341    fn create_run_opts_builder_sets_all_fields() {
342        let labels = HashMap::from([("env".to_string(), "prod".to_string())]);
343        let scheduled = Utc::now();
344
345        let new_run = CreateRunOpts::new()
346            .trigger(TriggerKind::Webhook {
347                path: "/hooks/gh".to_string(),
348            })
349            .payload(json!({"ref": "main"}))
350            .max_retries(3)
351            .scheduled_at(scheduled)
352            .idempotency_key("key-123")
353            .labels(labels.clone())
354            .max_cost_usd(Decimal::new(500, 2))
355            .build("deploy", Some("2.0.0"), None);
356
357        assert_eq!(new_run.workflow_name, "deploy");
358        assert_eq!(
359            new_run.trigger,
360            TriggerKind::Webhook {
361                path: "/hooks/gh".to_string()
362            }
363        );
364        assert_eq!(new_run.payload, json!({"ref": "main"}));
365        assert_eq!(new_run.max_retries, 3);
366        assert_eq!(new_run.handler_version, Some("2.0.0".to_string()));
367        assert_eq!(new_run.scheduled_at, Some(scheduled));
368        assert_eq!(new_run.idempotency_key, Some("key-123".to_string()));
369        assert_eq!(new_run.labels, labels);
370        assert_eq!(new_run.max_cost_usd, Some(Decimal::new(500, 2)));
371    }
372
373    #[test]
374    fn create_run_opts_build_carries_the_concurrency_key() {
375        let new_run = CreateRunOpts::new()
376            .concurrency_key("issue:12")
377            .build("deploy", None, None);
378
379        assert_eq!(new_run.concurrency_key.as_deref(), Some("issue:12"));
380        assert_eq!(new_run.idempotency_key, None);
381    }
382
383    #[test]
384    fn create_run_opts_build_uses_handler_metadata() {
385        let new_run =
386            CreateRunOpts::new().build("my-handler", Some("3.0.0"), Some(Decimal::new(1000, 2)));
387
388        assert_eq!(new_run.workflow_name, "my-handler");
389        assert_eq!(new_run.handler_version, Some("3.0.0".to_string()));
390        assert_eq!(new_run.max_cost_usd, Some(Decimal::new(1000, 2)));
391    }
392
393    #[test]
394    fn create_run_opts_explicit_max_cost_overrides_handler_default() {
395        let new_run = CreateRunOpts::new()
396            .max_cost_usd(Decimal::new(200, 2))
397            .build("handler", Some("1"), Some(Decimal::new(1000, 2)));
398
399        assert_eq!(new_run.max_cost_usd, Some(Decimal::new(200, 2)));
400    }
401
402    #[tokio::test]
403    async fn run_creator_blanket_impl_with_in_memory_store() {
404        use ironflow_store::memory::InMemoryStore;
405
406        let store = InMemoryStore::new();
407        let creator: &dyn RunCreator = &store;
408
409        let new_run =
410            CreateRunOpts::new()
411                .trigger(TriggerKind::Api)
412                .build("blanket-test", None, None);
413
414        let creation = creator.create_run(new_run).await.expect("create_run");
415        let run = creation.into_run();
416        assert_eq!(run.workflow_name, "blanket-test");
417    }
418
419    #[tokio::test]
420    async fn create_run_with_reused_idempotency_key_returns_existing() {
421        use ironflow_store::memory::InMemoryStore;
422
423        let store = InMemoryStore::new();
424        let creator: &dyn RunCreator = &store;
425
426        let first = creator
427            .create_run(CreateRunOpts::new().idempotency_key("dedup-1").build(
428                "idem-test",
429                None,
430                None,
431            ))
432            .await
433            .expect("first create_run");
434        assert!(first.is_created());
435
436        let second = creator
437            .create_run(CreateRunOpts::new().idempotency_key("dedup-1").build(
438                "idem-test",
439                None,
440                None,
441            ))
442            .await
443            .expect("second create_run");
444        assert!(!second.is_created());
445        assert_eq!(first.into_run().id, second.into_run().id);
446    }
447}
ironflow_engine/run_creator.rs

ironflow_engine/
run_creator.rs
