use std::collections::BTreeMap;
use crate::error::{FlowError, Result};
use super::{
FlowEvent, FlowEventEnvelope, HookSnapshot, HookStatus, StepSnapshot, StepStatus, WaitSnapshot,
WaitStatus, WorkflowRunSnapshot, WorkflowRunStatus,
};
pub(crate) fn project_run(
run_id: &str,
events: &[FlowEventEnvelope],
) -> Result<WorkflowRunSnapshot> {
let first = events
.first()
.ok_or_else(|| FlowError::RunNotFound(run_id.to_string()))?;
let (spec, input) = match &first.event {
FlowEvent::RunCreated { spec, input } => (spec.clone(), input.clone()),
_ => {
return Err(FlowError::InvalidTransition(
"first run event must be run_created".to_string(),
))
}
};
let mut snapshot = WorkflowRunSnapshot {
run_id: run_id.to_string(),
spec,
input,
status: WorkflowRunStatus::Pending,
steps: BTreeMap::new(),
waits: BTreeMap::new(),
hooks: BTreeMap::new(),
output: None,
error: None,
last_sequence: first.sequence,
};
for (index, envelope) in events.iter().enumerate() {
let expected_sequence = index as u64 + 1;
if envelope.sequence != expected_sequence {
return Err(FlowError::InvalidTransition(format!(
"event sequence must be contiguous for run {run_id}: expected {expected_sequence}, got {}",
envelope.sequence
)));
}
if envelope.run_id != run_id {
return Err(FlowError::InvalidTransition(format!(
"event {} belongs to run {} not {}",
envelope.event_id, envelope.run_id, run_id
)));
}
if index > 0 && snapshot.status.is_terminal() {
return Err(FlowError::InvalidTransition(format!(
"event {} appears after terminal run state",
envelope.event.event_key()
)));
}
snapshot.last_sequence = envelope.sequence;
match &envelope.event {
FlowEvent::RunCreated { .. } => {
if index > 0 {
return Err(FlowError::InvalidTransition(
"run_created must only appear as the first event".to_string(),
));
}
}
FlowEvent::RunStarted => {
if snapshot.status != WorkflowRunStatus::Pending {
return Err(FlowError::InvalidTransition(
"run_started can only follow a pending run".to_string(),
));
}
snapshot.status = WorkflowRunStatus::Running;
}
FlowEvent::RunCompleted { output } => {
snapshot.status = WorkflowRunStatus::Completed;
snapshot.output = Some(output.clone());
snapshot.error = None;
}
FlowEvent::RunFailed { error } => {
snapshot.status = WorkflowRunStatus::Failed;
snapshot.error = Some(error.clone());
}
FlowEvent::RunCancelled { reason } => {
snapshot.status = WorkflowRunStatus::Cancelled;
snapshot.error = reason.clone();
}
FlowEvent::StepCreated {
step_id,
step_name,
input,
retry,
} => {
if snapshot.steps.contains_key(step_id) {
return Err(FlowError::InvalidTransition(format!(
"step_created duplicates step {step_id}"
)));
}
snapshot.steps.insert(
step_id.clone(),
StepSnapshot {
step_id: step_id.clone(),
step_name: step_name.clone(),
status: StepStatus::Pending,
input: input.clone(),
retry: *retry,
output: None,
error: None,
attempt: 0,
retry_after: None,
},
);
}
FlowEvent::StepStarted { step_id, attempt } => {
let step = snapshot.steps.get_mut(step_id).ok_or_else(|| {
FlowError::InvalidTransition(format!(
"step_started references unknown step {step_id}"
))
})?;
if step.status != StepStatus::Pending {
return Err(FlowError::InvalidTransition(format!(
"step_started cannot follow {:?} for step {step_id}",
step.status
)));
}
step.status = StepStatus::Running;
step.attempt = *attempt;
step.retry_after = None;
}
FlowEvent::StepCompleted { step_id, output } => {
let step = snapshot.steps.get_mut(step_id).ok_or_else(|| {
FlowError::InvalidTransition(format!(
"step_completed references unknown step {step_id}"
))
})?;
if step.status != StepStatus::Running {
return Err(FlowError::InvalidTransition(format!(
"step_completed cannot follow {:?} for step {step_id}",
step.status
)));
}
step.status = StepStatus::Completed;
step.output = Some(output.clone());
step.error = None;
step.retry_after = None;
}
FlowEvent::StepRetrying {
step_id,
attempt,
error,
retry_after,
} => {
let step = snapshot.steps.get_mut(step_id).ok_or_else(|| {
FlowError::InvalidTransition(format!(
"step_retrying references unknown step {step_id}"
))
})?;
if step.status != StepStatus::Running {
return Err(FlowError::InvalidTransition(format!(
"step_retrying cannot follow {:?} for step {step_id}",
step.status
)));
}
step.status = StepStatus::Pending;
step.attempt = *attempt;
step.error = Some(error.clone());
step.retry_after = *retry_after;
}
FlowEvent::StepFailed {
step_id,
attempt,
error,
} => {
let step = snapshot.steps.get_mut(step_id).ok_or_else(|| {
FlowError::InvalidTransition(format!(
"step_failed references unknown step {step_id}"
))
})?;
if step.status != StepStatus::Running {
return Err(FlowError::InvalidTransition(format!(
"step_failed cannot follow {:?} for step {step_id}",
step.status
)));
}
step.status = StepStatus::Failed;
step.attempt = *attempt;
step.error = Some(error.clone());
step.retry_after = None;
}
FlowEvent::WaitCreated { wait_id, resume_at } => {
if snapshot.waits.contains_key(wait_id) {
return Err(FlowError::InvalidTransition(format!(
"wait_created duplicates wait {wait_id}"
)));
}
snapshot.waits.insert(
wait_id.clone(),
WaitSnapshot {
wait_id: wait_id.clone(),
status: WaitStatus::Waiting,
resume_at: *resume_at,
},
);
}
FlowEvent::WaitCompleted { wait_id } => {
let wait = snapshot.waits.get_mut(wait_id).ok_or_else(|| {
FlowError::InvalidTransition(format!(
"wait_completed references unknown wait {wait_id}"
))
})?;
if wait.status != WaitStatus::Waiting {
return Err(FlowError::InvalidTransition(format!(
"wait_completed cannot follow {:?} for wait {wait_id}",
wait.status
)));
}
wait.status = WaitStatus::Completed;
}
FlowEvent::HookCreated {
hook_id,
token,
metadata,
} => {
if snapshot.hooks.contains_key(hook_id) {
return Err(FlowError::InvalidTransition(format!(
"hook_created duplicates hook {hook_id}"
)));
}
snapshot.hooks.insert(
hook_id.clone(),
HookSnapshot {
hook_id: hook_id.clone(),
token: token.clone(),
status: HookStatus::Active,
metadata: metadata.clone(),
payload: None,
},
);
}
FlowEvent::HookReceived { hook_id, payload } => {
let hook = snapshot.hooks.get_mut(hook_id).ok_or_else(|| {
FlowError::InvalidTransition(format!(
"hook_received references unknown hook {hook_id}"
))
})?;
if hook.status != HookStatus::Active {
return Err(FlowError::InvalidTransition(format!(
"hook_received cannot follow {:?} for hook {hook_id}",
hook.status
)));
}
hook.status = HookStatus::Received;
hook.payload = Some(payload.clone());
}
FlowEvent::HookDisposed { hook_id } => {
let hook = snapshot.hooks.get_mut(hook_id).ok_or_else(|| {
FlowError::InvalidTransition(format!(
"hook_disposed references unknown hook {hook_id}"
))
})?;
if hook.status != HookStatus::Active {
return Err(FlowError::InvalidTransition(format!(
"hook_disposed cannot follow {:?} for hook {hook_id}",
hook.status
)));
}
hook.status = HookStatus::Disposed;
}
}
}
if snapshot.status == WorkflowRunStatus::Running && snapshot.has_open_suspension() {
snapshot.status = WorkflowRunStatus::Suspended;
}
Ok(snapshot)
}