#[cfg(feature = "ir")]
use crate::expr::evaluate as evaluate_expression;
use crate::expr::ExprFailure;
#[cfg(feature = "ir")]
use crate::plan::{run_plan, ExecutionPlanSpec, OpSpec, RelationKind};
use crate::plan::{ExecOutcome, PlanFailure};
use crate::value::Value;
#[cfg(feature = "ir")]
use serde_json::Value as J;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum BehaviorFailureCode {
UnknownComponent,
UnknownNodeKind,
MapOverNotArray,
MapIntoElementNotObject,
MapBatchResultMismatch,
UnknownEntry,
}
impl BehaviorFailureCode {
pub fn as_str(self) -> &'static str {
match self {
BehaviorFailureCode::UnknownComponent => "UNKNOWN_COMPONENT",
BehaviorFailureCode::UnknownNodeKind => "UNKNOWN_NODE_KIND",
BehaviorFailureCode::MapOverNotArray => "MAP_OVER_NOT_ARRAY",
BehaviorFailureCode::MapIntoElementNotObject => "MAP_INTO_ELEMENT_NOT_OBJECT",
BehaviorFailureCode::MapBatchResultMismatch => "MAP_BATCH_RESULT_MISMATCH",
BehaviorFailureCode::UnknownEntry => "UNKNOWN_ENTRY",
}
}
}
#[derive(Debug, Clone)]
pub struct BehaviorError {
code: String,
pub message: String,
}
impl BehaviorError {
pub fn new(code: impl Into<String>, message: impl Into<String>) -> Self {
BehaviorError {
code: code.into(),
message: message.into(),
}
}
pub fn code(&self) -> &str {
&self.code
}
}
impl std::fmt::Display for BehaviorError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{}: {}", self.code, self.message)
}
}
impl std::error::Error for BehaviorError {}
impl From<ExprFailure> for BehaviorError {
fn from(e: ExprFailure) -> Self {
BehaviorError {
code: e.code.as_str().to_string(),
message: e.message,
}
}
}
impl From<PlanFailure> for BehaviorError {
fn from(e: PlanFailure) -> Self {
BehaviorError {
code: e.code.as_str().to_string(),
message: e.message,
}
}
}
#[cfg(feature = "ir")]
fn bfail<T>(code: BehaviorFailureCode, message: impl Into<String>) -> Result<T, BehaviorError> {
Err(BehaviorError {
code: code.as_str().to_string(),
message: message.into(),
})
}
pub trait ComponentExec {
fn exec(
&mut self,
component: &str,
ports: &[(String, Value)],
bound: Option<&Value>,
) -> Option<ExecOutcome>;
fn exec_ctx(
&mut self,
node_id: &str,
component: &str,
ports: &[(String, Value)],
bound: Option<&Value>,
) -> Option<ExecOutcome> {
let _ = node_id;
self.exec(component, ports, bound)
}
}
impl ComponentExec for &mut dyn ComponentExec {
fn exec(
&mut self,
component: &str,
ports: &[(String, Value)],
bound: Option<&Value>,
) -> Option<ExecOutcome> {
(**self).exec(component, ports, bound)
}
fn exec_ctx(
&mut self,
node_id: &str,
component: &str,
ports: &[(String, Value)],
bound: Option<&Value>,
) -> Option<ExecOutcome> {
(**self).exec_ctx(node_id, component, ports, bound)
}
}
#[cfg(feature = "ir")]
fn scope_get<'a>(scope: &'a [(String, Value)], key: &str) -> Option<&'a Value> {
scope.iter().find(|(k, _)| k == key).map(|(_, v)| v)
}
#[cfg(feature = "ir")]
fn node_kind(n: &J) -> Result<&'static str, BehaviorError> {
if n.get("map").is_some() {
Ok("map")
} else if n.get("cond").is_some() {
Ok("cond")
} else if n.get("component").is_some() {
Ok("componentRef")
} else {
bfail(
BehaviorFailureCode::UnknownNodeKind,
format!(
"body node '{}' is not componentRef/map/cond",
n.get("id").and_then(|v| v.as_str()).unwrap_or("?")
),
)
}
}
#[cfg(feature = "ir")]
fn node_parent(n: &J) -> Option<&str> {
let sub = if let Some(m) = n.get("map") {
m
} else if let Some(c) = n.get("cond") {
c
} else {
n
};
sub.get("parent").and_then(|v| v.as_str())
}
#[cfg(feature = "ir")]
fn node_bind_field(n: &J) -> Option<String> {
if n.get("map").is_some() || n.get("cond").is_some() {
return None;
}
n.get("bindField")
.and_then(|v| v.as_str())
.map(str::to_string)
}
#[cfg(feature = "ir")]
fn node_relation_kind(n: &J) -> Option<RelationKind> {
let sub = if let Some(m) = n.get("map") { m } else { n };
if n.get("cond").is_some() {
return None;
}
match sub.get("relationKind").and_then(|v| v.as_str()) {
Some("connection") => Some(RelationKind::Connection),
Some(_) => Some(RelationKind::Single),
None => None,
}
}
#[cfg(feature = "ir")]
fn node_relation_kind_str(n: &J) -> Option<&str> {
let sub = if let Some(m) = n.get("map") { m } else { n };
if n.get("cond").is_some() {
return None;
}
sub.get("relationKind").and_then(|v| v.as_str())
}
#[cfg(feature = "ir")]
fn node_policy(n: &J) -> Option<String> {
if n.get("cond").is_some() {
return None;
}
let sub = if let Some(m) = n.get("map") { m } else { n };
sub.get("policy")
.and_then(|v| v.as_str())
.map(str::to_string)
}
#[cfg(feature = "ir")]
fn eval_ports(ports: &J, scope: &[(String, Value)]) -> Result<Vec<(String, Value)>, BehaviorError> {
let obj = ports.as_object().ok_or_else(|| BehaviorError {
code: BehaviorFailureCode::UnknownNodeKind.as_str().to_string(),
message: "ports must be an object".into(),
})?;
let mut out = Vec::with_capacity(obj.len());
for (k, v) in obj {
out.push((k.clone(), evaluate_expression(v, scope)?));
}
Ok(out)
}
#[cfg(feature = "ir")]
fn parse_plan(plan: Option<&J>) -> Option<ExecutionPlanSpec> {
let p = plan?;
if p.is_null() {
return None;
}
let groups = p
.get("groups")?
.as_array()?
.iter()
.map(|g| {
g.as_array()
.map(|row| {
row.iter()
.filter_map(|i| i.as_u64().map(|x| x as usize))
.collect()
})
.unwrap_or_default()
})
.collect();
let concurrency = p.get("concurrency").and_then(|c| c.as_i64()).unwrap_or(1);
Some(ExecutionPlanSpec {
groups,
concurrency,
})
}
#[cfg(feature = "ir")]
pub fn run_behavior(
ir: &J,
handlers: &mut dyn ComponentExec,
input: &[(String, Value)],
entry: Option<&str>,
) -> Result<Value, BehaviorError> {
let components = ir
.get("components")
.and_then(|c| c.as_array())
.ok_or_else(|| BehaviorError {
code: BehaviorFailureCode::UnknownEntry.as_str().to_string(),
message: "IR.components must be an array".into(),
})?;
let comp = match entry {
Some(name) => components
.iter()
.find(|c| c.get("name").and_then(|n| n.as_str()) == Some(name)),
None => components.first(),
};
let comp = match comp {
Some(c) => c,
None => {
return bfail(
BehaviorFailureCode::UnknownEntry,
format!("component '{}' not found in IR", entry.unwrap_or("<first>")),
)
}
};
let body: Vec<J> = comp
.get("body")
.and_then(|b| b.as_array())
.cloned()
.unwrap_or_default();
let index_of = |id: &str| {
body.iter()
.position(|n| n.get("id").and_then(|v| v.as_str()) == Some(id))
};
let mut ops: Vec<OpSpec> = Vec::with_capacity(body.len());
for n in &body {
let parent = node_parent(n).and_then(index_of);
ops.push(OpSpec {
id: n
.get("id")
.and_then(|v| v.as_str())
.unwrap_or_default()
.to_string(),
parent,
bind_field: node_bind_field(n),
relation_kind: node_relation_kind(n),
policy: node_policy(n),
});
}
run_behavior_inner(
&body,
&ops,
parse_plan(comp.get("plan")),
comp.get("output").cloned().unwrap_or(J::Null),
input,
handlers,
)
}
#[cfg(feature = "ir")]
fn run_behavior_inner(
body: &[J],
ops: &[OpSpec],
plan: Option<ExecutionPlanSpec>,
output: J,
input: &[(String, Value)],
handlers: &mut dyn ComponentExec,
) -> Result<Value, BehaviorError> {
let index_of = |id: &str| {
body.iter()
.position(|n| n.get("id").and_then(|v| v.as_str()) == Some(id))
};
let mut results: Vec<(String, Value)> = Vec::new();
let mut pending_err: Option<BehaviorError> = None;
let run = {
let results_cell = &mut results;
let err_cell = &mut pending_err;
let exec = |op: &OpSpec, _bound: Option<&Value>| -> ExecOutcome {
if err_cell.is_some() {
return ExecOutcome::Error("aborted".into());
}
let idx = match index_of(&op.id) {
Some(i) => i,
None => {
*err_cell = Some(BehaviorError {
code: BehaviorFailureCode::UnknownNodeKind.as_str().to_string(),
message: format!("no body node for op '{}'", op.id),
});
return ExecOutcome::Error("aborted".into());
}
};
let node = &body[idx];
let base_scope = |extra: Option<(&str, &Value)>| -> Vec<(String, Value)> {
let mut s: Vec<(String, Value)> = input.to_vec();
for (k, v) in results_cell.iter() {
s.push((k.clone(), v.clone()));
}
if let Some((k, v)) = extra {
s.push((k.to_string(), v.clone()));
}
s
};
let outcome = match node_kind(node) {
Err(e) => {
*err_cell = Some(e);
return ExecOutcome::Error("aborted".into());
}
Ok("cond") => {
let c = node.get("cond").unwrap();
let cond_expr = serde_json::json!({
"cond": [c.get("if"), c.get("then"), c.get("else")]
});
match evaluate_expression(&cond_expr, &base_scope(None)) {
Ok(v) => ExecOutcome::Ok(v),
Err(e) => {
*err_cell = Some(e.into());
return ExecOutcome::Error("aborted".into());
}
}
}
Ok("map") => {
let m = node.get("map").unwrap();
let over = match evaluate_expression(
m.get("over").unwrap_or(&J::Null),
&base_scope(None),
) {
Ok(v) => v,
Err(e) => {
*err_cell = Some(e.into());
return ExecOutcome::Error("aborted".into());
}
};
let arr = match &over {
Value::Arr(a) => a.clone(),
_ => {
*err_cell = Some(BehaviorError {
code: BehaviorFailureCode::MapOverNotArray.as_str().to_string(),
message: format!("map '{}': 'over' is not an array", op.id),
});
return ExecOutcome::Error("aborted".into());
}
};
let component = m.get("component").and_then(|v| v.as_str()).unwrap_or("");
let as_name = m.get("as").and_then(|v| v.as_str()).unwrap_or("$");
let ports_j = m
.get("ports")
.cloned()
.unwrap_or(J::Object(Default::default()));
let when = m.get("when");
let batched = m.get("batched").and_then(|v| v.as_bool()).unwrap_or(false);
let into = m.get("into").and_then(|v| v.as_str());
let keep = |scope: &[(String, Value)]| -> Result<bool, BehaviorError> {
match when {
None => Ok(true),
Some(w) => {
let cond_expr = serde_json::json!({ "cond": [w, true, false] });
Ok(matches!(
evaluate_expression(&cond_expr, scope)?,
Value::Bool(true)
))
}
}
};
let mut kept_idx: Vec<usize> = Vec::new(); let collected: Vec<Value>;
if batched {
let mut items: Vec<Value> = Vec::new();
for (i, el) in arr.iter().enumerate() {
let scope = base_scope(Some((as_name, el)));
match keep(&scope) {
Ok(false) => continue,
Ok(true) => {}
Err(e) => {
*err_cell = Some(e);
return ExecOutcome::Error("aborted".into());
}
}
match eval_ports(&ports_j, &scope) {
Ok(p) => items.push(Value::Obj(p)),
Err(e) => {
*err_cell = Some(e);
return ExecOutcome::Error("aborted".into());
}
}
kept_idx.push(i);
}
if items.is_empty() {
collected = Vec::new(); } else {
let want = items.len();
let batch_ports = vec![("items".to_string(), Value::Arr(items))];
match handlers.exec_ctx(&op.id, component, &batch_ports, None) {
None => {
*err_cell = Some(BehaviorError {
code: BehaviorFailureCode::UnknownComponent
.as_str()
.to_string(),
message: format!(
"component '{component}' has no handler (fail-closed)"
),
});
return ExecOutcome::Error("aborted".into());
}
Some(ExecOutcome::Error(e)) => return ExecOutcome::Error(e),
Some(ExecOutcome::Ok(Value::Arr(r))) if r.len() == want => {
collected = r;
}
Some(ExecOutcome::Ok(_)) => {
*err_cell = Some(BehaviorError {
code: BehaviorFailureCode::MapBatchResultMismatch
.as_str()
.to_string(),
message: format!(
"map '{}': batched handler must return a list aligned to items (want {want})",
op.id
),
});
return ExecOutcome::Error("aborted".into());
}
}
}
} else {
let mut out: Vec<Value> = Vec::with_capacity(arr.len());
for (i, el) in arr.iter().enumerate() {
let scope = base_scope(Some((as_name, el)));
match keep(&scope) {
Ok(false) => continue,
Ok(true) => {}
Err(e) => {
*err_cell = Some(e);
return ExecOutcome::Error("aborted".into());
}
}
let ports = match eval_ports(&ports_j, &scope) {
Ok(p) => p,
Err(e) => {
*err_cell = Some(e);
return ExecOutcome::Error("aborted".into());
}
};
match handlers.exec_ctx(&op.id, component, &ports, Some(el)) {
None => {
*err_cell = Some(BehaviorError {
code: BehaviorFailureCode::UnknownComponent
.as_str()
.to_string(),
message: format!(
"component '{component}' has no handler (fail-closed)"
),
});
return ExecOutcome::Error("aborted".into());
}
Some(ExecOutcome::Error(e)) => return ExecOutcome::Error(e),
Some(ExecOutcome::Ok(v)) => out.push(v),
}
kept_idx.push(i);
}
collected = out;
}
match into {
None => ExecOutcome::Ok(Value::Arr(collected)),
Some(key) => {
let mut augmented: Vec<Value> = Vec::with_capacity(arr.len());
let mut k = 0usize;
for (i, el) in arr.iter().enumerate() {
if k < kept_idx.len() && kept_idx[k] == i {
let fields = match el {
Value::Obj(f) => f.clone(),
_ => {
*err_cell = Some(BehaviorError {
code: BehaviorFailureCode::MapIntoElementNotObject
.as_str()
.to_string(),
message: format!(
"map '{}': 'into' requires object elements (element {i} is not an object)",
op.id
),
});
return ExecOutcome::Error("aborted".into());
}
};
let mut fields = fields;
let val = collected[k].clone();
match fields.iter_mut().find(|(fk, _)| fk == key) {
Some(slot) => slot.1 = val,
None => fields.push((key.to_string(), val)),
}
augmented.push(Value::Obj(fields));
k += 1;
} else {
augmented.push(el.clone());
}
}
ExecOutcome::Ok(Value::Arr(augmented))
}
}
}
Ok(_) => {
let component = node.get("component").and_then(|v| v.as_str()).unwrap_or("");
let ports_j = node
.get("ports")
.cloned()
.unwrap_or(J::Object(Default::default()));
let ports = match eval_ports(&ports_j, &base_scope(None)) {
Ok(p) => p,
Err(e) => {
*err_cell = Some(e);
return ExecOutcome::Error("aborted".into());
}
};
match handlers.exec_ctx(&op.id, component, &ports, None) {
None => {
*err_cell = Some(BehaviorError {
code: BehaviorFailureCode::UnknownComponent.as_str().to_string(),
message: format!(
"component '{component}' has no handler (fail-closed)"
),
});
return ExecOutcome::Error("aborted".into());
}
Some(o) => o,
}
}
};
if let ExecOutcome::Ok(v) = &outcome {
results_cell.push((op.id.clone(), v.clone()));
}
outcome
};
run_plan(plan.as_ref(), ops, exec)
};
if let Some(e) = pending_err {
return Err(e);
}
let run = run?;
for (i, id_val) in run.skipped.iter().enumerate() {
let _ = i;
if let Some(idx) = index_of(id_val) {
let rk = node_relation_kind_str(&body[idx]);
let unproduced = if rk == Some("connection") {
Value::Obj(vec![
("items".into(), Value::Arr(vec![])),
("cursor".into(), Value::Null),
])
} else {
Value::Null
};
if scope_get(&results, id_val).is_none() {
results.push((id_val.clone(), unproduced));
}
}
}
let scope: Vec<(String, Value)> = {
let mut s: Vec<(String, Value)> = input.to_vec();
s.extend(results.iter().cloned());
s
};
Ok(evaluate_expression(&output, &scope)?)
}