backbone_bucket/application/workflows/
media_processing_workflow.rs

1//! MediaProcessing flow implementation
2//!
3//! Generated by metaphor-schema
4//!
5//! Asynchronous media processing workflow triggered after a file is uploaded.
6//! Generates thumbnails and previews for images, videos, and documents.
7//! Reusable by any upload flow that emits FileUploadedEvent.
8
9use serde::{Deserialize, Serialize};
10use std::sync::Arc;
11use chrono;
12use backbone_core::flow::{WorkflowStep, WorkflowContext};
13
14/// Error type for flow execution
15#[derive(Debug, Clone, thiserror::Error)]
16pub enum FlowError {
17    #[error("No current step to execute")]
18    NoCurrentStep,
19
20    #[error("Step execution failed: {0}")]
21    StepFailed(String),
22
23    #[error("Condition evaluation failed: {0}")]
24    ConditionFailed(String),
25
26    #[error("Compensation failed: {0}")]
27    CompensationFailed(String),
28
29    #[error("Flow timed out")]
30    Timeout,
31
32    #[error("Flow cancelled")]
33    Cancelled,
34
35    #[error("Invalid state transition: {from} -> {to}")]
36    InvalidTransition { from: String, to: String },
37}
38
39/// Execution status for MediaProcessing flow
40#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
41#[serde(rename_all = "snake_case")]
42pub enum MediaProcessingFlowStatus {
43    /// Flow is pending execution
44    Pending,
45    /// Flow is currently running
46    Running,
47    /// Flow is waiting for an event or condition
48    Waiting,
49    /// Flow completed successfully
50    Completed,
51    /// Flow failed
52    Failed,
53    /// Flow was cancelled
54    Cancelled,
55    /// Flow is compensating (rolling back)
56    Compensating,
57}
58
59/// Steps in MediaProcessing flow
60#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
61#[serde(rename_all = "snake_case")]
62pub enum MediaProcessingFlowStep {
63    /// check_media_type
64    CheckMediaType,
65    /// create_image_thumbnail
66    CreateImageThumbnail,
67    /// create_video_thumbnail
68    CreateVideoThumbnail,
69    /// create_document_preview
70    CreateDocumentPreview,
71    /// no_processing_needed
72    NoProcessingNeeded,
73    /// complete
74    Complete,
75}
76
77/// Instance of MediaProcessing flow execution
78#[derive(Debug, Clone, Serialize, Deserialize)]
79pub struct MediaProcessingFlowInstance {
80    /// Unique instance ID
81    pub id: String,
82    /// Current status
83    pub status: MediaProcessingFlowStatus,
84    /// Current step
85    pub current_step: Option<MediaProcessingFlowStep>,
86    /// Flow context (variables)
87    pub context: serde_json::Value,
88    /// Completed steps
89    pub completed_steps: Vec<MediaProcessingFlowStep>,
90    /// Error if failed
91    pub error: Option<String>,
92    /// Created timestamp
93    pub created_at: chrono::DateTime<chrono::Utc>,
94    /// Updated timestamp
95    pub updated_at: chrono::DateTime<chrono::Utc>,
96}
97
98impl MediaProcessingFlowInstance {
99    /// Create a new flow instance
100    pub fn new(id: impl Into<String>) -> Self {
101        let now = chrono::Utc::now();
102        Self {
103            id: id.into(),
104            status: MediaProcessingFlowStatus::Pending,
105            current_step: None,
106            context: serde_json::json!({}),
107            completed_steps: Vec::new(),
108            error: None,
109            created_at: now,
110            updated_at: now,
111        }
112    }
113
114    /// Check if flow is complete
115    pub fn is_complete(&self) -> bool {
116        matches!(
117            self.status,
118            MediaProcessingFlowStatus::Completed | MediaProcessingFlowStatus::Failed | MediaProcessingFlowStatus::Cancelled
119        )
120    }
121
122    /// Check if flow is running
123    pub fn is_running(&self) -> bool {
124        matches!(
125            self.status,
126            MediaProcessingFlowStatus::Running | MediaProcessingFlowStatus::Waiting
127        )
128    }
129
130    /// Set a context variable
131    pub fn set_context(&mut self, key: &str, value: serde_json::Value) {
132        if let serde_json::Value::Object(ref mut map) = self.context {
133            map.insert(key.to_string(), value);
134        }
135        self.updated_at = chrono::Utc::now();
136    }
137
138    /// Get a context variable
139    pub fn get_context(&self, key: &str) -> Option<&serde_json::Value> {
140        if let serde_json::Value::Object(ref map) = self.context {
141            map.get(key)
142        } else {
143            None
144        }
145    }
146}
147
148// Phase 1: FlowInstance satisfies backbone_core::flow::WorkflowContext.
149// The entity() accessor requires the entity to be carried in the instance;
150// add `pub entity: MediaProcessing` in the // <<< CUSTOM section and remove the todo!.
151#[allow(unused_variables)]
152impl WorkflowContext<MediaProcessingFlowInstance> for MediaProcessingFlowInstance {
153    fn entity(&self) -> &MediaProcessingFlowInstance { self }
154    fn set_var(&mut self, key: &str, value: serde_json::Value) {
155        self.set_context(key, value);
156    }
157    fn get_var(&self, key: &str) -> Option<&serde_json::Value> {
158        self.get_context(key)
159    }
160}
161
162/// Step handler trait for MediaProcessing flow
163#[async_trait::async_trait]
164pub trait MediaProcessingStepHandler: Send + Sync {
165    /// Evaluate condition for check_media_type step
166    async fn evaluate_check_media_type(
167        &self,
168        instance: &MediaProcessingFlowInstance,
169    ) -> Result<MediaProcessingFlowStep, FlowError>;
170
171    /// Handle create_image_thumbnail step
172    async fn handle_create_image_thumbnail(
173        &self,
174        instance: &mut MediaProcessingFlowInstance,
175    ) -> Result<Option<MediaProcessingFlowStep>, FlowError>;
176
177    /// Handle create_video_thumbnail step
178    async fn handle_create_video_thumbnail(
179        &self,
180        instance: &mut MediaProcessingFlowInstance,
181    ) -> Result<Option<MediaProcessingFlowStep>, FlowError>;
182
183    /// Handle create_document_preview step
184    async fn handle_create_document_preview(
185        &self,
186        instance: &mut MediaProcessingFlowInstance,
187    ) -> Result<Option<MediaProcessingFlowStep>, FlowError>;
188
189    /// Handle terminal step no_processing_needed
190    async fn handle_no_processing_needed(
191        &self,
192        instance: &mut MediaProcessingFlowInstance,
193    ) -> Result<(), FlowError>;
194
195    /// Handle terminal step complete
196    async fn handle_complete(
197        &self,
198        instance: &mut MediaProcessingFlowInstance,
199    ) -> Result<(), FlowError>;
200
201}
202
203/// Executor for MediaProcessing flow
204pub struct MediaProcessingFlowExecutor<H: MediaProcessingStepHandler> {
205    handler: Arc<H>,
206}
207
208impl<H: MediaProcessingStepHandler> MediaProcessingFlowExecutor<H> {
209    /// Create a new flow executor
210    pub fn new(handler: Arc<H>) -> Self {
211        Self { handler }
212    }
213
214    /// Start a new flow instance
215    pub async fn start(&self, instance_id: impl Into<String>) -> Result<MediaProcessingFlowInstance, FlowError> {
216        let mut instance = MediaProcessingFlowInstance::new(instance_id);
217        instance.status = MediaProcessingFlowStatus::Running;
218        instance.current_step = Some(MediaProcessingFlowStep::CheckMediaType);
219        Ok(instance)
220    }
221
222    /// Execute the current step
223    pub async fn execute_step(&self, instance: &mut MediaProcessingFlowInstance) -> Result<(), FlowError> {
224        let current_step = match instance.current_step {
225            Some(step) => step,
226            None => return Err(FlowError::NoCurrentStep),
227        };
228
229        let next_step = match current_step {
230            MediaProcessingFlowStep::CheckMediaType => {
231                let next = self.handler.evaluate_check_media_type(instance).await?;
232                Some(next)
233            }
234            MediaProcessingFlowStep::CreateImageThumbnail => {
235                self.handler.handle_create_image_thumbnail(instance).await?
236            }
237            MediaProcessingFlowStep::CreateVideoThumbnail => {
238                self.handler.handle_create_video_thumbnail(instance).await?
239            }
240            MediaProcessingFlowStep::CreateDocumentPreview => {
241                self.handler.handle_create_document_preview(instance).await?
242            }
243            MediaProcessingFlowStep::NoProcessingNeeded => {
244                self.handler.handle_no_processing_needed(instance).await?;
245                None // Terminal step
246            }
247            MediaProcessingFlowStep::Complete => {
248                self.handler.handle_complete(instance).await?;
249                None // Terminal step
250            }
251        };
252
253        // Mark current step as completed
254        instance.completed_steps.push(current_step);
255
256        // Move to next step
257        match next_step {
258            Some(next) => {
259                instance.current_step = Some(next);
260            }
261            None => {
262                instance.current_step = None;
263                instance.status = MediaProcessingFlowStatus::Completed;
264            }
265        }
266
267        instance.updated_at = chrono::Utc::now();
268        Ok(())
269    }
270
271    /// Run the flow to completion
272    pub async fn run(&self, instance: &mut MediaProcessingFlowInstance) -> Result<(), FlowError> {
273        while !instance.is_complete() && instance.status != MediaProcessingFlowStatus::Waiting {
274            self.execute_step(instance).await?;
275        }
276        Ok(())
277    }
278
279    /// Cancel the flow
280    pub fn cancel(&self, instance: &mut MediaProcessingFlowInstance) {
281        instance.status = MediaProcessingFlowStatus::Cancelled;
282        instance.updated_at = chrono::Utc::now();
283    }
284
285    /// Mark flow as failed
286    pub fn fail(&self, instance: &mut MediaProcessingFlowInstance, error: impl Into<String>) {
287        instance.status = MediaProcessingFlowStatus::Failed;
288        instance.error = Some(error.into());
289        instance.updated_at = chrono::Utc::now();
290    }
291}
292
293#[cfg(test)]
294mod tests {
295    use super::*;
296
297    #[test]
298    fn test_flow_instance_creation() {
299        let instance = MediaProcessingFlowInstance::new("test-1");
300        assert_eq!(instance.id, "test-1");
301        assert_eq!(instance.status, MediaProcessingFlowStatus::Pending);
302        assert!(!instance.is_complete());
303    }
304
305    #[test]
306    fn test_flow_context() {
307        let mut instance = MediaProcessingFlowInstance::new("test-2");
308        instance.set_context("key", serde_json::json!("value"));
309        let value = instance.get_context("key");
310        assert_eq!(value, Some(&serde_json::json!("value")));
311    }
312
313    #[test]
314    fn test_flow_status_transitions() {
315        let mut instance = MediaProcessingFlowInstance::new("test-3");
316        assert!(!instance.is_running());
317
318        instance.status = MediaProcessingFlowStatus::Running;
319        assert!(instance.is_running());
320        assert!(!instance.is_complete());
321
322        instance.status = MediaProcessingFlowStatus::Completed;
323        assert!(instance.is_complete());
324        assert!(!instance.is_running());
325    }
326}
backbone_bucket/application/workflows/media_processing_workflow.rs

backbone_bucket/application/workflows/
media_processing_workflow.rs
