ironflow_api/entities/run.rs
1//! Run-related DTOs and query parameters.
2
3use std::collections::HashMap;
4
5use chrono::{DateTime, Utc};
6use ironflow_store::models::{ConcurrencyLimit, ProviderKind, Run, RunStatus, TriggerKind};
7use rust_decimal::Decimal;
8use serde::{Deserialize, Serialize};
9use serde_json::Value;
10use uuid::Uuid;
11
12use super::{CreatedBy, StepResponse};
13
14/// Run response DTO — public API representation of a run.
15///
16/// Maps from the internal [`Run`] model, exposing only necessary fields.
17///
18/// # Examples
19///
20/// ```
21/// use ironflow_store::models::{Run, RunStatus, TriggerKind};
22/// use ironflow_api::entities::RunResponse;
23/// ```
24#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
25#[derive(Debug, Serialize, Deserialize)]
26pub struct RunResponse {
27 /// Unique run identifier.
28 pub id: Uuid,
29 /// Workflow name.
30 pub workflow_name: String,
31 /// Current status.
32 pub status: RunStatus,
33 /// How the run was triggered.
34 pub trigger: TriggerKind,
35 /// Optional error message.
36 pub error: Option<String>,
37 /// Number of times retried.
38 pub retry_count: u32,
39 /// Maximum allowed retries.
40 pub max_retries: u32,
41 /// Aggregated cost in USD.
42 #[cfg_attr(feature = "openapi", schema(value_type = f64))]
43 pub cost_usd: Decimal,
44 /// Total duration in milliseconds.
45 pub duration_ms: u64,
46 /// When created.
47 pub created_at: DateTime<Utc>,
48 /// When last updated.
49 pub updated_at: DateTime<Utc>,
50 /// When execution started.
51 pub started_at: Option<DateTime<Utc>>,
52 /// When execution completed.
53 pub completed_at: Option<DateTime<Utc>>,
54 /// Version of the handler that created this run.
55 pub handler_version: Option<String>,
56 /// User-defined key-value labels.
57 #[serde(default, skip_serializing_if = "HashMap::is_empty")]
58 pub labels: HashMap<String, String>,
59 /// Scheduled execution time. `None` means the run executed immediately.
60 #[serde(default, skip_serializing_if = "Option::is_none")]
61 pub scheduled_at: Option<DateTime<Utc>>,
62 /// Provider kind the run is waiting capacity for, set only while the
63 /// run is `sleeping` because every targeted Provider Account is rate
64 /// limited. `scheduled_at` then holds when the run wakes.
65 #[serde(default, skip_serializing_if = "Option::is_none")]
66 #[cfg_attr(feature = "openapi", schema(value_type = Option<String>))]
67 pub capacity_wait_kind: Option<ProviderKind>,
68 /// Who triggered the run. Always present.
69 pub created_by: CreatedBy,
70 /// Idempotency key that produced this run, when one was supplied.
71 #[serde(default, skip_serializing_if = "Option::is_none")]
72 pub idempotency_key: Option<String>,
73 /// Exclusivity key held by this run until it reaches a terminal state,
74 /// when one was supplied.
75 #[serde(default, skip_serializing_if = "Option::is_none")]
76 pub concurrency_key: Option<String>,
77 /// Queue priority, from -100 to 100. Workers pick the pending run with
78 /// the highest priority first, then the oldest among equal priorities.
79 /// `0` is the default.
80 #[serde(default)]
81 pub priority: i16,
82 /// Concurrency groups the run belongs to, with the limit it was created
83 /// with. Omitted when the run belongs to no group.
84 #[serde(default, skip_serializing_if = "Vec::is_empty")]
85 pub concurrency_limits: Vec<ConcurrencyLimit>,
86 /// Cumulative cost cap for this run, in USD. `None` means no cap.
87 #[cfg_attr(feature = "openapi", schema(value_type = Option<f64>))]
88 #[serde(default, skip_serializing_if = "Option::is_none")]
89 pub max_cost_usd: Option<Decimal>,
90 /// Tags a worker must carry to take this run, sorted. Empty when any
91 /// worker can take it.
92 #[serde(default)]
93 pub worker_tags: Vec<String>,
94 /// Typed output the handler set with `WorkflowContext::set_output`.
95 ///
96 /// Written when the run ends (completed, warning, failed or cancelled).
97 /// Omitted when the handler set no output.
98 #[serde(default, skip_serializing_if = "Option::is_none")]
99 pub output: Option<Value>,
100}
101
102impl From<Run> for RunResponse {
103 fn from(run: Run) -> Self {
104 let created_by = CreatedBy::from(&run);
105 RunResponse {
106 id: run.id,
107 workflow_name: run.workflow_name,
108 status: run.status.state,
109 trigger: run.trigger,
110 error: run.error,
111 retry_count: run.retry_count,
112 max_retries: run.max_retries,
113 cost_usd: run.cost_usd,
114 duration_ms: run.duration_ms,
115 created_at: run.created_at,
116 updated_at: run.updated_at,
117 started_at: run.started_at,
118 completed_at: run.completed_at,
119 handler_version: run.handler_version,
120 labels: run.labels,
121 scheduled_at: run.scheduled_at,
122 capacity_wait_kind: run.capacity_wait_kind,
123 created_by,
124 idempotency_key: run.idempotency_key,
125 concurrency_key: run.concurrency_key,
126 priority: run.priority,
127 concurrency_limits: run.concurrency_limits,
128 max_cost_usd: run.max_cost_usd,
129 worker_tags: run.worker_tags,
130 output: run.output,
131 }
132 }
133}
134
135/// Run detail response — includes steps and payload.
136#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
137#[derive(Debug, Serialize)]
138pub struct RunDetailResponse {
139 /// The run.
140 pub run: RunResponse,
141 /// Associated steps, ordered by position.
142 pub steps: Vec<StepResponse>,
143 /// Input payload that triggered this run.
144 pub payload: serde_json::Value,
145 /// Sub-workflow runs below this run, at any depth, that are not finished:
146 /// cancelling the run cancels them too.
147 pub active_descendant_count: u64,
148 /// Workers seen recently that could take the run. Only filled while the
149 /// run waits in the queue (`pending` or `retrying`), so a run no worker
150 /// can take is visible instead of waiting in silence.
151 #[serde(default, skip_serializing_if = "Option::is_none")]
152 pub worker_routing: Option<WorkerRouting>,
153}
154
155/// Workers the API saw recently, as seen from one queued run.
156///
157/// Counted from the pick requests received in the last few minutes by this
158/// API process: with several replicas, each one only counts the workers that
159/// polled it.
160///
161/// # Examples
162///
163/// ```
164/// use ironflow_api::entities::WorkerRouting;
165///
166/// let routing = WorkerRouting {
167/// seen_workers: 2,
168/// eligible_workers: 0,
169/// };
170/// assert!(routing.eligible_workers < routing.seen_workers);
171/// ```
172#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
173#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
174pub struct WorkerRouting {
175 /// Workers that asked for a run recently.
176 pub seen_workers: u32,
177 /// Among them, the workers that registered the run's workflow and carry
178 /// every tag it requires.
179 pub eligible_workers: u32,
180}
181
182/// Response of `POST /api/v1/runs/:id/cancel`: the cancelled run, with the
183/// sub-workflow runs cancelled along with it.
184///
185/// The run's fields stay at the top level, as before the descendants were
186/// listed, so existing clients read the same document.
187#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
188#[derive(Debug, Serialize)]
189pub struct CancelRunResponse {
190 /// The cancelled run.
191 #[serde(flatten)]
192 pub run: RunResponse,
193 /// Sub-workflow runs below it that this request cancelled, oldest first.
194 /// Empty when none was still active.
195 pub cancelled_descendants: Vec<Uuid>,
196}
197
198/// Query parameters for listing runs.
199#[cfg_attr(feature = "openapi", derive(utoipa::IntoParams, utoipa::ToSchema))]
200#[derive(Debug, Deserialize)]
201pub struct ListRunsQuery {
202 /// Filter by workflow name.
203 pub workflow: Option<String>,
204 /// Filter by run status.
205 pub status: Option<RunStatus>,
206 /// Filter by step presence (only applies to completed/cancelled runs).
207 /// Non-terminal runs (pending, running, etc.) are always included.
208 /// When `true`, only return completed/cancelled runs that have steps.
209 /// When `false`, only return completed/cancelled runs without steps.
210 pub has_steps: Option<bool>,
211 /// Filter by labels. Comma-separated `key:value` pairs.
212 pub label: Option<String>,
213 /// Filter by author: the user ID that triggered the run.
214 ///
215 /// Also matches runs triggered by one of that user's API keys.
216 pub created_by: Option<Uuid>,
217 /// Filter by concurrency group: only runs that belong to this group.
218 pub concurrency_group: Option<String>,
219 /// Filter by priority: only runs with exactly this priority.
220 pub priority: Option<i16>,
221 /// Page number (1-based).
222 pub page: Option<u32>,
223 /// Items per page.
224 pub per_page: Option<u32>,
225}
226
227impl ListRunsQuery {
228 /// Parse the comma-separated `label` param into a `HashMap`.
229 pub fn parse_labels(&self) -> Option<HashMap<String, String>> {
230 parse_label_param(&self.label)
231 }
232}
233
234/// Parse a comma-separated `key:value` label query param into a `HashMap`.
235///
236/// Entries without a `:` are ignored. Returns `None` when the param is
237/// absent or contains no valid entry.
238pub(crate) fn parse_label_param(raw: &Option<String>) -> Option<HashMap<String, String>> {
239 raw.as_ref().and_then(|raw| {
240 let mut map = HashMap::new();
241 for entry in raw.split(',') {
242 let entry = entry.trim();
243 if let Some((k, v)) = entry.split_once(':') {
244 map.insert(k.to_string(), v.to_string());
245 }
246 }
247 if map.is_empty() { None } else { Some(map) }
248 })
249}