dataflow-rs 4.0.1

//! `SyncCtx`: what a built-in sees when it runs.
//!
//! A built-in reads the message through the run view and writes through the
//! run context, so a write is spliced into the view and marked, never
//! converted, until the next checkpoint. Crate-private: a host implements
//! [`Function`](super::Function), whose calls go through the async chain.

use crate::base::{AuthoredPath, Change, ErrorRecord, ErrorSite, Result, SiteKind};
use crate::expr::arena::with_arena;
use crate::expr::{Expr, ExprEngine, FromValue, Path};
use crate::run::{ConditionErrorPolicy, TaskBudget};
use crate::state::RunContext;
use datavalue::{DataValue, OwnedDataValue};
use std::sync::Arc;

/// What a built-in sees: the run context, the run's facts, and the buffers
/// the recorder reads.
///
/// `'p` is the program's: the config, the expression engine and the ids live
/// as long as it does, so a result evaluated from the config can be spliced
/// into the view.
pub(crate) struct SyncCtx<'c, 'm, 'p> {
    /// The run context: the view to read, the writes to make.
    pub(crate) rc: &'c mut RunContext<'m, 'p>,
    /// The expression engine the program was compiled with.
    pub(crate) exprs: &'p ExprEngine,
    /// The run's shared task and expression op budget.
    pub(crate) budget: &'c TaskBudget<'c>,
    /// How condition evaluation errors are handled during this run.
    pub(crate) condition_error_policy: ConditionErrorPolicy,
    /// The message's payload.
    pub(crate) payload: &'c OwnedDataValue,
    /// The workflow's id, shared with the program's own `Arc<str>`.
    pub(crate) workflow_id: &'c Arc<str>,
    /// The task's id, shared as `workflow_id` is.
    pub(crate) task_id: &'c Arc<str>,
    /// The task's authored path in the workflow definition.
    pub(crate) task_path: &'p AuthoredPath,
    /// The current loop counter, when inside a workflow loop.
    pub(crate) loop_counter: Option<i64>,
    /// Whether `Change`s are recorded: only when the audit or the trace
    /// keeps them.
    pub(crate) capture: bool,
    /// Whether the run's trace keeps changes: only then does a built-in
    /// record a write the trace alone needs ([`Change::for_trace`]).
    pub(crate) traced: bool,
    /// This task's changes, read by the recorder.
    pub(crate) changes: &'c mut Vec<Change>,
    /// Set by a built-in that recorded a change the trace alone keeps, so
    /// the recorder makes the audit's copy without it, and only then.
    pub(crate) marked: bool,
    /// Where `map` notes each mapping's end in `changes`, lent by the
    /// trace only while it keeps changes; `None` otherwise, so a run that
    /// is not traced notes nothing.
    pub(crate) mapping_ends: Option<&'c mut Vec<u32>>,
    /// The message's error list.
    pub(crate) errors: &'c mut Vec<ErrorRecord>,
    /// The run's one clock reading.
    pub(crate) now: chrono::DateTime<chrono::Utc>,
    /// The run's clock reading, RFC 3339, for error records.
    pub(crate) now_text: &'c mut Option<Arc<str>>,
}

impl std::fmt::Debug for SyncCtx<'_, '_, '_> {
    /// The ids and the capture flag; not the context.
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SyncCtx")
            .field("workflow_id", &self.workflow_id)
            .field("task_id", &self.task_id)
            .field("capture", &self.capture)
            .finish_non_exhaustive()
    }
}

impl<'p> SyncCtx<'_, '_, 'p> {
    /// Read-only evaluation against the view, in a pooled arena; a folded
    /// constant is lent.
    ///
    /// # Errors
    ///
    /// [`ErrorKind::Budget`](crate::base::ErrorKind::Budget) when the
    /// operation ceiling is exceeded; [`ErrorKind::Eval`](crate::base::ErrorKind::Eval)
    /// for any other failure, including a result that is not a `T`.
    pub(crate) fn eval<'e, T: FromValue>(&mut self, expr: &'e Expr<T>) -> Result<T::Out<'e>> {
        eval(self.rc, self.exprs, self.budget, expr)
    }

    /// A condition: only a literal `true` passes; the error is returned for
    /// the caller to classify.
    pub(crate) fn test(&mut self, expr: &Expr) -> Result<bool> {
        test(self.rc, self.exprs, self.budget, expr)
    }

    /// Builds an [`ErrorSite`] stamped with this task's workflow, task id,
    /// and loop counter.
    pub(crate) fn site(&self, kind: SiteKind, path: AuthoredPath) -> ErrorSite {
        ErrorSite {
            kind,
            workflow_id: self.workflow_id.clone(),
            task_id: Some(self.task_id.clone()),
            group_id: None,
            rule_index: None,
            mapping_index: None,
            target_path: None,
            authored_path: Some(path),
            loop_counter: self.loop_counter,
        }
    }

    /// Writes `value` at `target`, recording its `Change` under capture.
    pub(crate) fn write(&mut self, target: &Path, value: OwnedDataValue, hint: Option<usize>) {
        let changes = self.capture.then_some(&mut *self.changes);
        self.rc.set_recorded(target, value, hint, changes);
    }

    /// An error record stamped with this task's ids and the run's timestamp.
    pub(crate) fn record(&mut self, code: impl Into<String>, message: String) -> ErrorRecord {
        let now = self
            .now_text
            .get_or_insert_with(|| Arc::from(self.now.to_rfc3339()));
        task_record(code, message, self.workflow_id, self.task_id, now)
    }
}

/// Evaluates `expr` against the view, in a pooled arena, so nothing it
/// allocates stays in the run arena: [`SyncCtx::eval`], for the engine's own
/// reads, such as `loop.over`. A folded constant is lent and reads no view.
pub(crate) fn eval<'e, T: FromValue>(
    rc: &mut RunContext<'_, '_>,
    exprs: &ExprEngine,
    budget: &TaskBudget<'_>,
    expr: &'e Expr<T>,
) -> Result<T::Out<'e>> {
    if let Some(c) = expr.constant() {
        return Ok(T::lend(c));
    }
    rc.read(|_, root| with_arena(|arena| budget.eval_view(exprs, expr, root, arena)))
}

pub(crate) fn test(
    rc: &mut RunContext<'_, '_>,
    exprs: &ExprEngine,
    budget: &TaskBudget<'_>,
    expr: &Expr,
) -> Result<bool> {
    if let Some(c) = expr.constant() {
        return Ok(matches!(c, OwnedDataValue::Bool(true)));
    }
    rc.read(|_, root| {
        with_arena(|arena| {
            budget
                .eval_in(exprs, expr.logic(), expr.reads_secret(), root, arena)
                .map(|(v, _)| matches!(v, DataValue::Bool(true)))
        })
    })
}

/// An error record carrying a task's ids and the run's timestamp, as v3 fills
/// them: what a built-in records, and what the recorder records for it.
pub(crate) fn task_record(
    code: impl Into<String>,
    message: String,
    workflow_id: &Arc<str>,
    task_id: &Arc<str>,
    now: &Arc<str>,
) -> ErrorRecord {
    let mut record = ErrorRecord::new(code, message);
    record.stamp(workflow_id, Some(task_id), now, None);
    record
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::base::ErrorKind;
    use crate::expr::arena;
    use crate::state::Tuning;
    use serde_json::json;
    use std::borrow::Cow;

    const NOW: &str = "2026-10-14T00:00:00+00:00";

    /// Runs `f` with a context over `{data: {a: 1, t: true, zero: 0}}`; the
    /// changes it recorded.
    fn with_cx(capture: bool, f: impl FnOnce(&mut SyncCtx<'_, '_, '_>)) -> Vec<Change> {
        let exprs = ExprEngine::default();
        let budget = TaskBudget::unlimited();
        let task_path = AuthoredPath::root();
        let mut slot = OwnedDataValue::from(&json!({
            "data": {"a": 1, "t": true, "zero": 0}, "metadata": {}, "temp_data": {}
        }));
        let (mut changes, mut errors) = (Vec::new(), Vec::new());
        let mut rc = RunContext::new(&mut slot, Tuning::default());
        let (workflow_id, task_id, mut now_text) =
            (Arc::from("w"), Arc::from("t"), Some(Arc::from(NOW)));
        f(&mut SyncCtx {
            rc: &mut rc,
            exprs: &exprs,
            budget: &budget,
            condition_error_policy: ConditionErrorPolicy::Skip,
            payload: &OwnedDataValue::Null,
            workflow_id: &workflow_id,
            task_id: &task_id,
            task_path: &task_path,
            loop_counter: None,
            capture,
            traced: false,
            changes: &mut changes,
            marked: false,
            mapping_ends: None,
            errors: &mut errors,
            now: chrono::Utc::now(),
            now_text: &mut now_text,
        });
        rc.finish();
        changes
    }

    fn expr<T: FromValue>(raw: serde_json::Value) -> Expr<T> {
        ExprEngine::default().compile(&raw).expect("compiles")
    }

    #[test]
    fn eval_lends_a_constant_and_reads_the_view_as_written() {
        with_cx(true, |cx| {
            let five: Expr = expr(json!(5));
            assert!(matches!(cx.eval(&five), Ok(Cow::Borrowed(_))));

            let a: Expr = expr(json!({"cat": ["a=", {"var": "data.a"}]}));
            let run_arena =
                |cx: &mut SyncCtx<'_, '_, '_>| cx.rc.read(|owner, _| arena::used(&owner.arena));
            let before = run_arena(cx);
            assert_eq!(*cx.eval(&a).unwrap(), OwnedDataValue::from("a=1"));
            assert_eq!(
                run_arena(cx),
                before,
                "a read leaves the run arena as it was"
            );

            let path: crate::expr::Path = "data.a".parse().unwrap();
            cx.rc
                .edit(|e| e.set(&path, OwnedDataValue::from(2).to_arena(e.arena()), None));
            assert_eq!(*cx.eval(&a).unwrap(), OwnedDataValue::from("a=2"));

            let not_a_u64: Expr<u64> = expr(json!({"var": "data.t"}));
            let error = cx.eval(&not_a_u64).unwrap_err();
            assert_eq!(*error.kind(), ErrorKind::Eval);
            assert!(error.message().contains("did not give a u64"), "{error}");
        });
    }

    #[test]
    fn test_passes_only_a_literal_true() {
        with_cx(true, |cx| {
            for (raw, passes) in [
                (json!(true), true),
                (json!(1), false),
                (json!("true"), false),
                (json!({"var": "data.t"}), true),
                (json!({"var": "data.a"}), false),
                (json!({"var": "data.missing"}), false),
            ] {
                assert_eq!(cx.test(&expr(raw.clone())).unwrap(), passes, "{raw}");
            }
            let fails: Expr = expr(json!({"/": [1, {"var": "data.zero"}]}));
            assert_eq!(*cx.test(&fails).unwrap_err().kind(), ErrorKind::Eval);
        });
    }

    #[test]
    fn task_record_carries_the_ids_and_the_runs_timestamp() {
        let record = task_record(
            "X_FAILED",
            "it failed".into(),
            &Arc::from("w"),
            &Arc::from("t"),
            &Arc::from(NOW),
        );
        assert_eq!(
            serde_json::to_string(&record).unwrap(),
            r#"{"code":"X_FAILED","message":"it failed","path":null,"workflow_id":"w","task_id":"t","timestamp":"2026-10-14T00:00:00+00:00"}"#
        );
    }

    #[test]
    fn a_write_records_its_change_only_under_capture() {
        for capture in [true, false] {
            let changes = with_cx(capture, |cx| {
                cx.write(&"data.a".parse().unwrap(), 2.into(), None);
                cx.write(&"data.b".parse().unwrap(), 3.into(), None);
            });
            if capture {
                assert_eq!(
                    serde_json::to_string(&changes).unwrap(),
                    r#"[{"path":"data.a","old_value":1,"new_value":2},{"path":"data.b","old_value":null,"new_value":3}]"#
                );
            } else {
                assert!(changes.is_empty());
            }
        }
    }
}