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///
39/// `#[non_exhaustive]`: groups are produced by the workflow parser, never built
40/// by hand — `end` is an index into the *flattened* task list and means nothing
41/// on its own. Read the fields freely.
42#[derive(Clone, Debug)]
43#[non_exhaustive]
44pub struct TaskGroup {
45 /// Unique identifier for the group within the workflow. Shares the task
46 /// id namespace, so a group cannot reuse a task's id.
47 pub id: String,
48
49 /// Human-readable name.
50 pub name: Option<String>,
51
52 /// Optional description explaining what the group covers.
53 pub description: Option<String>,
54
55 /// JSONLogic condition gating the whole span. Defaults to `true`.
56 pub condition: Value,
57
58 /// Engine-internal: pre-compiled JSONLogic for `condition`, populated by
59 /// `LogicCompiler`. `None` is treated as "always enter" by the executor.
60 /// Not part of the stable API.
61 #[doc(hidden)]
62 pub compiled_condition: Option<Arc<Logic>>,
63
64 /// Whether reaching the end of this group ends the workflow.
65 pub terminal: bool,
66
67 /// Engine-internal: exclusive end of the span, as an index into
68 /// `Workflow::tasks`. The start is the index of the task carrying this
69 /// entry in its [`Task::group_starts`]. Not part of the stable API.
70 #[doc(hidden)]
71 pub end: usize,
72}
73
74/// A single processing unit within a workflow (also known as an Action in rules-engine terminology).
75///
76/// Tasks execute functions with optional conditions and error handling.
77/// They are processed sequentially within a workflow, allowing later tasks
78/// to depend on results from earlier ones.
79///
80/// # Example JSON Definition
81///
82/// ```json
83/// {
84/// "id": "validate_user",
85/// "name": "Validate User Data",
86/// "description": "Ensures user data meets requirements",
87/// "condition": {">=": [{"var": "data.order.total"}, 1000]},
88/// "function": {
89/// "name": "validation",
90/// "input": { "rules": [...] }
91/// },
92/// "continue_on_error": false,
93/// "terminal": false
94/// }
95/// ```
96/// A single unit of work inside a workflow.
97///
98/// `#[non_exhaustive]`: construct through [`Task::action`] and assign the
99/// public fields you need, or parse a workflow from JSON. Field reads and
100/// writes are unaffected, and `..` patterns keep working.
101///
102/// The attribute exists because three of this struct's fields — `id_arc`,
103/// `compiled_condition`, `group_starts` — are engine internals documented as
104/// *not part of the stable API*, yet struct-literal construction forced every
105/// caller to name them. Field additions had already broken those callers twice
106/// (3.3.0, 3.6.0); this is the change that stops it.
107#[derive(Clone, Debug, Deserialize)]
108#[non_exhaustive]
109pub struct Task {
110 /// Unique identifier for the task within the workflow.
111 pub id: String,
112
113 /// Engine-internal: `Arc<str>` mirror of `id`, populated by
114 /// `LogicCompiler::compile_workflows`. Audit-trail emission clones this
115 /// instead of allocating a fresh `Arc<str>`. Public for crate-internal
116 /// access from the compiler and tests; not part of the stable API.
117 #[doc(hidden)]
118 #[serde(skip)]
119 pub id_arc: Arc<str>,
120
121 /// Human-readable name for the task.
122 pub name: String,
123
124 /// Optional description explaining what the task does.
125 pub description: Option<String>,
126
127 /// JSONLogic condition that determines if the task should execute.
128 /// Conditions can access any context field (`data`, `metadata`, `temp_data`).
129 /// Defaults to `true` (always execute).
130 #[serde(default = "crate::engine::utils::default_condition")]
131 pub condition: Value,
132
133 /// Engine-internal: pre-compiled JSONLogic for `condition`, populated by
134 /// `LogicCompiler`. `None` is treated as "always run" by the executor.
135 /// Not part of the stable API.
136 #[doc(hidden)]
137 #[serde(skip)]
138 pub compiled_condition: Option<Arc<Logic>>,
139
140 /// The function configuration specifying what operation to perform.
141 /// Can be a built-in function (map, validation) or a custom function.
142 pub function: FunctionConfig,
143
144 /// Whether to continue workflow execution if this task fails.
145 /// When `true`, errors are recorded but don't stop the workflow.
146 /// Defaults to `false`.
147 #[serde(default)]
148 pub continue_on_error: bool,
149
150 /// Whether running this task ends the workflow. Defaults to `false`.
151 ///
152 /// `terminal` is a statement about *position* — "nothing after this runs" —
153 /// not about outcome:
154 ///
155 /// - a false `condition` means the task never ran, so nothing halts;
156 /// - [`TaskOutcome::Skip`](crate::engine::task_outcome::TaskOutcome::Skip)
157 /// does not halt, for the same reason;
158 /// - a task that *failed* under `continue_on_error: true` still halts, and
159 /// its error is still recorded on `message.errors()`.
160 ///
161 /// Halting stops this workflow only; later workflows registered on the same
162 /// engine still process the message. Inside a workflow carrying a
163 /// [`LoopConfig`](crate::engine::workflow::LoopConfig) it breaks the whole
164 /// loop, not one sweep — the same scope as
165 /// [`TaskOutcome::Halt`](crate::engine::task_outcome::TaskOutcome::Halt).
166 ///
167 /// The audit-trail entry keeps the task's *own* status (`200`, `404`, …)
168 /// rather than `HALT_STATUS_CODE`: the task did its job, and a `map` that
169 /// wrote a 404 response body should not report "a filter halted here".
170 #[serde(default)]
171 pub terminal: bool,
172
173 /// Engine-internal: groups opening at this task, outermost first. Populated
174 /// by the workflow parser; empty for a task in no group. Not part of the
175 /// stable API.
176 #[doc(hidden)]
177 #[serde(skip)]
178 pub group_starts: Vec<TaskGroup>,
179}
180
181impl Task {
182 /// Create a task (action) with default settings.
183 ///
184 /// This is a convenience constructor for the IFTTT-style rules engine pattern,
185 /// creating an action that always executes (condition defaults to `true`).
186 ///
187 /// # Arguments
188 /// * `id` - Unique identifier for the action
189 /// * `name` - Human-readable name
190 /// * `function` - The function configuration to execute
191 pub fn action(id: &str, name: &str, function: FunctionConfig) -> Self {
192 Task {
193 id: id.to_string(),
194 id_arc: Arc::from(id),
195 name: name.to_string(),
196 description: None,
197 condition: Value::Bool(true),
198 compiled_condition: None,
199 function,
200 continue_on_error: false,
201 terminal: false,
202 group_starts: Vec::new(),
203 }
204 }
205}