dataflow_rs/engine/task.rs
1//! # Task Module
2//!
3//! This module defines the `Task` structure, which represents a single
4//! processing unit within a workflow. Tasks are the fundamental building
5//! blocks of data processing pipelines.
6
7use crate::engine::functions::FunctionConfig;
8use datalogic_rs::Logic;
9use serde::Deserialize;
10use serde_json::Value;
11use std::sync::Arc;
12
13/// A contiguous run of tasks sharing one condition, and optionally ending the
14/// workflow once the run completes.
15///
16/// Groups are authored as nested objects inside a workflow's `tasks` list — an
17/// element carrying a `tasks` key is a group, one carrying a `function` key is
18/// a [`Task`]. The parser flattens the tree into `Workflow::tasks` and records
19/// each group's span on the task that opens it ([`Task::group_starts`]), so the
20/// executor keeps walking a flat slice.
21///
22/// The condition is evaluated **once, on entry**. A false result skips every
23/// task in the span; the individual tasks' own conditions are not evaluated.
24///
25/// # Example JSON Definition
26///
27/// ```json
28/// {
29/// "id": "have_videos",
30/// "condition": {">": [{"length": [{"var": "temp_data.videos"}]}, 0]},
31/// "terminal": false,
32/// "tasks": [
33/// {"id": "rank", "name": "Rank", "function": {"name": "map", "input": {}}},
34/// {"id": "trim", "name": "Trim", "function": {"name": "map", "input": {}}}
35/// ]
36/// }
37/// ```
38#[derive(Clone, Debug)]
39pub struct TaskGroup {
40 /// Unique identifier for the group within the workflow. Shares the task
41 /// id namespace, so a group cannot reuse a task's id.
42 pub id: String,
43
44 /// Human-readable name.
45 pub name: Option<String>,
46
47 /// Optional description explaining what the group covers.
48 pub description: Option<String>,
49
50 /// JSONLogic condition gating the whole span. Defaults to `true`.
51 pub condition: Value,
52
53 /// Engine-internal: pre-compiled JSONLogic for `condition`, populated by
54 /// `LogicCompiler`. `None` is treated as "always enter" by the executor.
55 /// Not part of the stable API.
56 #[doc(hidden)]
57 pub compiled_condition: Option<Arc<Logic>>,
58
59 /// Whether reaching the end of this group ends the workflow.
60 pub terminal: bool,
61
62 /// Engine-internal: exclusive end of the span, as an index into
63 /// `Workflow::tasks`. The start is the index of the task carrying this
64 /// entry in its [`Task::group_starts`]. Not part of the stable API.
65 #[doc(hidden)]
66 pub end: usize,
67}
68
69/// A single processing unit within a workflow (also known as an Action in rules-engine terminology).
70///
71/// Tasks execute functions with optional conditions and error handling.
72/// They are processed sequentially within a workflow, allowing later tasks
73/// to depend on results from earlier ones.
74///
75/// # Example JSON Definition
76///
77/// ```json
78/// {
79/// "id": "validate_user",
80/// "name": "Validate User Data",
81/// "description": "Ensures user data meets requirements",
82/// "condition": {">=": [{"var": "data.order.total"}, 1000]},
83/// "function": {
84/// "name": "validation",
85/// "input": { "rules": [...] }
86/// },
87/// "continue_on_error": false,
88/// "terminal": false
89/// }
90/// ```
91#[derive(Clone, Debug, Deserialize)]
92pub struct Task {
93 /// Unique identifier for the task within the workflow.
94 pub id: String,
95
96 /// Engine-internal: `Arc<str>` mirror of `id`, populated by
97 /// `LogicCompiler::compile_workflows`. Audit-trail emission clones this
98 /// instead of allocating a fresh `Arc<str>`. Public for crate-internal
99 /// access from the compiler and tests; not part of the stable API.
100 #[doc(hidden)]
101 #[serde(skip)]
102 pub id_arc: Arc<str>,
103
104 /// Human-readable name for the task.
105 pub name: String,
106
107 /// Optional description explaining what the task does.
108 pub description: Option<String>,
109
110 /// JSONLogic condition that determines if the task should execute.
111 /// Conditions can access any context field (`data`, `metadata`, `temp_data`).
112 /// Defaults to `true` (always execute).
113 #[serde(default = "crate::engine::utils::default_condition")]
114 pub condition: Value,
115
116 /// Engine-internal: pre-compiled JSONLogic for `condition`, populated by
117 /// `LogicCompiler`. `None` is treated as "always run" by the executor.
118 /// Not part of the stable API.
119 #[doc(hidden)]
120 #[serde(skip)]
121 pub compiled_condition: Option<Arc<Logic>>,
122
123 /// The function configuration specifying what operation to perform.
124 /// Can be a built-in function (map, validation) or a custom function.
125 pub function: FunctionConfig,
126
127 /// Whether to continue workflow execution if this task fails.
128 /// When `true`, errors are recorded but don't stop the workflow.
129 /// Defaults to `false`.
130 #[serde(default)]
131 pub continue_on_error: bool,
132
133 /// Whether running this task ends the workflow. Defaults to `false`.
134 ///
135 /// `terminal` is a statement about *position* — "nothing after this runs" —
136 /// not about outcome:
137 ///
138 /// - a false `condition` means the task never ran, so nothing halts;
139 /// - [`TaskOutcome::Skip`](crate::engine::task_outcome::TaskOutcome::Skip)
140 /// does not halt, for the same reason;
141 /// - a task that *failed* under `continue_on_error: true` still halts, and
142 /// its error is still recorded on `message.errors()`.
143 ///
144 /// Halting stops this workflow only; later workflows registered on the same
145 /// engine still process the message. Inside a workflow carrying a
146 /// [`LoopConfig`](crate::engine::workflow::LoopConfig) it breaks the whole
147 /// loop, not one sweep — the same scope as
148 /// [`TaskOutcome::Halt`](crate::engine::task_outcome::TaskOutcome::Halt).
149 ///
150 /// The audit-trail entry keeps the task's *own* status (`200`, `404`, …)
151 /// rather than `HALT_STATUS_CODE`: the task did its job, and a `map` that
152 /// wrote a 404 response body should not report "a filter halted here".
153 #[serde(default)]
154 pub terminal: bool,
155
156 /// Engine-internal: groups opening at this task, outermost first. Populated
157 /// by the workflow parser; empty for a task in no group. Not part of the
158 /// stable API.
159 #[doc(hidden)]
160 #[serde(skip)]
161 pub group_starts: Vec<TaskGroup>,
162}
163
164impl Task {
165 /// Create a task (action) with default settings.
166 ///
167 /// This is a convenience constructor for the IFTTT-style rules engine pattern,
168 /// creating an action that always executes (condition defaults to `true`).
169 ///
170 /// # Arguments
171 /// * `id` - Unique identifier for the action
172 /// * `name` - Human-readable name
173 /// * `function` - The function configuration to execute
174 pub fn action(id: &str, name: &str, function: FunctionConfig) -> Self {
175 Task {
176 id: id.to_string(),
177 id_arc: Arc::from(id),
178 name: name.to_string(),
179 description: None,
180 condition: Value::Bool(true),
181 compiled_condition: None,
182 function,
183 continue_on_error: false,
184 terminal: false,
185 group_starts: Vec::new(),
186 }
187 }
188}
189
190/// Parsing for a workflow's `tasks` list, which holds *steps* rather than
191/// plain tasks: an element carrying a `tasks` key is a [`TaskGroup`], anything
192/// else is a [`Task`].
193///
194/// The tree is flattened into the `Vec<Task>` the executor walks, with each
195/// group's span recorded on the task that opens it. Deliberately **not**
196/// `#[serde(untagged)]`: an untagged enum reports *"data did not match any
197/// variant"*, which would replace the precise `missing field 'function'` that
198/// makes a malformed task diagnosable at `Engine::build()` time.
199pub(crate) mod steps {
200 use super::{Task, TaskGroup};
201 use serde::Deserialize;
202 use serde::de::{Deserializer, Error as DeError};
203 use serde_json::Value;
204
205 /// Maximum group nesting. Deeper than this is a generated-JSON accident
206 /// rather than an authored control-flow shape, and the bound keeps the
207 /// per-task `group_starts` vector trivially small.
208 const MAX_GROUP_DEPTH: usize = 8;
209
210 /// The non-`tasks` half of a group element. `tasks` is carried too so the
211 /// whole element deserializes in one pass; unknown keys are ignored, as
212 /// everywhere else in the workflow schema.
213 #[derive(Deserialize)]
214 struct GroupHeader {
215 id: String,
216 #[serde(default)]
217 name: Option<String>,
218 #[serde(default)]
219 description: Option<String>,
220 #[serde(default = "crate::engine::utils::default_condition")]
221 condition: Value,
222 #[serde(default)]
223 terminal: bool,
224 tasks: Vec<Value>,
225 }
226
227 /// `deserialize_with` target for `Workflow::tasks`.
228 pub(crate) fn flatten<'de, D>(deserializer: D) -> Result<Vec<Task>, D::Error>
229 where
230 D: Deserializer<'de>,
231 {
232 let steps = Vec::<Value>::deserialize(deserializer)?;
233 let mut tasks = Vec::with_capacity(steps.len());
234 walk(&steps, 0, &mut tasks).map_err(D::Error::custom)?;
235 Ok(tasks)
236 }
237
238 /// Append `steps` to `out` in document order, recording group spans.
239 fn walk(steps: &[Value], depth: usize, out: &mut Vec<Task>) -> Result<(), String> {
240 for step in steps {
241 let is_group = step.get("tasks").is_some();
242 if !is_group {
243 let task: Task = serde_json::from_value(step.clone())
244 .map_err(|e| format!("invalid task in workflow tasks: {e}"))?;
245 out.push(task);
246 continue;
247 }
248
249 if depth >= MAX_GROUP_DEPTH {
250 return Err(format!(
251 "task groups nested deeper than {MAX_GROUP_DEPTH} levels"
252 ));
253 }
254
255 let header: GroupHeader = serde_json::from_value(step.clone())
256 .map_err(|e| format!("invalid task group in workflow tasks: {e}"))?;
257
258 let start = out.len();
259 walk(&header.tasks, depth + 1, out)?;
260 let end = out.len();
261 if end == start {
262 return Err(format!(
263 "task group '{}' contains no tasks — an empty group can only be a mistake",
264 header.id
265 ));
266 }
267
268 // Outermost first: an inner group nested at the same start index
269 // has already pushed its own entry, so this one goes in front of
270 // it. Bounded by `MAX_GROUP_DEPTH`, so the shift is trivial.
271 out[start].group_starts.insert(
272 0,
273 TaskGroup {
274 id: header.id,
275 name: header.name,
276 description: header.description,
277 condition: header.condition,
278 compiled_condition: None,
279 terminal: header.terminal,
280 end,
281 },
282 );
283 }
284 Ok(())
285 }
286}