a3s_flow/
context.rs

1use chrono::{DateTime, Utc};
2use serde::de::DeserializeOwned;
3
4use crate::error::{FlowError, Result};
5use crate::model::{
6    FlowEvent, FlowEventEnvelope, HookMetadata, JsonValue, RetryPolicy, RuntimeCommand, StepCommand,
7};
8use crate::runtime::WorkflowInvocation;
9
10/// Replay helper for Rust workflow runtimes.
11///
12/// `WorkflowContext` is a read-only view over a workflow invocation. It provides
13/// deterministic helpers for inspecting persisted history and returning the
14/// next command to the engine.
15pub struct WorkflowContext<'a> {
16    invocation: &'a WorkflowInvocation,
17}
18
19impl<'a> WorkflowContext<'a> {
20    pub fn new(invocation: &'a WorkflowInvocation) -> Self {
21        Self { invocation }
22    }
23
24    pub fn run_id(&self) -> &str {
25        &self.invocation.run_id
26    }
27
28    pub fn input(&self) -> &JsonValue {
29        &self.invocation.input
30    }
31
32    /// Decode the workflow input into a host-defined serde type.
33    pub fn input_as<T>(&self) -> Result<T>
34    where
35        T: DeserializeOwned,
36    {
37        self.invocation.input_as()
38    }
39
40    pub fn history(&self) -> &[FlowEventEnvelope] {
41        &self.invocation.history
42    }
43
44    pub fn step_output(&self, step_id: &str) -> Option<&JsonValue> {
45        self.history()
46            .iter()
47            .find_map(|envelope| match &envelope.event {
48                FlowEvent::StepCompleted {
49                    step_id: id,
50                    output,
51                } if id == step_id => Some(output),
52                _ => None,
53            })
54    }
55
56    pub fn step_output_as<T>(&self, step_id: &str) -> Result<Option<T>>
57    where
58        T: DeserializeOwned,
59    {
60        self.step_output(step_id)
61            .cloned()
62            .map(serde_json::from_value)
63            .transpose()
64            .map_err(FlowError::from)
65    }
66
67    pub fn step_completed(&self, step_id: &str) -> bool {
68        self.step_output(step_id).is_some()
69    }
70
71    pub fn step_failed(&self, step_id: &str) -> Option<&str> {
72        self.history()
73            .iter()
74            .rev()
75            .find_map(|envelope| match &envelope.event {
76                FlowEvent::StepFailed {
77                    step_id: id, error, ..
78                } if id == step_id => Some(error.as_str()),
79                _ => None,
80            })
81    }
82
83    pub fn wait_completed(&self, wait_id: &str) -> bool {
84        self.history().iter().any(|envelope| {
85            matches!(
86                &envelope.event,
87                FlowEvent::WaitCompleted { wait_id: id } if id == wait_id
88            )
89        })
90    }
91
92    pub fn hook_payload(&self, hook_id: &str) -> Option<&JsonValue> {
93        self.history()
94            .iter()
95            .find_map(|envelope| match &envelope.event {
96                FlowEvent::HookReceived {
97                    hook_id: id,
98                    payload,
99                } if id == hook_id => Some(payload),
100                _ => None,
101            })
102    }
103
104    pub fn hook_payload_as<T>(&self, hook_id: &str) -> Result<Option<T>>
105    where
106        T: DeserializeOwned,
107    {
108        self.hook_payload(hook_id)
109            .cloned()
110            .map(serde_json::from_value)
111            .transpose()
112            .map_err(FlowError::from)
113    }
114
115    pub fn hook_disposed(&self, hook_id: &str) -> bool {
116        self.history().iter().any(|envelope| {
117            matches!(
118                &envelope.event,
119                FlowEvent::HookDisposed { hook_id: id } if id == hook_id
120            )
121        })
122    }
123
124    pub fn complete(&self, output: JsonValue) -> RuntimeCommand {
125        RuntimeCommand::Complete { output }
126    }
127
128    pub fn fail(&self, error: impl Into<String>) -> RuntimeCommand {
129        RuntimeCommand::Fail {
130            error: error.into(),
131        }
132    }
133
134    pub fn schedule_step(
135        &self,
136        step_id: impl Into<String>,
137        step_name: impl Into<String>,
138        input: JsonValue,
139    ) -> RuntimeCommand {
140        RuntimeCommand::schedule_step(step_id, step_name, input)
141    }
142
143    pub fn schedule_step_with_retry(
144        &self,
145        step_id: impl Into<String>,
146        step_name: impl Into<String>,
147        input: JsonValue,
148        retry: RetryPolicy,
149    ) -> RuntimeCommand {
150        RuntimeCommand::ScheduleStep {
151            step_id: step_id.into(),
152            step_name: step_name.into(),
153            input,
154            retry,
155        }
156    }
157
158    pub fn step(
159        &self,
160        step_id: impl Into<String>,
161        step_name: impl Into<String>,
162        input: JsonValue,
163    ) -> StepCommand {
164        StepCommand::new(step_id, step_name, input)
165    }
166
167    pub fn step_with_retry(
168        &self,
169        step_id: impl Into<String>,
170        step_name: impl Into<String>,
171        input: JsonValue,
172        retry: RetryPolicy,
173    ) -> StepCommand {
174        StepCommand::new(step_id, step_name, input).with_retry(retry)
175    }
176
177    pub fn schedule_steps(&self, steps: Vec<StepCommand>) -> RuntimeCommand {
178        RuntimeCommand::schedule_steps(steps)
179    }
180
181    pub fn wait_until(
182        &self,
183        wait_id: impl Into<String>,
184        resume_at: DateTime<Utc>,
185    ) -> RuntimeCommand {
186        RuntimeCommand::WaitUntil {
187            wait_id: wait_id.into(),
188            resume_at,
189        }
190    }
191
192    pub fn create_hook(
193        &self,
194        hook_id: impl Into<String>,
195        token: impl Into<String>,
196        metadata: JsonValue,
197    ) -> RuntimeCommand {
198        RuntimeCommand::CreateHook {
199            hook_id: hook_id.into(),
200            token: token.into(),
201            metadata,
202        }
203    }
204
205    pub fn create_hook_with_metadata(
206        &self,
207        hook_id: impl Into<String>,
208        token: impl Into<String>,
209        metadata: HookMetadata,
210    ) -> Result<RuntimeCommand> {
211        Ok(self.create_hook(hook_id, token, metadata.into_json()?))
212    }
213}
a3s_flow/context.rs

a3s_flow/
context.rs
