dataflow_rs/def/

types.rs

//! The definition types: plain data, with raw expressions.

use super::Rollout;
use crate::base::{AuthoredPath, StepKind};
use chrono::{DateTime, Utc};
use serde_json::{Map, Value};

/// The value [`parse`](super::parse) reads for an absent key, which
/// `Serialize` leaves out. A tool that shows a definition reads them here
/// rather than keeping its own copy.
pub mod defaults {
    /// A workflow's `priority`.
    pub const PRIORITY: u32 = 0;
    /// A workflow's `version`.
    pub const VERSION: u32 = 1;
    /// A workflow's `channel`.
    pub const CHANNEL: &str = "default";
    /// A loop's `init`.
    pub const INIT: i64 = 0;
    /// A loop's `increment`.
    pub const INCREMENT: i64 = 1;
    /// A `for_each`'s `max_concurrency`.
    pub const MAX_CONCURRENCY: usize = 1;
}

/// Raw JSONLogic as authored. Only a compiler turns it into an
/// [`Expr`](crate::Expr).
#[derive(Clone, Debug, PartialEq)]
#[non_exhaustive]
pub struct Logic(pub Value);

impl Logic {
    /// Wraps an authored expression.
    pub fn new(value: Value) -> Self {
        Self(value)
    }
}

/// The keys every node shares: a workflow, a task and a group. `name` is
/// optional everywhere, which relaxes v3.
#[derive(Clone, Debug, Default, PartialEq)]
#[non_exhaustive]
pub struct NodeHeader {
    /// A human-readable name.
    pub(crate) name: Option<String>,
    /// What the node is for.
    pub(crate) description: Option<String>,
    /// Host and UI data, kept verbatim and never read: the authored `ext`
    /// object, then the unknown keys the compat reader moved there.
    pub(crate) ext: Map<String, Value>,
}

impl NodeHeader {
    /// The human-readable name, if authored.
    #[must_use]
    pub fn name(&self) -> Option<&str> {
        self.name.as_deref()
    }

    /// What the node is for, if authored.
    #[must_use]
    pub fn description(&self) -> Option<&str> {
        self.description.as_deref()
    }

    /// Host and UI data, kept verbatim and never read: the authored `ext`
    /// object, then the unknown keys the compat reader moved there.
    #[must_use]
    pub fn ext(&self) -> &Map<String, Value> {
        &self.ext
    }
}

/// A workflow definition: a condition and a tree of steps.
///
/// Only [`parse`](super::parse) makes one, and nothing changes one after:
/// every field is read through its accessor, so a definition a compiler sees
/// has passed every rule `parse` checks, and the interpreter can trust them.
///
/// ```
/// use dataflow_rs::def::{ParseOptions, StepDef, parse};
/// use serde_json::json;
///
/// let json = json!({"id": "w", "priority": 5, "tasks": [
///     {"id": "t", "function": {"name": "log", "input": {"message": "hi"}}}
/// ]});
/// let def = parse(&json, ParseOptions::default()).def.unwrap();
/// assert_eq!((def.id(), def.priority()), ("w", 5));
/// let StepDef::Task(task) = &def.tasks()[0] else { unreachable!() };
/// assert_eq!(task.function().name(), "log");
/// ```
///
/// A field cannot be assigned:
///
/// ```compile_fail
/// # use dataflow_rs::def::{ParseOptions, parse};
/// # use serde_json::json;
/// let json = json!({"id": "w", "tasks": []});
/// let mut def = parse(&json, ParseOptions::default()).def.unwrap();
/// def.priority = 5;
/// ```
#[derive(Clone, Debug, PartialEq)]
#[non_exhaustive]
pub struct WorkflowDef {
    /// The workflow's id: a non-empty string.
    pub(crate) id: String,
    /// The name, description and `ext`.
    pub(crate) header: NodeHeader,
    /// The run order: lower runs first, and equal priorities keep their
    /// order. Default 0.
    pub(crate) priority: u32,
    /// Whether a message enters the workflow. Absent means true.
    pub(crate) condition: Option<Logic>,
    /// The steps, as authored: tasks and groups.
    pub(crate) tasks: Vec<StepDef>,
    /// What a task's failure that reaches the workflow does. Default `fail`.
    pub(crate) on_failure: WorkflowFailurePolicy,
    /// The routing channel. Default `"default"`, as in v3.
    pub(crate) channel: String,
    /// The lifecycle status. Default `active`.
    pub(crate) status: Status,
    /// The definition's version. Default 1.
    pub(crate) version: u32,
    /// The workflow's slice of the traffic on its channel. `None` runs for
    /// every message.
    pub(crate) rollout: Option<Rollout>,
    /// The bounded loop over the task list, if any.
    pub(crate) r#loop: Option<LoopDef>,
    /// Tags for the host's filtering.
    pub(crate) tags: Vec<String>,
    /// When the definition was created. Kept for stored definitions, never
    /// read.
    pub(crate) created_at: Option<DateTime<Utc>>,
    /// When the definition was last updated. Kept, never read.
    pub(crate) updated_at: Option<DateTime<Utc>>,
}

impl WorkflowDef {
    /// The workflow's id: a non-empty string.
    #[must_use]
    pub fn id(&self) -> &str {
        &self.id
    }

    /// The name, description and `ext`.
    #[must_use]
    pub fn header(&self) -> &NodeHeader {
        &self.header
    }

    /// The run order: lower runs first, and equal priorities keep their
    /// order. Default 0.
    #[must_use]
    pub fn priority(&self) -> u32 {
        self.priority
    }

    /// Whether a message enters the workflow. Absent means true.
    #[must_use]
    pub fn condition(&self) -> Option<&Logic> {
        self.condition.as_ref()
    }

    /// The steps, as authored: tasks and groups.
    #[must_use]
    pub fn tasks(&self) -> &[StepDef] {
        &self.tasks
    }

    /// What a task's failure that reaches the workflow does. Default `fail`.
    #[must_use]
    pub fn on_failure(&self) -> WorkflowFailurePolicy {
        self.on_failure
    }

    /// The routing channel. Default `"default"`, as in v3.
    #[must_use]
    pub fn channel(&self) -> &str {
        &self.channel
    }

    /// The lifecycle status. Default `active`.
    #[must_use]
    pub fn status(&self) -> Status {
        self.status
    }

    /// The definition's version. Default 1.
    #[must_use]
    pub fn version(&self) -> u32 {
        self.version
    }

    /// The workflow's slice of the traffic on its channel. `None` runs for
    /// every message.
    #[must_use]
    pub fn rollout(&self) -> Option<&Rollout> {
        self.rollout.as_ref()
    }

    /// The bounded loop over the task list, if any.
    #[must_use]
    pub fn r#loop(&self) -> Option<&LoopDef> {
        self.r#loop.as_ref()
    }

    /// Tags for the host's filtering.
    #[must_use]
    pub fn tags(&self) -> &[String] {
        &self.tags
    }

    /// When the definition was created. Kept for stored definitions, never
    /// read.
    #[must_use]
    pub fn created_at(&self) -> Option<DateTime<Utc>> {
        self.created_at
    }

    /// When the definition was last updated. Kept, never read.
    #[must_use]
    pub fn updated_at(&self) -> Option<DateTime<Utc>> {
        self.updated_at
    }

    /// Every step with its authored path and kind, pre-order: `loop.setup`
    /// first, then `tasks`, each group before its members.
    ///
    /// A task's kind is [`StepKind::Task`] and a group's [`StepKind::Group`];
    /// every step under `loop.setup`, group or task, is [`StepKind::Setup`].
    pub fn walk(&self) -> impl Iterator<Item = (AuthoredPath, StepKind, &StepDef)> {
        let mut out = Vec::new();
        if let Some(config) = &self.r#loop {
            let mut at = AuthoredPath::root().join("loop").join("setup");
            collect(&config.setup, &mut at, true, &mut out);
        }
        let mut at = AuthoredPath::root().join("tasks");
        collect(&self.tasks, &mut at, false, &mut out);
        out.into_iter()
    }
}

/// Appends `steps` and their members to `out`, pre-order.
fn collect<'d>(
    steps: &'d [StepDef],
    at: &mut AuthoredPath,
    setup: bool,
    out: &mut Vec<(AuthoredPath, StepKind, &'d StepDef)>,
) {
    for (i, step) in steps.iter().enumerate() {
        at.push(i);
        out.push((at.clone(), step.kind(setup), step));
        if let StepDef::Group(group) = step {
            at.push("tasks");
            collect(&group.tasks, at, setup, out);
            at.pop();
        }
        at.pop();
    }
}

/// One element of a step list. An element carrying a `tasks` key is a group.
///
/// Not `#[non_exhaustive]`: the step grammar is closed, and a match on it is
/// a decision about every kind of step.
#[derive(Clone, Debug, PartialEq)]
#[allow(
    clippy::large_enum_variant,
    reason = "a definition is parse-time data, never touched per message; a task \
              is larger than a group because its JSON values are insertion-ordered \
              maps (serde_json's `preserve_order`), and boxing it would cost every \
              host that builds or matches a step a `Box` for no runtime gain"
)]
pub enum StepDef {
    /// A task: one function call.
    Task(TaskDef),
    /// A group: a span of steps sharing one condition.
    Group(GroupDef),
}

impl StepDef {
    /// The step's id. Tasks and groups share one namespace.
    pub fn id(&self) -> &str {
        match self {
            Self::Task(task) => &task.id,
            Self::Group(group) => &group.id,
        }
    }

    /// The kind a walk stamps on the step: [`StepKind::Setup`] for any step
    /// in a setup list, `setup`, and otherwise its own.
    pub(crate) fn kind(&self, setup: bool) -> StepKind {
        match self {
            _ if setup => StepKind::Setup,
            Self::Task(_) => StepKind::Task,
            Self::Group(_) => StepKind::Group,
        }
    }
}

/// A task: one function call, with its outcome policies.
#[derive(Clone, Debug, PartialEq)]
#[non_exhaustive]
pub struct TaskDef {
    /// The step id, unique across the workflow's steps and setup.
    pub(crate) id: String,
    /// The name, description and `ext`.
    pub(crate) header: NodeHeader,
    /// Whether the task runs. Absent means true.
    pub(crate) condition: Option<Logic>,
    /// The function the task runs, and its raw input.
    pub(crate) function: FunctionRef,
    /// What a failure (a 5xx or an `Err`) does. Default `fail`.
    pub(crate) on_failure: FailurePolicy,
    /// What a rejection (a 4xx) does. Default `continue`.
    pub(crate) on_rejection: RejectionPolicy,
    /// Whether running the task ends the workflow, applied after the
    /// policies. Default false.
    pub(crate) terminal: bool,
    /// Run the function once per element of an array. Refused on a built-in.
    pub(crate) for_each: Option<ForEachDef>,
}

impl TaskDef {
    /// The step id, unique across the workflow's steps and setup.
    #[must_use]
    pub fn id(&self) -> &str {
        &self.id
    }

    /// The name, description and `ext`.
    #[must_use]
    pub fn header(&self) -> &NodeHeader {
        &self.header
    }

    /// Whether the task runs. Absent means true.
    #[must_use]
    pub fn condition(&self) -> Option<&Logic> {
        self.condition.as_ref()
    }

    /// The function the task runs, and its raw input.
    #[must_use]
    pub fn function(&self) -> &FunctionRef {
        &self.function
    }

    /// What a failure (a 5xx or an `Err`) does. Default `fail`.
    #[must_use]
    pub fn on_failure(&self) -> FailurePolicy {
        self.on_failure
    }

    /// What a rejection (a 4xx) does. Default `continue`.
    #[must_use]
    pub fn on_rejection(&self) -> RejectionPolicy {
        self.on_rejection
    }

    /// Whether running the task ends the workflow, applied after the
    /// policies. Default false.
    #[must_use]
    pub fn terminal(&self) -> bool {
        self.terminal
    }

    /// The task's fan-out, if any: its function runs once per element.
    #[must_use]
    pub fn for_each(&self) -> Option<&ForEachDef> {
        self.for_each.as_ref()
    }
}

/// A group: a span of steps sharing one condition, evaluated once on entry.
#[derive(Clone, Debug, PartialEq)]
#[non_exhaustive]
pub struct GroupDef {
    /// The step id, unique across the workflow's steps and setup.
    pub(crate) id: String,
    /// The name, description and `ext`.
    pub(crate) header: NodeHeader,
    /// Whether the span runs. Absent means true.
    pub(crate) condition: Option<Logic>,
    /// Whether reaching the end of the group ends the workflow.
    pub(crate) terminal: bool,
    /// The members: never empty, nested at most
    /// [`MAX_GROUP_DEPTH`](super::MAX_GROUP_DEPTH) deep.
    pub(crate) tasks: Vec<StepDef>,
}

impl GroupDef {
    /// The step id, unique across the workflow's steps and setup.
    #[must_use]
    pub fn id(&self) -> &str {
        &self.id
    }

    /// The name, description and `ext`.
    #[must_use]
    pub fn header(&self) -> &NodeHeader {
        &self.header
    }

    /// Whether the span runs. Absent means true.
    #[must_use]
    pub fn condition(&self) -> Option<&Logic> {
        self.condition.as_ref()
    }

    /// Whether reaching the end of the group ends the workflow.
    #[must_use]
    pub fn terminal(&self) -> bool {
        self.terminal
    }

    /// The members: never empty, nested at most
    /// [`MAX_GROUP_DEPTH`](super::MAX_GROUP_DEPTH) deep.
    #[must_use]
    pub fn tasks(&self) -> &[StepDef] {
        &self.tasks
    }
}

/// A workflow's bounded loop: the task list runs once per sweep while
/// `counter < max`.
#[derive(Clone, Debug, PartialEq)]
#[non_exhaustive]
pub struct LoopDef {
    /// The `temp_data` slot the engine writes the counter to, if any.
    pub(crate) counter: Option<String>,
    /// The first counter value. Default 0.
    pub(crate) init: i64,
    /// Added to the counter after each sweep: at least 1. Default 1.
    pub(crate) increment: i64,
    /// The exclusive upper bound, above `init`. Required.
    pub(crate) max: i64,
    /// Steps run once, before the first sweep, with no counter.
    pub(crate) setup: Vec<StepDef>,
    /// The array iterated, evaluated once after setup.
    pub(crate) over: Option<Logic>,
    /// The `temp_data` slot holding `over[counter]`. Requires `over`.
    pub(crate) r#as: Option<String>,
    /// The `temp_data` slot reset to `{}` before every sweep.
    pub(crate) scratch: Option<String>,
}

impl LoopDef {
    /// The `temp_data` slot the engine writes the counter to, if any.
    #[must_use]
    pub fn counter(&self) -> Option<&str> {
        self.counter.as_deref()
    }

    /// The first counter value. Default 0.
    #[must_use]
    pub fn init(&self) -> i64 {
        self.init
    }

    /// Added to the counter after each sweep: at least 1. Default 1.
    #[must_use]
    pub fn increment(&self) -> i64 {
        self.increment
    }

    /// The exclusive upper bound, above `init`.
    #[must_use]
    pub fn max(&self) -> i64 {
        self.max
    }

    /// Steps run once, before the first sweep, with no counter.
    #[must_use]
    pub fn setup(&self) -> &[StepDef] {
        &self.setup
    }

    /// The array iterated, evaluated once after setup.
    #[must_use]
    pub fn over(&self) -> Option<&Logic> {
        self.over.as_ref()
    }

    /// The `temp_data` slot holding `over[counter]`. Only with `over`.
    #[must_use]
    pub fn r#as(&self) -> Option<&str> {
        self.r#as.as_deref()
    }

    /// The `temp_data` slot reset to `{}` before every sweep.
    #[must_use]
    pub fn scratch(&self) -> Option<&str> {
        self.scratch.as_deref()
    }
}

/// A task's fan-out: its function runs once per element of an array.
#[derive(Clone, Debug, PartialEq)]
#[non_exhaustive]
pub struct ForEachDef {
    /// The array, evaluated once after the task's condition.
    pub(crate) over: Logic,
    /// The `temp_data` slot holding the element; its index is at
    /// `temp_data.<as>_index`.
    pub(crate) r#as: String,
    /// Calls in flight at once: timing only, never results. Default 1.
    pub(crate) max_concurrency: usize,
    /// The context path each call writes its result to. Requires `into`.
    pub(crate) collect: Option<String>,
    /// The context path receiving the results in element order. Requires
    /// `collect`.
    pub(crate) into: Option<String>,
}

impl ForEachDef {
    /// The array, evaluated once after the task's condition.
    #[must_use]
    pub fn over(&self) -> &Logic {
        &self.over
    }

    /// The `temp_data` slot holding the element; its index is at
    /// `temp_data.<as>_index`.
    #[must_use]
    pub fn r#as(&self) -> &str {
        &self.r#as
    }

    /// Calls in flight at once: timing only, never results. Default 1.
    #[must_use]
    pub fn max_concurrency(&self) -> usize {
        self.max_concurrency
    }

    /// The context path each call writes its result to. Only with `into`.
    #[must_use]
    pub fn collect(&self) -> Option<&str> {
        self.collect.as_deref()
    }

    /// The context path receiving the results in element order. Only with
    /// `collect`.
    #[must_use]
    pub fn into(&self) -> Option<&str> {
        self.into.as_deref()
    }
}

/// The function a task runs.
#[derive(Clone, Debug, PartialEq)]
#[non_exhaustive]
pub struct FunctionRef {
    /// The registered name.
    pub(crate) name: String,
    /// The config, raw until a compiler reads it. Default `{}`.
    pub(crate) input: Value,
}

impl FunctionRef {
    /// The registered name.
    #[must_use]
    pub fn name(&self) -> &str {
        &self.name
    }

    /// The config, raw until a compiler reads it. Default `{}`.
    #[must_use]
    pub fn input(&self) -> &Value {
        &self.input
    }
}

/// A spelling table for an enum read from a string.
pub(super) trait Spelling: Copy + Default + 'static {
    /// Every variant, in the order the messages list them.
    const ALL: &'static [Self];

    /// The variant's authored spelling.
    fn spelling(self) -> &'static str;

    /// The variant spelled `text`, if any.
    fn from_spelling(text: &str) -> Option<Self> {
        Self::ALL.iter().copied().find(|v| v.spelling() == text)
    }
}

/// A workflow's lifecycle status.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash)]
#[non_exhaustive]
pub enum Status {
    /// Runs. The default.
    #[default]
    Active,
    /// Kept, not run.
    Paused,
    /// Retired.
    Archived,
}

impl Status {
    /// The authored spelling: `"active"`, `"paused"` or `"archived"`.
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Active => "active",
            Self::Paused => "paused",
            Self::Archived => "archived",
        }
    }
}

impl Spelling for Status {
    const ALL: &'static [Self] = &[Self::Active, Self::Paused, Self::Archived];
    fn spelling(self) -> &'static str {
        self.as_str()
    }
}

/// What a task's failure (a 5xx or an `Err`) does.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash)]
#[non_exhaustive]
pub enum FailurePolicy {
    /// The failure passes to the workflow's
    /// [`on_failure`](WorkflowDef::on_failure). The default.
    #[default]
    Fail,
    /// The workflow stops, and the run goes on.
    Halt,
    /// The failure is recorded and the next task runs.
    Continue,
}

impl FailurePolicy {
    /// The authored spelling: `"fail"`, `"halt"` or `"continue"`.
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Fail => "fail",
            Self::Halt => "halt",
            Self::Continue => "continue",
        }
    }
}

impl Spelling for FailurePolicy {
    const ALL: &'static [Self] = &[Self::Fail, Self::Halt, Self::Continue];
    fn spelling(self) -> &'static str {
        self.as_str()
    }
}

/// What a task's rejection (a 4xx) does.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash)]
#[non_exhaustive]
pub enum RejectionPolicy {
    /// The rejection is recorded and the next task runs. The default.
    #[default]
    Continue,
    /// The workflow stops, and the run goes on.
    Halt,
}

impl RejectionPolicy {
    /// The authored spelling: `"continue"` or `"halt"`.
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Continue => "continue",
            Self::Halt => "halt",
        }
    }
}

impl Spelling for RejectionPolicy {
    const ALL: &'static [Self] = &[Self::Continue, Self::Halt];
    fn spelling(self) -> &'static str {
        self.as_str()
    }
}

/// What a failure that reaches the workflow does.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash)]
#[non_exhaustive]
pub enum WorkflowFailurePolicy {
    /// The run stops. The default.
    #[default]
    Fail,
    /// The next sweep, or the next workflow, runs.
    Continue,
}

impl WorkflowFailurePolicy {
    /// The authored spelling: `"fail"` or `"continue"`.
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Fail => "fail",
            Self::Continue => "continue",
        }
    }
}

impl Spelling for WorkflowFailurePolicy {
    const ALL: &'static [Self] = &[Self::Fail, Self::Continue];
    fn spelling(self) -> &'static str {
        self.as_str()
    }
}