use crate::engine::error::{DataflowError, Result};
use crate::engine::task_context::TaskContext;
use datalogic_rs::Logic;
use datavalue::OwnedDataValue;
use serde::{Deserialize, Deserializer};
use serde_json::Value;
use std::borrow::Cow;
use std::sync::Arc;
pub(crate) fn plain_string_of(value: &OwnedDataValue) -> String {
match value {
OwnedDataValue::String(s) => s.clone(),
other => other.to_string(),
}
}
#[derive(Debug, Clone)]
pub struct Template {
raw: Value,
compiled: Option<Box<Compiled>>,
}
#[derive(Debug, Clone)]
struct Compiled {
logic: Arc<Logic>,
constant: Option<OwnedDataValue>,
}
impl<'de> Deserialize<'de> for Template {
fn deserialize<D: Deserializer<'de>>(d: D) -> std::result::Result<Self, D::Error> {
Ok(Self {
raw: Value::deserialize(d)?,
compiled: None,
})
}
}
impl Default for Template {
fn default() -> Self {
Self::from(Value::Null)
}
}
impl From<Value> for Template {
fn from(raw: Value) -> Self {
Self {
raw,
compiled: None,
}
}
}
impl Template {
pub fn compile(&mut self, c: &TemplateCompiler, label: &str) -> Result<()> {
let compiled = c
.engine
.compile_arc(&self.raw)
.map_err(|e| DataflowError::LogicEvaluation(format!("{label}: {e}")))?;
let constant = if compiled.is_constant() {
let empty = OwnedDataValue::Object(Vec::new());
Some(
crate::engine::executor::eval_to_owned(&c.engine, &compiled, &empty)
.map_err(|e| DataflowError::LogicEvaluation(format!("{label}: {e}")))?,
)
} else {
None
};
self.compiled = Some(Box::new(Compiled {
logic: compiled,
constant,
}));
Ok(())
}
pub fn is_constant(&self) -> bool {
self.constant().is_some()
}
fn constant(&self) -> Option<&OwnedDataValue> {
self.compiled.as_ref().and_then(|c| c.constant.as_ref())
}
pub fn constant_string(&self) -> Option<String> {
self.constant().map(plain_string_of)
}
pub fn resolve(&self, ctx: &TaskContext<'_>) -> Result<OwnedDataValue> {
if let Some(v) = self.constant() {
return Ok(v.clone());
}
if let Some(v) = self.uncompiled_literal() {
return Ok(v);
}
self.eval(ctx)
}
fn uncompiled_literal(&self) -> Option<OwnedDataValue> {
if self.compiled.is_some() {
return None;
}
match &self.raw {
Value::String(_) | Value::Number(_) | Value::Bool(_) => {
Some(OwnedDataValue::from(&self.raw))
}
_ => None,
}
}
pub fn resolve_string(&self, ctx: &TaskContext<'_>) -> Result<String> {
if let Some(v) = self.constant() {
return Ok(plain_string_of(v));
}
if let Some(v) = self.uncompiled_literal() {
return Ok(plain_string_of(&v));
}
self.eval_to_plain_string(ctx)
}
pub(crate) fn resolve_string_in_arena(
&self,
p: crate::engine::functions::path_template::ParamCtx<'_>,
) -> Result<String> {
Ok(self.resolve_str_in_arena(p)?.into_owned())
}
pub(crate) fn resolve_str_in_arena(
&self,
p: crate::engine::functions::path_template::ParamCtx<'_>,
) -> Result<Cow<'_, str>> {
match self.constant() {
Some(OwnedDataValue::String(s)) => return Ok(Cow::Borrowed(s)),
Some(other) => return Ok(Cow::Owned(plain_string_of(other))),
None => {}
}
if self.compiled.is_none() {
if let Value::String(s) = &self.raw {
return Ok(Cow::Borrowed(s));
}
if let Some(v) = self.uncompiled_literal() {
return Ok(Cow::Owned(plain_string_of(&v)));
}
}
let logic = self.compiled_or_err("resolve_str_in_arena")?;
let evaluated = p
.engine()
.evaluate(logic, *p.context(), p.arena())
.map_err(|e| DataflowError::LogicEvaluation(e.to_string()))?;
Ok(Cow::Owned(match evaluated {
datavalue::DataValue::String(s) => s.to_string(),
other => other.to_string(),
}))
}
fn compiled_or_err(&self, method: &str) -> Result<&Logic> {
self.compiled.as_ref().map(|c| &*c.logic).ok_or_else(|| {
DataflowError::LogicEvaluation(format!(
"Template::{method} called before Template::compile — the engine did not \
compile this field at construction time"
))
})
}
pub fn resolve_u64(&self, ctx: &TaskContext<'_>, label: &str) -> Result<u64> {
let value = self.resolve(ctx)?;
match &value {
OwnedDataValue::Number(n) => {
let f = n.as_f64();
(f.is_finite() && f >= 0.0 && f <= u64::MAX as f64).then_some(f as u64)
}
_ => None,
}
.ok_or_else(|| {
DataflowError::Validation(format!(
"{label} must evaluate to a non-negative number, got {value}"
))
})
}
pub fn eval(&self, ctx: &TaskContext<'_>) -> Result<OwnedDataValue> {
let logic = self.compiled.as_ref().map(|c| &*c.logic).ok_or_else(|| {
DataflowError::LogicEvaluation(
"Template::eval called before Template::compile — the engine did not compile \
this field at construction time"
.to_string(),
)
})?;
ctx.eval(logic)
}
pub fn eval_into<T: serde::de::DeserializeOwned>(&self, ctx: &TaskContext<'_>) -> Result<T> {
let logic = self.compiled.as_ref().map(|c| &*c.logic).ok_or_else(|| {
DataflowError::LogicEvaluation(
"Template::eval_into called before Template::compile — the engine did not \
compile this field at construction time"
.to_string(),
)
})?;
let json = ctx.eval_json(logic)?;
serde_json::from_value(json).map_err(DataflowError::from_serde)
}
pub fn eval_to_plain_string(&self, ctx: &TaskContext<'_>) -> Result<String> {
let logic = self.compiled.as_ref().map(|c| &*c.logic).ok_or_else(|| {
DataflowError::LogicEvaluation(
"Template::eval_to_plain_string called before Template::compile — the engine \
did not compile this field at construction time"
.to_string(),
)
})?;
ctx.eval_to_plain_string(logic)
}
pub fn as_json(&self) -> &Value {
&self.raw
}
pub fn is_compiled(&self) -> bool {
self.compiled.is_some()
}
}
pub struct TemplateCompiler {
engine: Arc<datalogic_rs::Engine>,
}
impl TemplateCompiler {
pub(crate) fn new(engine: Arc<datalogic_rs::Engine>) -> Self {
Self { engine }
}
pub fn engine(&self) -> &datalogic_rs::Engine {
&self.engine
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::engine::message::Message;
use serde_json::json;
fn engine() -> Arc<datalogic_rs::Engine> {
Arc::new(crate::engine::compiler::datalogic_engine_builder().build())
}
fn template_from(v: Value) -> Template {
serde_json::from_value(v).unwrap()
}
#[test]
fn deserializes_from_every_json_shape_and_as_json_is_verbatim() {
for v in [
json!({"a": 1}),
json!([1, 2, 3]),
json!("hello"),
json!(42),
json!(true),
json!(null),
json!({}),
] {
let t = template_from(v.clone());
assert_eq!(t.as_json(), &v);
assert!(!t.is_compiled());
}
}
#[test]
fn eval_before_compile_errors_without_panicking() {
let dl = engine();
let mut m = Message::from_value(&json!({}));
let ctx = TaskContext::new(&mut m, &dl);
let t = template_from(json!({"var": "data.x"}));
match t.eval(&ctx) {
Err(DataflowError::LogicEvaluation(msg)) => {
assert!(
msg.contains("compile"),
"message should name the cause: {msg}"
);
}
other => panic!("expected LogicEvaluation, got {other:?}"),
}
}
#[test]
fn compile_on_a_malformed_expression_names_the_label() {
let c = TemplateCompiler::new(engine());
let mut too_deep = json!(1);
for _ in 0..300 {
too_deep = json!({"var": too_deep});
}
let mut t = template_from(too_deep);
match t.compile(&c, "my_field for task t in workflow w") {
Err(DataflowError::LogicEvaluation(msg)) => {
assert!(
msg.contains("my_field for task t in workflow w"),
"got: {msg}"
);
}
other => panic!("expected LogicEvaluation, got {other:?}"),
}
}
#[test]
fn a_literal_template_evaluates_to_that_literal() {
let dl = engine();
let c = TemplateCompiler::new(Arc::clone(&dl));
let mut m = Message::from_value(&json!({}));
let ctx = TaskContext::new(&mut m, &dl);
for v in [
json!("hello"),
json!(42),
json!({}),
json!({"a": 1, "b": 2}),
] {
let mut t = template_from(v.clone());
t.compile(&c, "lbl").unwrap();
assert_eq!(t.eval_into::<Value>(&ctx).unwrap(), v);
}
}
#[test]
fn an_operator_named_key_evaluates_unless_it_is_escaped() {
let dl = engine();
let c = TemplateCompiler::new(Arc::clone(&dl));
let mut m = Message::from_value(&json!({}));
let ctx = TaskContext::new(&mut m, &dl);
let mut op = template_from(json!({"cat": ["a", "b"]}));
op.compile(&c, "lbl").unwrap();
assert_eq!(op.eval_into::<Value>(&ctx).unwrap(), json!("ab"));
let mut escaped = template_from(json!({"$cat": ["a", "b"]}));
escaped.compile(&c, "lbl").unwrap();
assert_eq!(
escaped.eval_into::<Value>(&ctx).unwrap(),
json!({"cat": ["a", "b"]}),
"an escaped key must emit the literal object"
);
let mut doubled = template_from(json!({"$$oid": "abc"}));
doubled.compile(&c, "lbl").unwrap();
assert_eq!(
doubled.eval_into::<Value>(&ctx).unwrap(),
json!({"$oid": "abc"})
);
}
#[test]
fn the_static_spelling_of_every_parameter_folds_to_a_constant() {
let c = TemplateCompiler::new(engine());
for v in [
json!("data.output"),
json!(30000),
json!(true),
json!(["a", "b"]),
json!({"cat": ["a", "b"]}), ] {
let mut t = template_from(v.clone());
t.compile(&c, "lbl").unwrap();
assert!(t.is_constant(), "{v} should fold to a constant");
}
for v in [
json!({"var": "data.x"}),
json!({"cat": [{"var": "data.x"}]}),
] {
let mut t = template_from(v.clone());
t.compile(&c, "lbl").unwrap();
assert!(!t.is_constant(), "{v} must not fold");
}
}
#[test]
fn constant_and_evaluated_plain_strings_agree() {
let dl = engine();
let c = TemplateCompiler::new(Arc::clone(&dl));
let mut m = Message::from_value(&json!({}));
let ctx = TaskContext::new(&mut m, &dl);
for v in [
json!("abc"),
json!(7),
json!(true),
json!(null),
json!(["a", 1]),
] {
let mut t = template_from(v.clone());
t.compile(&c, "lbl").unwrap();
assert!(t.is_constant(), "{v} should fold");
assert_eq!(
t.resolve_string(&ctx).unwrap(),
t.eval_to_plain_string(&ctx).unwrap(),
"cached and evaluated coercion disagree for {v}"
);
}
}
#[test]
fn an_escaped_key_does_not_fold_to_a_constant() {
let dl = engine();
let c = TemplateCompiler::new(Arc::clone(&dl));
let mut m = Message::from_value(&json!({}));
let ctx = TaskContext::new(&mut m, &dl);
let mut t = template_from(json!({"$a": 1}));
t.compile(&c, "lbl").unwrap();
assert!(!t.is_constant(), "escaped keys are not folded today");
assert_eq!(t.eval_into::<Value>(&ctx).unwrap(), json!({"a": 1}));
assert_eq!(
t.resolve_string(&ctx).unwrap(),
t.eval_to_plain_string(&ctx).unwrap()
);
}
#[test]
fn resolve_u64_accepts_numbers_and_rejects_everything_else() {
let dl = engine();
let c = TemplateCompiler::new(Arc::clone(&dl));
let mut m = Message::from_value(&json!({}));
let ctx = TaskContext::new(&mut m, &dl);
let mut ok = template_from(json!(30000));
ok.compile(&c, "lbl").unwrap();
assert_eq!(ok.resolve_u64(&ctx, "timeout_ms").unwrap(), 30000);
for bad in [json!(null), json!("30000"), json!(-1), json!({"a": 1})] {
let mut t = template_from(bad.clone());
t.compile(&c, "lbl").unwrap();
let err = t
.resolve_u64(&ctx, "timeout_ms")
.expect_err("{bad} must be rejected");
assert!(err.to_string().contains("timeout_ms"), "{err}");
}
}
#[test]
fn eval_to_plain_string_unquotes_and_coerces_non_strings() {
let dl = engine();
let c = TemplateCompiler::new(Arc::clone(&dl));
let mut m = Message::from_value(&json!({}));
let ctx = TaskContext::new(&mut m, &dl);
let mut string_t = template_from(json!("abc"));
string_t.compile(&c, "lbl").unwrap();
assert_eq!(string_t.eval_to_plain_string(&ctx).unwrap(), "abc");
let mut num_t = template_from(json!(7));
num_t.compile(&c, "lbl").unwrap();
assert_eq!(num_t.eval_to_plain_string(&ctx).unwrap(), "7");
let mut obj_t = template_from(json!({"a": 1}));
obj_t.compile(&c, "lbl").unwrap();
assert_eq!(obj_t.eval_to_plain_string(&ctx).unwrap(), "{\"a\":1}");
}
#[test]
fn eval_to_plain_string_before_compile_errors_without_panicking() {
let mut m = Message::from_value(&json!({}));
let dl = engine();
let ctx = TaskContext::new(&mut m, &dl);
let t = template_from(json!("abc"));
match t.eval_to_plain_string(&ctx) {
Err(DataflowError::LogicEvaluation(msg)) => {
assert!(
msg.contains("compile"),
"message should name the cause: {msg}"
);
}
other => panic!("expected LogicEvaluation, got {other:?}"),
}
}
#[test]
fn non_ascii_result_round_trips() {
let dl = engine();
let c = TemplateCompiler::new(Arc::clone(&dl));
let mut m = Message::from_value(&json!({}));
let ctx = TaskContext::new(&mut m, &dl);
let mut t = template_from(json!({"cat": ["über-", "größe"]}));
t.compile(&c, "lbl").unwrap();
assert_eq!(t.eval_into::<String>(&ctx).unwrap(), "über-größe");
}
#[test]
fn reading_an_absent_path_matches_the_engines_missing_path_result() {
let dl = engine();
let c = TemplateCompiler::new(Arc::clone(&dl));
let mut m = Message::from_value(&json!({}));
let ctx = TaskContext::new(&mut m, &dl);
let mut t = template_from(json!({"var": "data.nope"}));
t.compile(&c, "lbl").unwrap();
assert_eq!(t.eval(&ctx).unwrap(), OwnedDataValue::Null);
}
}