use crate::engine::error::{DataflowError, Result};
use crate::engine::functions::integration::{EnrichConfig, HttpCallConfig, PublishKafkaConfig};
use crate::engine::functions::template::{Template, TemplateCompiler};
use crate::engine::functions::{
FilterConfig, LogConfig, MapConfig, ParseConfig, PublishConfig, ValidationConfig,
};
use crate::engine::secrets::{SECRET_OPERATOR, SecretOperator, Secrets};
use crate::engine::{FunctionConfig, Workflow};
use datalogic_rs::{CustomOperator, Engine, Logic};
use log::debug;
use serde_json::Value;
use std::collections::HashMap;
use std::sync::Arc;
pub(crate) const TEMPLATE_KEY_ESCAPE: char = '$';
pub(crate) fn datalogic_engine_builder() -> datalogic_rs::EngineBuilder {
Engine::builder()
.with_templating(true)
.with_template_key_escape(TEMPLATE_KEY_ESCAPE)
}
struct SharedOperator(Arc<dyn CustomOperator>);
impl CustomOperator for SharedOperator {
#[inline]
fn evaluate<'a>(
&self,
args: &[&'a datalogic_rs::DataValue<'a>],
ctx: &mut datalogic_rs::operator::EvalContext<'_, 'a>,
arena: &'a datalogic_rs::bumpalo::Bump,
) -> datalogic_rs::Result<&'a datalogic_rs::DataValue<'a>> {
self.0.evaluate(args, ctx, arena)
}
}
pub struct LogicCompiler {
engine: Arc<Engine>,
template_compiler: TemplateCompiler,
}
impl Default for LogicCompiler {
fn default() -> Self {
Self::new()
}
}
impl LogicCompiler {
pub fn new() -> Self {
Self::with_operators(&HashMap::new())
}
pub fn with_operators(operators: &HashMap<String, Arc<dyn CustomOperator>>) -> Self {
Self::with_operators_and_secrets(operators, &Arc::new(Secrets::empty()))
}
pub(crate) fn with_operators_and_secrets(
operators: &HashMap<String, Arc<dyn CustomOperator>>,
secrets: &Arc<Secrets>,
) -> Self {
let mut builder = datalogic_engine_builder();
for (name, op) in operators {
builder = builder.add_operator(name.clone(), SharedOperator(Arc::clone(op)));
}
builder = builder.add_operator(SECRET_OPERATOR, SecretOperator(Arc::clone(secrets)));
let engine = Arc::new(builder.build());
let template_compiler = TemplateCompiler::new(Arc::clone(&engine));
Self {
engine,
template_compiler,
}
}
pub fn engine(&self) -> Arc<Engine> {
Arc::clone(&self.engine)
}
pub fn into_engine(self) -> Arc<Engine> {
self.engine
}
pub fn compile_workflows(&self, workflows: Vec<Workflow>) -> Result<Vec<Workflow>> {
let mut compiled_workflows = Vec::with_capacity(workflows.len());
for mut workflow in workflows {
workflow.validate()?;
workflow.id_arc = Arc::from(workflow.id.as_str());
for task in &mut workflow.tasks {
task.id_arc = Arc::from(task.id.as_str());
}
if let Some(loop_config) = workflow.loop_config.as_mut() {
loop_config.precompute_counter_path();
}
let label = format!("workflow {} condition", workflow.id);
workflow.compiled_condition = self.compile_condition(&workflow.condition, &label)?;
debug!("Workflow {} condition compiled", workflow.id);
self.compile_workflow_tasks(&mut workflow)?;
workflow.fully_sync = workflow.tasks.iter().all(|t| t.function.is_sync_builtin());
compiled_workflows.push(workflow);
}
compiled_workflows.sort_by_key(|w| w.priority);
Ok(compiled_workflows)
}
fn compile_workflow_tasks(&self, workflow: &mut Workflow) -> Result<()> {
for task in &mut workflow.tasks {
for group in &mut task.group_starts {
let label = format!("group {} condition (workflow {})", group.id, workflow.id);
group.compiled_condition = self.compile_condition(&group.condition, &label)?;
}
let label = format!("task {} condition (workflow {})", task.id, workflow.id);
task.compiled_condition = self.compile_condition(&task.condition, &label)?;
self.compile_function_logic(&mut task.function, &task.id, &workflow.id)?;
}
Ok(())
}
fn compile_function_logic(
&self,
function: &mut FunctionConfig,
task_id: &str,
workflow_id: &str,
) -> Result<()> {
match function {
FunctionConfig::Map { input, .. } => {
self.compile_map_logic(input, task_id, workflow_id)
}
FunctionConfig::Validation { input, .. } => {
self.compile_validation_logic(input, task_id, workflow_id)
}
FunctionConfig::Filter { input, .. } => {
self.compile_filter_logic(input, task_id, workflow_id)
}
FunctionConfig::Log { input, .. } => {
self.compile_log_logic(input, task_id, workflow_id)
}
FunctionConfig::HttpCall { input, .. } => {
self.compile_http_call_logic(input, task_id, workflow_id)
}
FunctionConfig::Enrich { input, .. } => {
self.compile_enrich_logic(input, task_id, workflow_id)
}
FunctionConfig::PublishKafka { input, .. } => {
self.compile_publish_kafka_logic(input, task_id, workflow_id)
}
FunctionConfig::ParseJson { input, .. } | FunctionConfig::ParseXml { input, .. } => {
self.compile_parse_logic(input, task_id, workflow_id)
}
FunctionConfig::PublishJson { input, .. }
| FunctionConfig::PublishXml { input, .. } => {
self.compile_publish_logic(input, task_id, workflow_id)
}
_ => Ok(()),
}
}
fn compile(&self, logic: &Value, ctx_label: &str) -> Result<Arc<Logic>> {
self.engine
.compile_arc(logic)
.map_err(|e| DataflowError::LogicEvaluation(format!("{}: {}", ctx_label, e)))
}
fn compile_condition(&self, condition: &Value, ctx_label: &str) -> Result<Option<Arc<Logic>>> {
if matches!(condition, Value::Bool(true)) {
return Ok(None);
}
Ok(Some(self.compile(condition, ctx_label)?))
}
fn compile_parse_logic(
&self,
config: &mut ParseConfig,
task_id: &str,
workflow_id: &str,
) -> Result<()> {
self.compile_template(&mut config.source, "parse source", task_id, workflow_id)?;
config.target.compile(
&self.template_compiler,
&label("parse target", task_id, workflow_id),
)
}
fn compile_publish_logic(
&self,
config: &mut PublishConfig,
task_id: &str,
workflow_id: &str,
) -> Result<()> {
self.compile_template(&mut config.source, "publish source", task_id, workflow_id)?;
self.compile_template(
&mut config.root_element,
"publish root_element",
task_id,
workflow_id,
)?;
config.target.compile(
&self.template_compiler,
&label("publish target", task_id, workflow_id),
)
}
fn compile_map_logic(
&self,
config: &mut MapConfig,
task_id: &str,
workflow_id: &str,
) -> Result<()> {
for mapping in &mut config.mappings {
let path_label = format!("map path for task {task_id} in workflow {workflow_id}");
mapping.path.compile(&self.template_compiler, &path_label)?;
let label = format!(
"map logic for task {} in workflow {} (path {})",
task_id,
workflow_id,
mapping.describe_path()
);
mapping.compiled_logic = Some(self.compile(&mapping.logic, &label)?);
}
Ok(())
}
fn compile_validation_logic(
&self,
config: &mut ValidationConfig,
task_id: &str,
workflow_id: &str,
) -> Result<()> {
for (idx, rule) in config.rules.iter_mut().enumerate() {
let label = format!(
"validation rule {} for task {} in workflow {}",
idx, task_id, workflow_id
);
rule.compiled_logic = Some(self.compile(&rule.logic, &label)?);
let message_label = format!(
"validation rule {idx} message for task {task_id} in workflow {workflow_id}"
);
rule.message
.compile(&self.template_compiler, &message_label)?;
}
Ok(())
}
fn compile_log_logic(
&self,
config: &mut LogConfig,
task_id: &str,
workflow_id: &str,
) -> Result<()> {
let msg_label = label("log message", task_id, workflow_id);
config.compiled_message = Some(self.compile(&config.message, &msg_label)?);
let mut keys: Vec<&String> = config.fields.keys().collect();
keys.sort_unstable();
let mut compiled_fields = Vec::with_capacity(config.fields.len());
for key in keys {
let label = format!(
"log field '{}' for task {} in workflow {}",
key, task_id, workflow_id
);
compiled_fields.push((
key.clone(),
Some(self.compile(&config.fields[key], &label)?),
));
}
config.compiled_fields = compiled_fields;
Ok(())
}
fn compile_filter_logic(
&self,
config: &mut FilterConfig,
task_id: &str,
workflow_id: &str,
) -> Result<()> {
let label = label("filter condition", task_id, workflow_id);
config.compiled_condition = Some(self.compile(&config.condition, &label)?);
Ok(())
}
fn compile_http_call_logic(
&self,
config: &mut HttpCallConfig,
task_id: &str,
workflow_id: &str,
) -> Result<()> {
self.compile_template(
&mut config.connector,
"http_call connector",
task_id,
workflow_id,
)?;
self.compile_template(
&mut config.timeout_ms,
"http_call timeout_ms",
task_id,
workflow_id,
)?;
for (name, value) in &mut config.headers {
let what = format!("http_call header {name}");
self.compile_template(value, &what, task_id, workflow_id)?;
}
for (what, field) in [
("http_call path", &mut config.path),
("http_call body", &mut config.body),
("http_call body_format", &mut config.body_format),
("http_call response_path", &mut config.response_path),
("http_call response_format", &mut config.response_format),
] {
self.compile_template_field(field, what, task_id, workflow_id)?;
}
Ok(())
}
fn compile_enrich_logic(
&self,
config: &mut EnrichConfig,
task_id: &str,
workflow_id: &str,
) -> Result<()> {
self.compile_template(
&mut config.connector,
"enrich connector",
task_id,
workflow_id,
)?;
self.compile_template(
&mut config.merge_path,
"enrich merge_path",
task_id,
workflow_id,
)?;
self.compile_template(
&mut config.timeout_ms,
"enrich timeout_ms",
task_id,
workflow_id,
)?;
self.compile_template_field(&mut config.path, "enrich path", task_id, workflow_id)
}
fn compile_publish_kafka_logic(
&self,
config: &mut PublishKafkaConfig,
task_id: &str,
workflow_id: &str,
) -> Result<()> {
self.compile_template(
&mut config.connector,
"publish_kafka connector",
task_id,
workflow_id,
)?;
self.compile_template(
&mut config.topic,
"publish_kafka topic",
task_id,
workflow_id,
)?;
self.compile_template_field(&mut config.key, "publish_kafka key", task_id, workflow_id)?;
self.compile_template_field(
&mut config.value,
"publish_kafka value",
task_id,
workflow_id,
)
}
fn compile_template(
&self,
field: &mut Template,
what: &str,
task_id: &str,
workflow_id: &str,
) -> Result<()> {
field.compile(&self.template_compiler, &label(what, task_id, workflow_id))
}
fn compile_template_field(
&self,
field: &mut Option<Template>,
what: &str,
task_id: &str,
workflow_id: &str,
) -> Result<()> {
if let Some(t) = field {
t.compile(&self.template_compiler, &label(what, task_id, workflow_id))?;
}
Ok(())
}
}
fn label(what: &str, task_id: &str, workflow_id: &str) -> String {
format!("{what} for task {task_id} in workflow {workflow_id}")
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
fn engine() -> Engine {
crate::engine::compiler::datalogic_engine_builder().build()
}
fn eval(engine: &Engine, logic: &Value) -> Value {
let compiled = engine.compile_arc(logic).expect("should compile");
let ctx = datavalue::OwnedDataValue::from(&json!({}));
serde_json::from_str(
&engine
.session()
.eval_str(&compiled, &ctx)
.expect("should evaluate"),
)
.expect("eval_str output should be valid JSON")
}
fn workflow_json(extra: &str) -> String {
format!(
r#"{{ "id": "w", "name": "w", {extra}
"tasks": [{{"id": "t", "name": "t",
"function": {{"name": "map", "input": {{"mappings": []}}}}}}] }}"#
)
}
#[test]
fn compile_workflows_precomputes_the_loop_counter_path() {
let workflow =
Workflow::from_json(&workflow_json(r#""loop": {"counter": "i", "max": 3},"#))
.expect("should parse");
let compiled = LogicCompiler::new()
.compile_workflows(vec![workflow])
.expect("should compile");
let cfg = compiled[0].loop_config.as_ref().expect("loop config");
let parts: Vec<&str> = cfg.counter_parts.iter().map(Arc::as_ref).collect();
assert_eq!(parts, ["temp_data", "i"]);
}
#[test]
fn compile_workflows_rejects_an_invalid_loop_config() {
let workflow =
Workflow::from_json(&workflow_json(r#""loop": {"init": 5, "max": 5},"#)).unwrap();
assert!(
LogicCompiler::new()
.compile_workflows(vec![workflow])
.is_err()
);
}
#[test]
fn compile_workflows_leaves_a_non_looping_workflow_without_a_loop() {
let workflow = Workflow::from_json(&workflow_json("")).expect("should parse");
let compiled = LogicCompiler::new()
.compile_workflows(vec![workflow])
.expect("should compile");
assert!(compiled[0].loop_config.is_none());
}
#[test]
fn empty_operand_results_this_crate_would_silently_break_on() {
let e = engine();
for (logic, expected) in [
(json!({"and": []}), json!(null)),
(json!({"or": []}), json!(null)),
(json!({"+": []}), json!(0)),
(json!({"*": []}), json!(1)),
(json!({"cat": []}), json!("")),
(json!({"merge": []}), json!([])),
(json!({"missing": []}), json!([])),
] {
assert_eq!(eval(&e, &logic), expected, "for {logic}");
}
}
#[test]
fn a_missing_var_path_resolves_to_null_not_an_error() {
let e = engine();
assert_eq!(
eval(&e, &json!({"var": "data.does_not_exist"})),
json!(null)
);
}
#[test]
fn truthy_falsy_matches_the_documented_semantics() {
let e = engine();
for (v, truthy) in [
(json!(0), false),
(json!(""), false),
(json!(false), false),
(json!(null), false),
(json!([]), false),
(json!({}), false),
(json!("x"), true),
(json!(1), true),
] {
assert_eq!(
eval(&e, &json!({"!!": v})),
json!(truthy),
"truthiness of {v}"
);
}
}
#[test]
fn an_unrecognised_operator_is_not_an_error_under_templating() {
let e = engine();
let logic = json!({"totally_made_up_op_xyz": ["a", "b"]});
assert_eq!(
eval(&e, &logic),
logic,
"an unrecognised operator must echo back verbatim, not error"
);
}
#[cfg(not(feature = "ext-string"))]
#[test]
fn a_gated_operator_echoes_back_while_its_family_is_off() {
let e = engine();
let logic = json!({"starts_with": ["hello", "he"]});
assert_eq!(
eval(&e, &logic),
logic,
"an operator behind an unenabled family must echo back, not error"
);
}
#[cfg(feature = "ext-string")]
#[test]
fn a_gated_operator_evaluates_once_its_family_is_on() {
let e = engine();
assert_eq!(
eval(&e, &json!({"starts_with": ["hello", "he"]})),
json!(true),
"with ext-string on, starts_with must evaluate, not echo"
);
}
#[test]
fn datetime_feature_changes_plain_string_comparison() {
let e = engine();
let logic = json!({"==": ["2024-01-15T00:00:00Z", "2024-01-15T01:00:00+01:00"]});
#[cfg(feature = "datetime")]
assert_eq!(eval(&e, &logic), json!(true));
#[cfg(not(feature = "datetime"))]
assert_eq!(eval(&e, &logic), json!(false));
}
#[cfg(feature = "ext-string")]
#[test]
fn ext_string_feature_reaches_datalogic() {
let e = engine();
assert_eq!(eval(&e, &json!({"upper": "ab"})), json!("AB"));
}
#[cfg(feature = "ext-array")]
#[test]
fn ext_array_feature_reaches_datalogic() {
let e = engine();
assert_eq!(eval(&e, &json!({"sort": [[3, 1, 2]]})), json!([1, 2, 3]));
}
#[cfg(feature = "ext-math")]
#[test]
fn ext_math_feature_reaches_datalogic() {
let e = engine();
assert_eq!(eval(&e, &json!({"abs": -5})), json!(5));
}
#[cfg(feature = "ext-control")]
#[test]
fn ext_control_feature_reaches_datalogic() {
let e = engine();
assert_eq!(
eval(&e, &json!({"??": [null, "fallback"]})),
json!("fallback")
);
}
#[cfg(feature = "ext-object")]
#[test]
fn ext_object_feature_reaches_datalogic() {
let e = engine();
assert_eq!(
eval(&e, &json!({"keys": [{"a": 1, "b": 2}]})),
json!(["a", "b"])
);
}
#[cfg(feature = "error-handling")]
#[test]
fn error_handling_feature_reaches_datalogic() {
let e = engine();
assert_eq!(
eval(&e, &json!({"try": [{"throw": "boom"}, "recovered"]})),
json!("recovered")
);
}
#[cfg(feature = "datetime")]
#[test]
fn datetime_feature_reaches_datalogic() {
let e = engine();
let logic = json!({"now": []});
let result = eval(&e, &logic);
assert_ne!(result, logic, "with datetime on, `now` must not echo back");
assert!(!result.is_null(), "`now` should produce a value, got null");
}
}