use futures_util::Stream;
use futures_util::stream::Empty;
use serde_json::{Map, Value};
use crate::workflow::definition::{JsonBoundary, MAX_COLLECTION_ITEMS, MAX_TEXT_BYTES, SchemaText};
use crate::workflow::execution::MAX_RUNTIME_BYTES;
use crate::workflow::{RunId, StateChange, WorkflowId, WorkflowState};
use crate::{AppError, WorkspaceId};
pub trait ActionBackend: Send + Sync + 'static {
type Stream: Stream<Item = Result<ActionEvent, AppError>> + 'static;
fn declaration(&self) -> &ActionDeclaration;
fn start(&self, request: ActionRequest) -> Self::Stream;
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ActionMode {
Synchronous,
Asynchronous,
Both,
}
pub struct ActionDeclaration {
pub key: String,
pub revision: String,
pub reads: Vec<String>,
pub writes: Vec<String>,
pub required_inputs: Vec<String>,
pub optional_inputs: Vec<String>,
pub bound_inputs: Map<String, Value>,
pub default_inputs: Map<String, Value>,
pub tags: Vec<String>,
pub source: Option<String>,
pub mode: ActionMode,
}
pub struct ActionRequest {
pub state: WorkflowState,
pub inputs: Map<String, Value>,
pub context: ActionContext,
}
#[derive(Clone)]
pub struct ActionContext {
pub workspace: WorkspaceId,
pub workflow: WorkflowId,
pub run: RunId,
pub occurrence: String,
pub checkpoint: u64,
pub action_ordinal: u64,
pub node: String,
pub attempt: u64,
pub mode: ActionMode,
pub parent: Option<RunId>,
pub parent_occurrence: Option<String>,
}
pub enum ActionEvent {
Partial {
result: Map<String, Value>,
},
Completed {
result: Map<String, Value>,
writes: Vec<StateChange>,
},
SpanStarted {
id: String,
name: String,
parent: Option<String>,
dependencies: Vec<String>,
},
SpanAttribute {
span: Option<String>,
key: String,
value: Value,
tags: Map<String, Value>,
},
SpanFinished {
id: String,
},
}
impl ActionEvent {
pub(crate) fn validate_span(&self) -> Result<(), AppError> {
let bytes = match self {
Self::SpanStarted {
id,
name,
parent,
dependencies,
} => {
if dependencies.len() > MAX_COLLECTION_ITEMS
|| [id.as_str(), name.as_str()]
.into_iter()
.chain(parent.as_deref())
.chain(dependencies.iter().map(String::as_str))
.any(|text| text.len() > MAX_TEXT_BYTES)
{
return Err(AppError::InvalidInput(
"caller span metadata exceeds its boundary".into(),
));
}
dependencies.iter().fold(
id.len()
.saturating_add(name.len())
.saturating_add(parent.as_ref().map_or(0, String::len)),
|bytes, dependency| bytes.saturating_add(dependency.len()),
)
}
Self::SpanAttribute {
span,
key,
value,
tags,
} => {
if key.len() > MAX_TEXT_BYTES
|| span
.as_ref()
.is_some_and(|span| span.len() > MAX_TEXT_BYTES)
|| tags.len() > MAX_COLLECTION_ITEMS
|| tags.values().any(|value| {
value
.as_str()
.is_none_or(|value| value.len() > MAX_TEXT_BYTES)
})
{
return Err(AppError::InvalidInput(
"caller span attribute exceeds its boundary".into(),
));
}
JsonBoundary::validate(value)?;
JsonBoundary::validate_map(tags)?;
let value_bytes = serde_json::to_vec(value)
.map_err(|_| AppError::InvalidInput("caller span attribute is invalid".into()))?
.len();
let tag_bytes = serde_json::to_vec(tags)
.map_err(|_| AppError::InvalidInput("caller span tags are invalid".into()))?
.len();
key.len()
.saturating_add(span.as_ref().map_or(0, String::len))
.saturating_add(value_bytes)
.saturating_add(tag_bytes)
}
Self::SpanFinished { id } => {
if id.len() > MAX_TEXT_BYTES {
return Err(AppError::InvalidInput(
"caller span identity exceeds its boundary".into(),
));
}
id.len()
}
Self::Partial { .. } | Self::Completed { .. } => {
return Err(AppError::Integration(
"action event is not span metadata".into(),
));
}
};
if bytes > MAX_RUNTIME_BYTES {
return Err(AppError::InvalidInput(
"caller span event exceeds its byte boundary".into(),
));
}
Ok(())
}
}
pub trait ConditionBackend: Send + Sync + 'static {
fn declaration(&self) -> &ConditionDeclaration;
fn evaluate(&self, state: &WorkflowState) -> Result<bool, AppError>;
}
pub struct ConditionDeclaration {
pub key: String,
pub revision: String,
pub reads: Vec<String>,
}
pub enum NoActions {}
pub enum NoConditions {}
impl ActionBackend for NoActions {
type Stream = Empty<Result<ActionEvent, AppError>>;
fn declaration(&self) -> &ActionDeclaration {
match *self {}
}
fn start(&self, _request: ActionRequest) -> Self::Stream {
match *self {}
}
}
impl ConditionBackend for NoConditions {
fn declaration(&self) -> &ConditionDeclaration {
match *self {}
}
fn evaluate(&self, _state: &WorkflowState) -> Result<bool, AppError> {
match *self {}
}
}
pub(crate) enum ActionInputs {
Ready(Map<String, Value>),
Required(Vec<String>),
}
impl ActionInputs {
pub(crate) fn into_values(self) -> Result<Map<String, Value>, AppError> {
match self {
Self::Ready(values) => Ok(values),
Self::Required(_) => Err(AppError::InvalidInput(
"required action input is missing".into(),
)),
}
}
}
impl ActionDeclaration {
pub(crate) fn result_projection(&self, result: &Map<String, Value>) -> StateChange {
let mut values = Map::new();
for key in &self.writes {
if let Some(value) = result.get(key) {
values.insert(key.clone(), value.clone());
}
}
StateChange::Assign { values }
}
pub(crate) fn resolve_inputs(
&self,
runtime: &Map<String, Value>,
mode: ActionMode,
) -> Result<ActionInputs, AppError> {
if !matches!(
(mode, self.mode),
(
ActionMode::Synchronous,
ActionMode::Synchronous | ActionMode::Both
) | (
ActionMode::Asynchronous,
ActionMode::Asynchronous | ActionMode::Both
)
) {
return Err(AppError::InvalidInput(
"action does not support the selected execution mode".into(),
));
}
JsonBoundary::validate_inputs(runtime)?;
let mut resolved = self.default_inputs.clone();
for (key, value) in &self.bound_inputs {
resolved.insert(key.clone(), value.clone());
}
for (key, value) in runtime {
if self.bound_inputs.contains_key(key) {
return Err(AppError::InvalidInput(
"runtime input cannot override a bound action input".into(),
));
}
resolved.insert(key.clone(), value.clone());
}
JsonBoundary::validate_map(&resolved)?;
let mut required = self
.required_inputs
.iter()
.filter(|key| !resolved.contains_key(*key))
.cloned()
.collect::<Vec<_>>();
required.sort_unstable();
required.dedup();
if required.is_empty() {
Ok(ActionInputs::Ready(resolved))
} else {
Ok(ActionInputs::Required(required))
}
}
pub(crate) fn validate_reads(&self, state: &WorkflowState) -> Result<(), AppError> {
if self
.reads
.iter()
.any(|key| !state.values().contains_key(key))
{
return Err(AppError::InvalidInput(
"declared action state read is missing".into(),
));
}
Ok(())
}
pub(crate) fn validate(&self) -> Result<(), AppError> {
SchemaText::node_or_state("action binding key", &self.key)?;
SchemaText::revision(&self.revision)?;
if [
self.reads.len(),
self.writes.len(),
self.required_inputs.len(),
self.optional_inputs.len(),
self.bound_inputs.len(),
self.default_inputs.len(),
self.tags.len(),
]
.into_iter()
.any(|length| length > MAX_COLLECTION_ITEMS)
{
return Err(AppError::InvalidInput(
"action declaration exceeds the private collection bound".into(),
));
}
for key in self.reads.iter().chain(&self.writes) {
JsonBoundary::validate_key(key)?;
}
for key in self
.required_inputs
.iter()
.chain(&self.optional_inputs)
.chain(self.bound_inputs.keys())
.chain(self.default_inputs.keys())
{
JsonBoundary::validate_key(key)?;
if key.starts_with("__") {
return Err(AppError::InvalidInput(
"action input uses the reserved runtime namespace".into(),
));
}
}
if self
.default_inputs
.keys()
.any(|key| self.required_inputs.contains(key) || !self.optional_inputs.contains(key))
{
return Err(AppError::InvalidInput(
"action defaults require optional, non-required inputs".into(),
));
}
if self.tags.iter().any(|tag| tag.len() > MAX_TEXT_BYTES)
|| self
.source
.as_ref()
.is_some_and(|source| source.len() > MAX_TEXT_BYTES)
{
return Err(AppError::InvalidInput(
"action declaration text exceeds the private byte bound".into(),
));
}
JsonBoundary::validate_map(&self.bound_inputs)?;
JsonBoundary::validate_map(&self.default_inputs)?;
JsonBoundary::validate_size(&(
&self.key,
&self.revision,
&self.reads,
&self.writes,
&self.required_inputs,
&self.optional_inputs,
&self.bound_inputs,
&self.default_inputs,
&self.tags,
&self.source,
))
}
}
impl ConditionDeclaration {
pub(crate) fn validate(&self) -> Result<(), AppError> {
SchemaText::node_or_state("condition binding key", &self.key)?;
SchemaText::revision(&self.revision)?;
if self.reads.len() > MAX_COLLECTION_ITEMS {
return Err(AppError::InvalidInput(
"condition declaration exceeds the private collection bound".into(),
));
}
for key in &self.reads {
JsonBoundary::validate_key(key)?;
}
JsonBoundary::validate_size(&(&self.key, &self.revision, &self.reads))
}
}