a3s_flow/model/
snapshot.rs

1use chrono::{DateTime, Utc};
2use serde::de::DeserializeOwned;
3use serde::{Deserialize, Serialize};
4use std::collections::BTreeMap;
5
6use crate::error::{FlowError, Result};
7
8use super::{JsonValue, RetryPolicy, WorkflowSpec};
9
10#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
11#[serde(rename_all = "snake_case")]
12pub enum WorkflowRunStatus {
13    Pending,
14    Running,
15    Suspended,
16    Completed,
17    Failed,
18    Cancelled,
19}
20
21impl WorkflowRunStatus {
22    pub fn is_terminal(self) -> bool {
23        matches!(self, Self::Completed | Self::Failed | Self::Cancelled)
24    }
25}
26
27#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
28#[serde(rename_all = "snake_case")]
29pub enum StepStatus {
30    Pending,
31    Running,
32    Completed,
33    Failed,
34}
35
36#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
37pub struct StepSnapshot {
38    pub step_id: String,
39    pub step_name: String,
40    pub status: StepStatus,
41    pub input: JsonValue,
42    pub retry: RetryPolicy,
43    pub output: Option<JsonValue>,
44    pub error: Option<String>,
45    pub attempt: u32,
46    pub retry_after: Option<DateTime<Utc>>,
47}
48
49impl StepSnapshot {
50    /// Decode the persisted step output into a host-defined serde type.
51    pub fn output_as<T>(&self) -> Result<Option<T>>
52    where
53        T: DeserializeOwned,
54    {
55        self.output
56            .clone()
57            .map(serde_json::from_value)
58            .transpose()
59            .map_err(FlowError::from)
60    }
61}
62
63#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
64#[serde(rename_all = "snake_case")]
65pub enum WaitStatus {
66    Waiting,
67    Completed,
68}
69
70#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
71pub struct WaitSnapshot {
72    pub wait_id: String,
73    pub status: WaitStatus,
74    pub resume_at: DateTime<Utc>,
75}
76
77#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
78#[serde(rename_all = "snake_case")]
79pub enum HookStatus {
80    Active,
81    Received,
82    Disposed,
83}
84
85#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
86pub struct HookSnapshot {
87    pub hook_id: String,
88    pub token: String,
89    pub status: HookStatus,
90    pub metadata: JsonValue,
91    pub payload: Option<JsonValue>,
92}
93
94impl HookSnapshot {
95    /// Decode the persisted hook metadata into a host-defined serde type.
96    pub fn metadata_as<T>(&self) -> Result<T>
97    where
98        T: DeserializeOwned,
99    {
100        serde_json::from_value(self.metadata.clone()).map_err(FlowError::from)
101    }
102
103    /// Decode the received hook payload into a host-defined serde type.
104    pub fn payload_as<T>(&self) -> Result<Option<T>>
105    where
106        T: DeserializeOwned,
107    {
108        self.payload
109            .clone()
110            .map(serde_json::from_value)
111            .transpose()
112            .map_err(FlowError::from)
113    }
114}
115
116/// Active external callback hook with the run that owns it.
117#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
118pub struct ActiveHookSnapshot {
119    pub run_id: String,
120    pub hook: HookSnapshot,
121}
122
123impl ActiveHookSnapshot {
124    /// Decode the active hook metadata into a host-defined serde type.
125    pub fn metadata_as<T>(&self) -> Result<T>
126    where
127        T: DeserializeOwned,
128    {
129        self.hook.metadata_as()
130    }
131}
132
133/// Aggregated run counts for host dashboards and health probes.
134#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
135pub struct WorkflowRunSummary {
136    pub total_runs: usize,
137    pub pending_runs: usize,
138    pub running_runs: usize,
139    pub suspended_runs: usize,
140    pub completed_runs: usize,
141    pub failed_runs: usize,
142    pub cancelled_runs: usize,
143    pub terminal_runs: usize,
144    pub non_terminal_runs: usize,
145    pub open_waits: usize,
146    pub active_hooks: usize,
147    pub pending_retries: usize,
148}
149
150impl WorkflowRunSummary {
151    pub fn from_snapshots(snapshots: &[WorkflowRunSnapshot]) -> Self {
152        let mut summary = Self::default();
153        for snapshot in snapshots {
154            summary.record(snapshot);
155        }
156        summary
157    }
158
159    pub fn record(&mut self, snapshot: &WorkflowRunSnapshot) {
160        self.total_runs += 1;
161        match snapshot.status {
162            WorkflowRunStatus::Pending => self.pending_runs += 1,
163            WorkflowRunStatus::Running => self.running_runs += 1,
164            WorkflowRunStatus::Suspended => self.suspended_runs += 1,
165            WorkflowRunStatus::Completed => self.completed_runs += 1,
166            WorkflowRunStatus::Failed => self.failed_runs += 1,
167            WorkflowRunStatus::Cancelled => self.cancelled_runs += 1,
168        }
169
170        if snapshot.status.is_terminal() {
171            self.terminal_runs += 1;
172            return;
173        }
174
175        self.non_terminal_runs += 1;
176        self.open_waits += snapshot
177            .waits
178            .values()
179            .filter(|wait| wait.status == WaitStatus::Waiting)
180            .count();
181        self.active_hooks += snapshot
182            .hooks
183            .values()
184            .filter(|hook| hook.status == HookStatus::Active)
185            .count();
186        self.pending_retries += snapshot
187            .steps
188            .values()
189            .filter(|step| step.status == StepStatus::Pending && step.retry_after.is_some())
190            .count();
191    }
192}
193
194/// Open suspension projected for host dashboards and operator consoles.
195#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
196#[serde(tag = "kind", rename_all = "snake_case")]
197pub enum WorkflowRunSuspension {
198    Wait {
199        run_id: String,
200        wait: WaitSnapshot,
201        due: bool,
202    },
203    Hook {
204        run_id: String,
205        hook: HookSnapshot,
206    },
207    Retry {
208        run_id: String,
209        step: StepSnapshot,
210        due: bool,
211    },
212}
213
214impl WorkflowRunSuspension {
215    pub fn run_id(&self) -> &str {
216        match self {
217            Self::Wait { run_id, .. } | Self::Hook { run_id, .. } | Self::Retry { run_id, .. } => {
218                run_id
219            }
220        }
221    }
222
223    pub fn subject_id(&self) -> &str {
224        match self {
225            Self::Wait { wait, .. } => &wait.wait_id,
226            Self::Hook { hook, .. } => &hook.hook_id,
227            Self::Retry { step, .. } => &step.step_id,
228        }
229    }
230
231    pub(crate) fn kind_order(&self) -> u8 {
232        match self {
233            Self::Wait { .. } => 0,
234            Self::Hook { .. } => 1,
235            Self::Retry { .. } => 2,
236        }
237    }
238
239    pub fn is_due(&self) -> bool {
240        match self {
241            Self::Wait { due, .. } | Self::Retry { due, .. } => *due,
242            Self::Hook { .. } => false,
243        }
244    }
245
246    /// Scheduled resume time for wait and delayed-retry suspensions.
247    pub fn scheduled_at(&self) -> Option<DateTime<Utc>> {
248        match self {
249            Self::Wait { wait, .. } => Some(wait.resume_at),
250            Self::Retry { step, .. } => step.retry_after,
251            Self::Hook { .. } => None,
252        }
253    }
254}
255
256/// Materialized state of a workflow run.
257#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
258pub struct WorkflowRunSnapshot {
259    pub run_id: String,
260    pub spec: WorkflowSpec,
261    pub input: JsonValue,
262    pub status: WorkflowRunStatus,
263    pub steps: BTreeMap<String, StepSnapshot>,
264    pub waits: BTreeMap<String, WaitSnapshot>,
265    pub hooks: BTreeMap<String, HookSnapshot>,
266    pub output: Option<JsonValue>,
267    pub error: Option<String>,
268    pub last_sequence: u64,
269}
270
271impl WorkflowRunSnapshot {
272    /// Decode the workflow input into a host-defined serde type.
273    pub fn input_as<T>(&self) -> Result<T>
274    where
275        T: DeserializeOwned,
276    {
277        serde_json::from_value(self.input.clone()).map_err(FlowError::from)
278    }
279
280    /// Decode the terminal workflow output into a host-defined serde type.
281    pub fn output_as<T>(&self) -> Result<Option<T>>
282    where
283        T: DeserializeOwned,
284    {
285        self.output
286            .clone()
287            .map(serde_json::from_value)
288            .transpose()
289            .map_err(FlowError::from)
290    }
291
292    pub fn step_output(&self, step_id: &str) -> Option<&JsonValue> {
293        self.steps
294            .get(step_id)
295            .and_then(|step| step.output.as_ref())
296    }
297
298    /// Decode a persisted step output into a host-defined serde type.
299    pub fn step_output_as<T>(&self, step_id: &str) -> Result<Option<T>>
300    where
301        T: DeserializeOwned,
302    {
303        match self.steps.get(step_id) {
304            Some(step) => step.output_as(),
305            None => Ok(None),
306        }
307    }
308
309    pub fn hook_payload(&self, hook_id: &str) -> Option<&JsonValue> {
310        self.hooks
311            .get(hook_id)
312            .and_then(|hook| hook.payload.as_ref())
313    }
314
315    /// Decode persisted hook metadata into a host-defined serde type.
316    pub fn hook_metadata_as<T>(&self, hook_id: &str) -> Result<Option<T>>
317    where
318        T: DeserializeOwned,
319    {
320        match self.hooks.get(hook_id) {
321            Some(hook) => hook.metadata_as().map(Some),
322            None => Ok(None),
323        }
324    }
325
326    /// Decode a received hook payload into a host-defined serde type.
327    pub fn hook_payload_as<T>(&self, hook_id: &str) -> Result<Option<T>>
328    where
329        T: DeserializeOwned,
330    {
331        match self.hooks.get(hook_id) {
332            Some(hook) => hook.payload_as(),
333            None => Ok(None),
334        }
335    }
336
337    pub fn has_open_suspension(&self) -> bool {
338        self.waits
339            .values()
340            .any(|wait| wait.status == WaitStatus::Waiting)
341            || self
342                .hooks
343                .values()
344                .any(|hook| hook.status == HookStatus::Active)
345            || self.steps.values().any(|step| step.retry_after.is_some())
346    }
347
348    pub fn due_retries(&self, now: DateTime<Utc>) -> Vec<(String, DateTime<Utc>)> {
349        self.steps
350            .values()
351            .filter_map(|step| match step.retry_after {
352                Some(retry_after) if step.status == StepStatus::Pending && retry_after <= now => {
353                    Some((step.step_id.clone(), retry_after))
354                }
355                _ => None,
356            })
357            .collect()
358    }
359
360    pub fn has_future_retry(&self, now: DateTime<Utc>) -> bool {
361        self.steps.values().any(|step| {
362            step.status == StepStatus::Pending
363                && step
364                    .retry_after
365                    .map(|retry_after| retry_after > now)
366                    .unwrap_or(false)
367        })
368    }
369}
a3s_flow/model/snapshot.rs

a3s_flow/model/
snapshot.rs
