dataflow_rs/def/
reader.rs

1//! The compat reader: a pre-pass that rewrites v3 spellings into v4's, so
2//! `parse`'s walk reads one spelling.
3//!
4//! It runs over a copy of the authored JSON, only when
5//! [`ParseOptions::compat`](super::ParseOptions::compat) is on, and reports
6//! as it goes. Its issues come before the walk's, in document order: each
7//! object's keys in the order they were written (the crate turns on
8//! serde_json's `preserve_order`), the workflow first, then the steps of `loop`
9//! and `tasks` as their keys come, each node before its members.
10//!
11//! | Where | v3 spelling | v4 spelling |
12//! | --- | --- | --- |
13//! | Workflow | `continue_on_error` | `on_failure: continue` or `fail` |
14//! | Task | `continue_on_error`, `halt_on` | `on_failure` and `on_rejection`, by the policy table below |
15//! | `function.name` | `validate` | `validation` |
16//!
17//! Each rewrite is a `V3_SPELLING` Advisory at the v3 key, and a
18//! [`Rewrite`]. The integration configs (`http_call`, `enrich`,
19//! `publish_kafka`) are their hosts': their inputs are left as written. Two
20//! moves are not rewrites, and are reported as the keys they are:
21//!
22//! - **A key a built-in's input does not know** moves into the task's
23//!   `ext` as `function.input.<path>`, an `UNKNOWN_KEY` or a `LIKELY_TYPO`.
24//!   So does an unknown key of a `map` mapping or a `validation` rule that is
25//!   an object, as `function.input.mappings[0].<key>`. v3's configs ignored
26//!   them. The known keys are `keys.rs`'s tables, which equal the configs'
27//!   fields. An input or an element that is not an object is left for the
28//!   config to refuse.
29//! - **A group's `continue_on_error`** moves into the group's `ext`, a
30//!   `GROUP_CONTINUE_ON_ERROR` when it is `true` (v3 reported it and never
31//!   honoured it) and an `UNKNOWN_KEY` otherwise.
32//!
33//! The reader refuses a `continue_on_error` that is not a bool
34//! (`PARSE_FAILED`), a `halt_on` that is not `"never"` or `"failure"`
35//! (`INVALID_HALT_ON`, v3's message), and a v3 key written beside a v4 one
36//! (`MIXED_SPELLING` at the v3 key): on one node, any v3 policy key
37//! (`continue_on_error`, a task's `halt_on`) beside any v4 policy key
38//! (`on_failure`, a task's `on_rejection`). The two flags make one policy,
39//! so a v3 flag beside the v4 key of the other axis would be a guess at how
40//! the author meant them to combine. A refused key is taken out all the
41//! same, so the walk does not report it twice.
42
43use super::keys::{
44    BuiltinInput, Ext, TASK_POLICIES, TASK_V3_KEYS, WORKFLOW_POLICIES, builtin_input,
45};
46use super::parse::{descends, is_group, step_kind};
47use super::{V3_VALIDATE, VALIDATION};
48use crate::base::typo::unknown_key;
49use crate::base::{AuthoredPath, Issue, IssueCode, StepKind};
50use serde_json::{Map, Value};
51use std::collections::HashMap;
52
53/// One v3 spelling the compat reader rewrote.
54///
55/// `v3` and `v4` are short human-readable spellings, not JSON:
56///
57/// - a policy flag is `key: value`, and its v4 side lists the policies the
58///   reader wrote, such as `continue_on_error: true` read as
59///   `on_failure: continue, on_rejection: continue`. A task's two flags are
60///   read together, so each of its rewrites names the same policies (a flag
61///   beside a v4 policy is refused, not rewritten);
62/// - a function name is the bare name: `validate` read as `validation`.
63#[derive(Clone, Debug, PartialEq, Eq)]
64#[non_exhaustive]
65pub struct Rewrite {
66    /// The v3 key's authored path, such as `tasks[0].continue_on_error`
67    /// or `tasks[0].function.name`.
68    pub path: AuthoredPath,
69    /// What it was read as, in a form a host can apply.
70    pub kind: RewriteKind,
71}
72
73/// What a [`Rewrite`] changed: the v4 keys and values the reader wrote, so
74/// a host that migrates a stored definition applies them instead of parsing
75/// [`Display`](std::fmt::Display)'s prose.
76#[derive(Clone, Debug, PartialEq, Eq)]
77#[non_exhaustive]
78pub enum RewriteKind {
79    /// A v3 policy flag, read as the v4 policy keys written in its place.
80    Policy {
81        /// The v3 spelling as authored, such as `continue_on_error: true`;
82        /// display only.
83        v3: String,
84        /// The v4 keys written and their values, in the definition's order.
85        v4: Vec<(String, String)>,
86    },
87    /// A v3 function name, read as v4's: `validate` as `validation`.
88    Function {
89        /// The v3 name.
90        v3: String,
91        /// The v4 name.
92        v4: String,
93    },
94}
95
96impl std::fmt::Display for RewriteKind {
97    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
98        match self {
99            Self::Policy { v3, v4 } => {
100                let v4 = v4
101                    .iter()
102                    .map(|(key, value)| format!("{key}: {value}"))
103                    .collect::<Vec<_>>()
104                    .join(", ");
105                write!(f, "`{v3}` is a v3 spelling, read as `{v4}`")
106            }
107            Self::Function { v3, v4 } => {
108                write!(f, "`{v3}` is a v3 spelling, read as `{v4}`")
109            }
110        }
111    }
112}
113
114impl std::fmt::Display for Rewrite {
115    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
116        self.kind.fmt(f)
117    }
118}
119
120/// What the reader found: its issues, in document order, its rewrites, and
121/// the `ext` of each node it moved a key into, by the node's path, for the
122/// walk to go on building.
123pub(super) struct Read {
124    pub(super) issues: Vec<Issue>,
125    pub(super) rewrites: Vec<Rewrite>,
126    pub(super) exts: HashMap<AuthoredPath, Ext>,
127}
128
129/// Rewrites the v3 spellings in `workflow`, in place.
130pub(super) fn read(workflow: &mut Map<String, Value>) -> Read {
131    let mut reader = Read::new();
132    reader.workflow(workflow);
133    reader
134}
135
136/// Rewrites the v3 spellings in the steps of a list at `at`, in place, as
137/// [`read`] rewrites a workflow's: under `loop.setup` when `setup` is set,
138/// inside `depth` groups.
139pub(super) fn read_steps(
140    items: &mut [Value],
141    at: &AuthoredPath,
142    setup: bool,
143    depth: usize,
144) -> Read {
145    let mut reader = Read::new();
146    reader.steps(items, at, setup, depth);
147    reader
148}
149
150/// The step an issue sits in, which stamps it as the walk's issues are.
151struct Step {
152    kind: StepKind,
153    id: Option<String>,
154}
155
156/// A v3 policy flag as read.
157enum Flag {
158    Absent,
159    /// A spelling the reader rewrites: its `key: value` text and whether it
160    /// is set (`true`, `"failure"`).
161    Valid {
162        text: String,
163        set: bool,
164    },
165    /// Refused, with the issue that says why.
166    Refused {
167        code: IssueCode,
168        message: String,
169    },
170}
171
172impl Flag {
173    fn is_set(&self) -> bool {
174        matches!(self, Self::Valid { set: true, .. })
175    }
176}
177
178/// Removes `key` from `obj`, keeping the order of the keys after it:
179/// `Map::remove` would swap the last key into its place, and the walk reports
180/// a node's unknown keys in its order.
181fn take(obj: &mut Map<String, Value>, key: &str) -> Option<Value> {
182    obj.shift_remove(key)
183}
184
185/// The keys of `obj` among `wanted`, in the object's order.
186fn keys_among(obj: &Map<String, Value>, wanted: &[&'static str]) -> Vec<&'static str> {
187    obj.keys()
188        .filter_map(|key| wanted.iter().copied().find(|w| *w == key))
189        .collect()
190}
191
192/// The v4 policy keys of `policies` that `obj` holds, in the table's order.
193fn policies_in(obj: &Map<String, Value>, policies: &[&'static str]) -> Vec<&'static str> {
194    policies
195        .iter()
196        .copied()
197        .filter(|key| obj.contains_key(*key))
198        .collect()
199}
200
201/// A v3 policy key written beside v4 policy keys on one `node`, whose v4
202/// spelling is its `policies`.
203fn mixed(v3: &str, v4: &[&str], node: &str, policies: &str) -> String {
204    let v4: Vec<String> = v4.iter().map(|key| format!("`{key}`")).collect();
205    format!(
206        "v3's `{v3}` and v4's {} on one {node}; write only the v4 {policies}",
207        v4.join(" and ")
208    )
209}
210
211impl Read {
212    fn new() -> Self {
213        Self {
214            issues: Vec::new(),
215            rewrites: Vec::new(),
216            exts: HashMap::new(),
217        }
218    }
219
220    fn report(
221        &mut self,
222        code: IssueCode,
223        path: AuthoredPath,
224        message: impl Into<String>,
225        step: Option<&Step>,
226        field: Option<&str>,
227    ) {
228        let mut issue = Issue::new(code, path, message);
229        if let Some(step) = step {
230            issue = issue.with_step(step.kind, step.id.as_deref());
231        }
232        if let Some(field) = field {
233            issue = issue.with_field(field);
234        }
235        self.issues.push(issue);
236    }
237
238    /// A rewrite: its `V3_SPELLING` and its record.
239    fn rewrite(
240        &mut self,
241        path: AuthoredPath,
242        kind: RewriteKind,
243        step: Option<&Step>,
244        field: Option<&str>,
245    ) {
246        let message = kind.to_string();
247        self.report(IssueCode::V3Spelling, path.clone(), message, step, field);
248        self.rewrites.push(Rewrite { path, kind });
249    }
250
251    fn workflow(&mut self, obj: &mut Map<String, Value>) {
252        const WANTED: &[&str] = &["continue_on_error", "loop", "tasks"];
253        let root = AuthoredPath::root();
254        for key in keys_among(obj, WANTED) {
255            match key {
256                "continue_on_error" => self.workflow_flag(obj, &root),
257                "loop" => {
258                    if let Some(Value::Object(config)) = obj.get_mut("loop")
259                        && let Some(Value::Array(setup)) = config.get_mut("setup")
260                    {
261                        self.steps(setup, &root.join("loop").join("setup"), true, 0);
262                    }
263                }
264                "tasks" => {
265                    if let Some(Value::Array(tasks)) = obj.get_mut("tasks") {
266                        self.steps(tasks, &root.join("tasks"), false, 0);
267                    }
268                }
269                _ => {}
270            }
271        }
272    }
273
274    /// A workflow's `continue_on_error`: its `on_failure`.
275    fn workflow_flag(&mut self, obj: &mut Map<String, Value>, at: &AuthoredPath) {
276        let Some(raw) = take(obj, "continue_on_error") else {
277            return;
278        };
279        let path = at.join("continue_on_error");
280        let Value::Bool(flag) = raw else {
281            self.report(
282                IssueCode::ParseFailed,
283                path,
284                "expected a boolean",
285                None,
286                None,
287            );
288            return;
289        };
290        let v4 = policies_in(obj, WORKFLOW_POLICIES);
291        if !v4.is_empty() {
292            let message = mixed("continue_on_error", &v4, "workflow", "policy");
293            self.report(IssueCode::MixedSpelling, path, message, None, None);
294            return;
295        }
296        let policy = if flag { "continue" } else { "fail" };
297        obj.insert("on_failure".into(), Value::String(policy.into()));
298        let (v3, v4) = (
299            format!("continue_on_error: {flag}"),
300            vec![("on_failure".to_string(), policy.to_string())],
301        );
302        self.rewrite(path, RewriteKind::Policy { v3, v4 }, None, None);
303    }
304
305    /// The steps of a list, as the walk visits them.
306    fn steps(&mut self, items: &mut [Value], at: &AuthoredPath, setup: bool, depth: usize) {
307        for (i, node) in items.iter_mut().enumerate() {
308            if let Value::Object(obj) = node {
309                self.step(obj, at.join(i), setup, depth);
310            }
311        }
312    }
313
314    fn step(&mut self, obj: &mut Map<String, Value>, at: AuthoredPath, setup: bool, depth: usize) {
315        let group = is_group(obj);
316        let step = Step {
317            kind: step_kind(obj, setup),
318            id: obj
319                .get("id")
320                .and_then(Value::as_str)
321                .filter(|id| !id.is_empty())
322                .map(str::to_string),
323        };
324        if group {
325            self.group(obj, &at, &step, setup, depth);
326        } else {
327            self.task(obj, &at, &step);
328        }
329    }
330
331    fn group(
332        &mut self,
333        obj: &mut Map<String, Value>,
334        at: &AuthoredPath,
335        step: &Step,
336        setup: bool,
337        depth: usize,
338    ) {
339        // A group too deep is not descended into, as the walk does not.
340        const WANTED: &[&str] = &["continue_on_error", "tasks"];
341        for key in keys_among(obj, WANTED) {
342            if key == "continue_on_error" {
343                self.group_flag(obj, at, step);
344            } else if descends(depth)
345                && let Some(Value::Array(members)) = obj.get_mut("tasks")
346            {
347                self.steps(members, &at.join("tasks"), setup, depth + 1);
348            }
349        }
350    }
351
352    /// A group's `continue_on_error`: moved into its `ext`, never honoured.
353    fn group_flag(&mut self, obj: &mut Map<String, Value>, at: &AuthoredPath, step: &Step) {
354        const KEY: &str = "continue_on_error";
355        let Some(value) = take(obj, KEY) else {
356            return;
357        };
358        let (code, message) = if value == Value::Bool(true) {
359            // v3's message, from `check_group_continue_on_error`.
360            let message = format!(
361                "group '{}' carries continue_on_error, which the engine does not honour — error \
362                 handling is per task and per workflow, and a group only gates a span. Put it \
363                 on the tasks inside the group, or on the workflow",
364                step.id.as_deref().unwrap_or_default()
365            );
366            (IssueCode::GroupContinueOnError, message)
367        } else {
368            let message = format!(
369                "unknown key `{KEY}`: a group has no failure policy (kept in `ext` as `{KEY}`)"
370            );
371            (IssueCode::UnknownKey, message)
372        };
373        let moved = Moved {
374            key: KEY,
375            name: KEY.to_string(),
376            value,
377            path: at.join(KEY),
378            field: None,
379        };
380        self.keep_in_ext(obj, at, moved, code, message, step);
381    }
382
383    fn task(&mut self, obj: &mut Map<String, Value>, at: &AuthoredPath, step: &Step) {
384        if !TASK_V3_KEYS.iter().any(|key| obj.contains_key(*key)) {
385            self.function(obj, at, step);
386            return;
387        }
388        const WANTED: &[&str] = &["continue_on_error", "halt_on", "function"];
389        let order = keys_among(obj, WANTED);
390        // Any v4 policy beside any v3 flag refuses every flag.
391        let v4_policies = policies_in(obj, TASK_POLICIES);
392        let refuse_mixed = |v3: &str| Flag::Refused {
393            code: IssueCode::MixedSpelling,
394            message: mixed(v3, &v4_policies, "task", "policies"),
395        };
396        let continue_on_error = match take(obj, "continue_on_error") {
397            None => Flag::Absent,
398            Some(Value::Bool(_)) if !v4_policies.is_empty() => refuse_mixed("continue_on_error"),
399            Some(Value::Bool(set)) => Flag::Valid {
400                text: format!("continue_on_error: {set}"),
401                set,
402            },
403            Some(_) => Flag::Refused {
404                code: IssueCode::ParseFailed,
405                message: "expected a boolean".into(),
406            },
407        };
408        let halt_on = match take(obj, "halt_on") {
409            None => Flag::Absent,
410            Some(raw) => match raw.as_str() {
411                Some("never" | "failure") if !v4_policies.is_empty() => refuse_mixed("halt_on"),
412                Some(spelling @ ("never" | "failure")) => Flag::Valid {
413                    text: format!("halt_on: {spelling}"),
414                    set: spelling == "failure",
415                },
416                // v3's message, from `check_shape`.
417                _ => Flag::Refused {
418                    code: IssueCode::InvalidHaltOn,
419                    message: "halt_on must be one of \"never\", \"failure\"".into(),
420                },
421            },
422        };
423
424        // 06's policy table: the two flags make one policy. A flag beside a
425        // v4 policy was refused above, so here neither policy is written.
426        let v4: Vec<(String, String)> = if v4_policies.is_empty() {
427            let (on_failure, on_rejection) = match (continue_on_error.is_set(), halt_on.is_set()) {
428                (false, false) => ("fail", "continue"),
429                (true, false) => ("continue", "continue"),
430                (false, true) => ("fail", "halt"),
431                (true, true) => ("halt", "halt"),
432            };
433            obj.insert("on_failure".into(), Value::String(on_failure.into()));
434            obj.insert("on_rejection".into(), Value::String(on_rejection.into()));
435            vec![
436                ("on_failure".to_string(), on_failure.to_string()),
437                ("on_rejection".to_string(), on_rejection.to_string()),
438            ]
439        } else {
440            Vec::new()
441        };
442
443        for key in order {
444            match key {
445                "function" => self.function(obj, at, step),
446                _ => {
447                    let flag = if key == "halt_on" {
448                        &halt_on
449                    } else {
450                        &continue_on_error
451                    };
452                    let path = at.join(key);
453                    match flag {
454                        Flag::Absent => {}
455                        Flag::Valid { text, .. } => {
456                            self.rewrite(
457                                path,
458                                RewriteKind::Policy {
459                                    v3: text.clone(),
460                                    v4: v4.clone(),
461                                },
462                                Some(step),
463                                None,
464                            );
465                        }
466                        Flag::Refused { code, message } => {
467                            self.report(*code, path, message.clone(), Some(step), None);
468                        }
469                    }
470                }
471            }
472        }
473    }
474
475    /// A task's function: `validate`, then its input by function name.
476    fn function(&mut self, obj: &mut Map<String, Value>, at: &AuthoredPath, step: &Step) {
477        let Some(Value::Object(slot)) = obj.get_mut("function") else {
478            return;
479        };
480        // Taken out of the task for the moment, so the task's authored `ext`
481        // can be read as its input changes; put back in its place below.
482        let mut fields = std::mem::take(slot);
483        let task = at;
484        let at = at.join("function");
485        let (is_validate, builtin) = match fields.get("name").and_then(Value::as_str) {
486            Some(V3_VALIDATE) => (true, builtin_input(VALIDATION)),
487            Some(name) => (false, builtin_input(name)),
488            None => (false, None),
489        };
490        for key in keys_among(&fields, &["input", "name"]) {
491            if key == "name" {
492                if is_validate {
493                    fields.insert("name".into(), Value::String(VALIDATION.into()));
494                    self.rewrite(
495                        at.join("name"),
496                        RewriteKind::Function {
497                            v3: V3_VALIDATE.into(),
498                            v4: VALIDATION.into(),
499                        },
500                        Some(step),
501                        None,
502                    );
503                }
504            } else if let Some(Value::Object(input)) = fields.get_mut("input")
505                && let Some(known) = builtin
506            {
507                self.builtin_input(obj, task, input, known, step);
508            }
509        }
510        if let Some(Value::Object(slot)) = obj.get_mut("function") {
511            *slot = fields;
512        }
513    }
514
515    /// A built-in's input, in its keys' order. A key the built-in does
516    /// not know moves into the `ext` of the task at `task`, and so does an
517    /// unknown key of an object element of its nested configs.
518    fn builtin_input(
519        &mut self,
520        obj: &Map<String, Value>,
521        task: &AuthoredPath,
522        input: &mut Map<String, Value>,
523        known: BuiltinInput,
524        step: &Step,
525    ) {
526        input.retain(|key, value| {
527            if !known.keys.contains(&key.as_str()) {
528                let at = StrayKey {
529                    key,
530                    field: key,
531                    below: AuthoredPath::root().join(key.as_str()),
532                };
533                self.stray(obj, task, at, known.keys, std::mem::take(value), step);
534                return false;
535            }
536            if let Some((array, element_keys)) = known.elements
537                && key == array
538                && let Value::Array(items) = value
539            {
540                for (i, item) in items.iter_mut().enumerate() {
541                    let Value::Object(element) = item else {
542                        continue;
543                    };
544                    element.retain(|stray, val| {
545                        if element_keys.contains(&stray.as_str()) {
546                            return true;
547                        }
548                        let at = StrayKey {
549                            key: stray,
550                            field: array,
551                            below: AuthoredPath::root()
552                                .join(array)
553                                .join(i)
554                                .join(stray.as_str()),
555                        };
556                        self.stray(obj, task, at, element_keys, std::mem::take(val), step);
557                        false
558                    });
559                }
560            }
561            true
562        });
563    }
564
565    /// A built-in input's key that its level does not know (`known` are the
566    /// level's keys), moved into the `ext` of the task at `task` as
567    /// `function.input` followed by the key's path below the input.
568    fn stray(
569        &mut self,
570        obj: &Map<String, Value>,
571        task: &AuthoredPath,
572        at: StrayKey<'_>,
573        known: &[&str],
574        value: Value,
575        step: &Step,
576    ) {
577        let StrayKey { key, field, below } = at;
578        let rel = AuthoredPath::root()
579            .join("function")
580            .join("input")
581            .extend(&below);
582        let name = rel.to_string();
583        let (code, message) = unknown_key(key, known);
584        let message = format!("{message} (kept in `ext` as `{name}`)");
585        let moved = Moved {
586            key,
587            path: task.extend(&rel),
588            field: Some(field),
589            name,
590            value,
591        };
592        self.keep_in_ext(obj, task, moved, code, message, step);
593    }
594
595    /// Puts a moved key into the `ext` of the node `obj` at `node`, which the
596    /// walk goes on building, and reports it; a collision is refused by
597    /// [`Ext::keep`], the walk's own rule.
598    fn keep_in_ext(
599        &mut self,
600        obj: &Map<String, Value>,
601        node: &AuthoredPath,
602        moved: Moved<'_>,
603        code: IssueCode,
604        message: String,
605        step: &Step,
606    ) {
607        let Moved {
608            key,
609            name,
610            value,
611            path,
612            field,
613        } = moved;
614        let ext = self
615            .exts
616            .entry(node.clone())
617            .or_insert_with(|| Ext::authored(obj.get("ext")));
618        match ext.keep(key, name, value, path.clone()) {
619            Ok(()) => self.report(code, path, message, Some(step), field),
620            Err(collision) => {
621                self.report(IssueCode::ExtCollision, path, collision, Some(step), field);
622            }
623        }
624    }
625}
626
627/// A built-in input's key that its level does not know.
628struct StrayKey<'k> {
629    /// The key as authored.
630    key: &'k str,
631    /// The input's field it sits in: the key itself, or the array holding
632    /// the element it sits in.
633    field: &'k str,
634    /// Its path below the input.
635    below: AuthoredPath,
636}
637
638/// A key moving into a node's `ext`.
639struct Moved<'k> {
640    /// The key as authored.
641    key: &'k str,
642    /// Its name in `ext`.
643    name: String,
644    value: Value,
645    path: AuthoredPath,
646    /// The config field, for a key under `function.input`.
647    field: Option<&'k str>,
648}
dataflow_rs/def/reader.rs

dataflow_rs/def/
reader.rs
