backbone_bucket/application/workflows/
file_processing_workflow.rs

1//! FileProcessing flow implementation
2//!
3//! Generated by metaphor-schema
4//!
5//! Async file processing workflow including:
6//! - Image thumbnail generation
7//! - Video thumbnail generation
8//! - Document preview generation
9//! - File compression
10//! - Virus scanning
11//! - Content deduplication checking
12//! - CDN URL generation
13
14use serde::{Deserialize, Serialize};
15use std::sync::Arc;
16use chrono;
17use backbone_core::flow::{WorkflowStep, WorkflowContext};
18
19/// Error type for flow execution
20#[derive(Debug, Clone, thiserror::Error)]
21pub enum FlowError {
22    #[error("No current step to execute")]
23    NoCurrentStep,
24
25    #[error("Step execution failed: {0}")]
26    StepFailed(String),
27
28    #[error("Condition evaluation failed: {0}")]
29    ConditionFailed(String),
30
31    #[error("Compensation failed: {0}")]
32    CompensationFailed(String),
33
34    #[error("Flow timed out")]
35    Timeout,
36
37    #[error("Flow cancelled")]
38    Cancelled,
39
40    #[error("Invalid state transition: {from} -> {to}")]
41    InvalidTransition { from: String, to: String },
42}
43
44/// Execution status for FileProcessing flow
45#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
46#[serde(rename_all = "snake_case")]
47pub enum FileProcessingFlowStatus {
48    /// Flow is pending execution
49    Pending,
50    /// Flow is currently running
51    Running,
52    /// Flow is waiting for an event or condition
53    Waiting,
54    /// Flow completed successfully
55    Completed,
56    /// Flow failed
57    Failed,
58    /// Flow was cancelled
59    Cancelled,
60    /// Flow is compensating (rolling back)
61    Compensating,
62}
63
64/// Steps in FileProcessing flow
65#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
66#[serde(rename_all = "snake_case")]
67pub enum FileProcessingFlowStep {
68    /// load_job
69    LoadJob,
70    /// mark_running
71    MarkRunning,
72    /// load_file
73    LoadFile,
74    /// load_bucket
75    LoadBucket,
76    /// route_by_job_type
77    RouteByJobType,
78    /// process_image_thumbnail
79    ProcessImageThumbnail,
80    /// store_image_thumbnail
81    StoreImageThumbnail,
82    /// update_file_thumbnail
83    UpdateFileThumbnail,
84    /// process_video_thumbnail
85    ProcessVideoThumbnail,
86    /// store_video_thumbnail
87    StoreVideoThumbnail,
88    /// update_file_video_thumbnail
89    UpdateFileVideoThumbnail,
90    /// process_document_preview
91    ProcessDocumentPreview,
92    /// store_document_preview
93    StoreDocumentPreview,
94    /// update_file_document_preview
95    UpdateFileDocumentPreview,
96    /// process_compression
97    ProcessCompression,
98    /// store_compressed_file
99    StoreCompressedFile,
100    /// update_file_compression
101    UpdateFileCompression,
102    /// update_bucket_compression_stats
103    UpdateBucketCompressionStats,
104    /// process_virus_scan
105    ProcessVirusScan,
106    /// check_threat_level
107    CheckThreatLevel,
108    /// update_scan_safe
109    UpdateScanSafe,
110    /// update_scan_warning
111    UpdateScanWarning,
112    /// notify_scan_warning
113    NotifyScanWarning,
114    /// update_scan_threat
115    UpdateScanThreat,
116    /// notify_threat_detected
117    NotifyThreatDetected,
118    /// process_deduplication
119    ProcessDeduplication,
120    /// check_existing_hash
121    CheckExistingHash,
122    /// update_existing_hash
123    UpdateExistingHash,
124    /// link_file_hash
125    LinkFileHash,
126    /// create_new_hash
127    CreateNewHash,
128    /// mark_complete
129    MarkComplete,
130    /// emit_complete_event
131    EmitCompleteEvent,
132    /// complete
133    Complete,
134    /// retry_or_fail
135    RetryOrFail,
136    /// schedule_retry
137    ScheduleRetry,
138    /// retry_terminal
139    RetryTerminal,
140    /// mark_failed
141    MarkFailed,
142    /// emit_failed_event
143    EmitFailedEvent,
144    /// failed_terminal
145    FailedTerminal,
146    /// job_not_found
147    JobNotFound,
148    /// file_not_found
149    FileNotFound,
150    /// unknown_job_type
151    UnknownJobType,
152}
153
154/// Instance of FileProcessing flow execution
155#[derive(Debug, Clone, Serialize, Deserialize)]
156pub struct FileProcessingFlowInstance {
157    /// Unique instance ID
158    pub id: String,
159    /// Current status
160    pub status: FileProcessingFlowStatus,
161    /// Current step
162    pub current_step: Option<FileProcessingFlowStep>,
163    /// Flow context (variables)
164    pub context: serde_json::Value,
165    /// Completed steps
166    pub completed_steps: Vec<FileProcessingFlowStep>,
167    /// Error if failed
168    pub error: Option<String>,
169    /// Created timestamp
170    pub created_at: chrono::DateTime<chrono::Utc>,
171    /// Updated timestamp
172    pub updated_at: chrono::DateTime<chrono::Utc>,
173}
174
175impl FileProcessingFlowInstance {
176    /// Create a new flow instance
177    pub fn new(id: impl Into<String>) -> Self {
178        let now = chrono::Utc::now();
179        Self {
180            id: id.into(),
181            status: FileProcessingFlowStatus::Pending,
182            current_step: None,
183            context: serde_json::json!({}),
184            completed_steps: Vec::new(),
185            error: None,
186            created_at: now,
187            updated_at: now,
188        }
189    }
190
191    /// Check if flow is complete
192    pub fn is_complete(&self) -> bool {
193        matches!(
194            self.status,
195            FileProcessingFlowStatus::Completed | FileProcessingFlowStatus::Failed | FileProcessingFlowStatus::Cancelled
196        )
197    }
198
199    /// Check if flow is running
200    pub fn is_running(&self) -> bool {
201        matches!(
202            self.status,
203            FileProcessingFlowStatus::Running | FileProcessingFlowStatus::Waiting
204        )
205    }
206
207    /// Set a context variable
208    pub fn set_context(&mut self, key: &str, value: serde_json::Value) {
209        if let serde_json::Value::Object(ref mut map) = self.context {
210            map.insert(key.to_string(), value);
211        }
212        self.updated_at = chrono::Utc::now();
213    }
214
215    /// Get a context variable
216    pub fn get_context(&self, key: &str) -> Option<&serde_json::Value> {
217        if let serde_json::Value::Object(ref map) = self.context {
218            map.get(key)
219        } else {
220            None
221        }
222    }
223}
224
225// Phase 1: FlowInstance satisfies backbone_core::flow::WorkflowContext.
226// The entity() accessor requires the entity to be carried in the instance;
227// add `pub entity: FileProcessing` in the // <<< CUSTOM section and remove the todo!.
228#[allow(unused_variables)]
229impl WorkflowContext<FileProcessingFlowInstance> for FileProcessingFlowInstance {
230    fn entity(&self) -> &FileProcessingFlowInstance { self }
231    fn set_var(&mut self, key: &str, value: serde_json::Value) {
232        self.set_context(key, value);
233    }
234    fn get_var(&self, key: &str) -> Option<&serde_json::Value> {
235        self.get_context(key)
236    }
237}
238
239/// Step handler trait for FileProcessing flow
240#[async_trait::async_trait]
241pub trait FileProcessingStepHandler: Send + Sync {
242    /// Handle load_job step
243    async fn handle_load_job(
244        &self,
245        instance: &mut FileProcessingFlowInstance,
246    ) -> Result<Option<FileProcessingFlowStep>, FlowError>;
247
248    /// Handle mark_running step
249    async fn handle_mark_running(
250        &self,
251        instance: &mut FileProcessingFlowInstance,
252    ) -> Result<Option<FileProcessingFlowStep>, FlowError>;
253
254    /// Handle load_file step
255    async fn handle_load_file(
256        &self,
257        instance: &mut FileProcessingFlowInstance,
258    ) -> Result<Option<FileProcessingFlowStep>, FlowError>;
259
260    /// Handle load_bucket step
261    async fn handle_load_bucket(
262        &self,
263        instance: &mut FileProcessingFlowInstance,
264    ) -> Result<Option<FileProcessingFlowStep>, FlowError>;
265
266    /// Evaluate condition for route_by_job_type step
267    async fn evaluate_route_by_job_type(
268        &self,
269        instance: &FileProcessingFlowInstance,
270    ) -> Result<FileProcessingFlowStep, FlowError>;
271
272    /// Handle process_image_thumbnail step
273    async fn handle_process_image_thumbnail(
274        &self,
275        instance: &mut FileProcessingFlowInstance,
276    ) -> Result<Option<FileProcessingFlowStep>, FlowError>;
277
278    /// Handle store_image_thumbnail step
279    async fn handle_store_image_thumbnail(
280        &self,
281        instance: &mut FileProcessingFlowInstance,
282    ) -> Result<Option<FileProcessingFlowStep>, FlowError>;
283
284    /// Handle update_file_thumbnail step
285    async fn handle_update_file_thumbnail(
286        &self,
287        instance: &mut FileProcessingFlowInstance,
288    ) -> Result<Option<FileProcessingFlowStep>, FlowError>;
289
290    /// Handle process_video_thumbnail step
291    async fn handle_process_video_thumbnail(
292        &self,
293        instance: &mut FileProcessingFlowInstance,
294    ) -> Result<Option<FileProcessingFlowStep>, FlowError>;
295
296    /// Handle store_video_thumbnail step
297    async fn handle_store_video_thumbnail(
298        &self,
299        instance: &mut FileProcessingFlowInstance,
300    ) -> Result<Option<FileProcessingFlowStep>, FlowError>;
301
302    /// Handle update_file_video_thumbnail step
303    async fn handle_update_file_video_thumbnail(
304        &self,
305        instance: &mut FileProcessingFlowInstance,
306    ) -> Result<Option<FileProcessingFlowStep>, FlowError>;
307
308    /// Handle process_document_preview step
309    async fn handle_process_document_preview(
310        &self,
311        instance: &mut FileProcessingFlowInstance,
312    ) -> Result<Option<FileProcessingFlowStep>, FlowError>;
313
314    /// Handle store_document_preview step
315    async fn handle_store_document_preview(
316        &self,
317        instance: &mut FileProcessingFlowInstance,
318    ) -> Result<Option<FileProcessingFlowStep>, FlowError>;
319
320    /// Handle update_file_document_preview step
321    async fn handle_update_file_document_preview(
322        &self,
323        instance: &mut FileProcessingFlowInstance,
324    ) -> Result<Option<FileProcessingFlowStep>, FlowError>;
325
326    /// Handle process_compression step
327    async fn handle_process_compression(
328        &self,
329        instance: &mut FileProcessingFlowInstance,
330    ) -> Result<Option<FileProcessingFlowStep>, FlowError>;
331
332    /// Handle store_compressed_file step
333    async fn handle_store_compressed_file(
334        &self,
335        instance: &mut FileProcessingFlowInstance,
336    ) -> Result<Option<FileProcessingFlowStep>, FlowError>;
337
338    /// Handle update_file_compression step
339    async fn handle_update_file_compression(
340        &self,
341        instance: &mut FileProcessingFlowInstance,
342    ) -> Result<Option<FileProcessingFlowStep>, FlowError>;
343
344    /// Handle update_bucket_compression_stats step
345    async fn handle_update_bucket_compression_stats(
346        &self,
347        instance: &mut FileProcessingFlowInstance,
348    ) -> Result<Option<FileProcessingFlowStep>, FlowError>;
349
350    /// Handle process_virus_scan step
351    async fn handle_process_virus_scan(
352        &self,
353        instance: &mut FileProcessingFlowInstance,
354    ) -> Result<Option<FileProcessingFlowStep>, FlowError>;
355
356    /// Evaluate condition for check_threat_level step
357    async fn evaluate_check_threat_level(
358        &self,
359        instance: &FileProcessingFlowInstance,
360    ) -> Result<FileProcessingFlowStep, FlowError>;
361
362    /// Handle update_scan_safe step
363    async fn handle_update_scan_safe(
364        &self,
365        instance: &mut FileProcessingFlowInstance,
366    ) -> Result<Option<FileProcessingFlowStep>, FlowError>;
367
368    /// Handle update_scan_warning step
369    async fn handle_update_scan_warning(
370        &self,
371        instance: &mut FileProcessingFlowInstance,
372    ) -> Result<Option<FileProcessingFlowStep>, FlowError>;
373
374    /// Handle notify_scan_warning step
375    async fn handle_notify_scan_warning(
376        &self,
377        instance: &mut FileProcessingFlowInstance,
378    ) -> Result<Option<FileProcessingFlowStep>, FlowError>;
379
380    /// Handle update_scan_threat step
381    async fn handle_update_scan_threat(
382        &self,
383        instance: &mut FileProcessingFlowInstance,
384    ) -> Result<Option<FileProcessingFlowStep>, FlowError>;
385
386    /// Handle notify_threat_detected step
387    async fn handle_notify_threat_detected(
388        &self,
389        instance: &mut FileProcessingFlowInstance,
390    ) -> Result<Option<FileProcessingFlowStep>, FlowError>;
391
392    /// Handle process_deduplication step
393    async fn handle_process_deduplication(
394        &self,
395        instance: &mut FileProcessingFlowInstance,
396    ) -> Result<Option<FileProcessingFlowStep>, FlowError>;
397
398    /// Handle check_existing_hash step
399    async fn handle_check_existing_hash(
400        &self,
401        instance: &mut FileProcessingFlowInstance,
402    ) -> Result<Option<FileProcessingFlowStep>, FlowError>;
403
404    /// Handle update_existing_hash step
405    async fn handle_update_existing_hash(
406        &self,
407        instance: &mut FileProcessingFlowInstance,
408    ) -> Result<Option<FileProcessingFlowStep>, FlowError>;
409
410    /// Handle link_file_hash step
411    async fn handle_link_file_hash(
412        &self,
413        instance: &mut FileProcessingFlowInstance,
414    ) -> Result<Option<FileProcessingFlowStep>, FlowError>;
415
416    /// Handle create_new_hash step
417    async fn handle_create_new_hash(
418        &self,
419        instance: &mut FileProcessingFlowInstance,
420    ) -> Result<Option<FileProcessingFlowStep>, FlowError>;
421
422    /// Handle mark_complete step
423    async fn handle_mark_complete(
424        &self,
425        instance: &mut FileProcessingFlowInstance,
426    ) -> Result<Option<FileProcessingFlowStep>, FlowError>;
427
428    /// Handle emit_complete_event step
429    async fn handle_emit_complete_event(
430        &self,
431        instance: &mut FileProcessingFlowInstance,
432    ) -> Result<Option<FileProcessingFlowStep>, FlowError>;
433
434    /// Handle terminal step complete
435    async fn handle_complete(
436        &self,
437        instance: &mut FileProcessingFlowInstance,
438    ) -> Result<(), FlowError>;
439
440    /// Evaluate condition for retry_or_fail step
441    async fn evaluate_retry_or_fail(
442        &self,
443        instance: &FileProcessingFlowInstance,
444    ) -> Result<FileProcessingFlowStep, FlowError>;
445
446    /// Handle schedule_retry step
447    async fn handle_schedule_retry(
448        &self,
449        instance: &mut FileProcessingFlowInstance,
450    ) -> Result<Option<FileProcessingFlowStep>, FlowError>;
451
452    /// Handle terminal step retry_terminal
453    async fn handle_retry_terminal(
454        &self,
455        instance: &mut FileProcessingFlowInstance,
456    ) -> Result<(), FlowError>;
457
458    /// Handle mark_failed step
459    async fn handle_mark_failed(
460        &self,
461        instance: &mut FileProcessingFlowInstance,
462    ) -> Result<Option<FileProcessingFlowStep>, FlowError>;
463
464    /// Handle emit_failed_event step
465    async fn handle_emit_failed_event(
466        &self,
467        instance: &mut FileProcessingFlowInstance,
468    ) -> Result<Option<FileProcessingFlowStep>, FlowError>;
469
470    /// Handle terminal step failed_terminal
471    async fn handle_failed_terminal(
472        &self,
473        instance: &mut FileProcessingFlowInstance,
474    ) -> Result<(), FlowError>;
475
476    /// Handle terminal step job_not_found
477    async fn handle_job_not_found(
478        &self,
479        instance: &mut FileProcessingFlowInstance,
480    ) -> Result<(), FlowError>;
481
482    /// Handle terminal step file_not_found
483    async fn handle_file_not_found(
484        &self,
485        instance: &mut FileProcessingFlowInstance,
486    ) -> Result<(), FlowError>;
487
488    /// Handle terminal step unknown_job_type
489    async fn handle_unknown_job_type(
490        &self,
491        instance: &mut FileProcessingFlowInstance,
492    ) -> Result<(), FlowError>;
493
494    /// Execute compensation (rollback)
495    async fn compensate(
496        &self,
497        instance: &mut FileProcessingFlowInstance,
498    ) -> Result<(), FlowError>;
499
500}
501
502/// Executor for FileProcessing flow
503pub struct FileProcessingFlowExecutor<H: FileProcessingStepHandler> {
504    handler: Arc<H>,
505}
506
507impl<H: FileProcessingStepHandler> FileProcessingFlowExecutor<H> {
508    /// Create a new flow executor
509    pub fn new(handler: Arc<H>) -> Self {
510        Self { handler }
511    }
512
513    /// Start a new flow instance
514    pub async fn start(&self, instance_id: impl Into<String>) -> Result<FileProcessingFlowInstance, FlowError> {
515        let mut instance = FileProcessingFlowInstance::new(instance_id);
516        instance.status = FileProcessingFlowStatus::Running;
517        instance.current_step = Some(FileProcessingFlowStep::LoadJob);
518        Ok(instance)
519    }
520
521    /// Execute the current step
522    pub async fn execute_step(&self, instance: &mut FileProcessingFlowInstance) -> Result<(), FlowError> {
523        let current_step = match instance.current_step {
524            Some(step) => step,
525            None => return Err(FlowError::NoCurrentStep),
526        };
527
528        let next_step = match current_step {
529            FileProcessingFlowStep::LoadJob => {
530                self.handler.handle_load_job(instance).await?
531            }
532            FileProcessingFlowStep::MarkRunning => {
533                self.handler.handle_mark_running(instance).await?
534            }
535            FileProcessingFlowStep::LoadFile => {
536                self.handler.handle_load_file(instance).await?
537            }
538            FileProcessingFlowStep::LoadBucket => {
539                self.handler.handle_load_bucket(instance).await?
540            }
541            FileProcessingFlowStep::RouteByJobType => {
542                let next = self.handler.evaluate_route_by_job_type(instance).await?;
543                Some(next)
544            }
545            FileProcessingFlowStep::ProcessImageThumbnail => {
546                self.handler.handle_process_image_thumbnail(instance).await?
547            }
548            FileProcessingFlowStep::StoreImageThumbnail => {
549                self.handler.handle_store_image_thumbnail(instance).await?
550            }
551            FileProcessingFlowStep::UpdateFileThumbnail => {
552                self.handler.handle_update_file_thumbnail(instance).await?
553            }
554            FileProcessingFlowStep::ProcessVideoThumbnail => {
555                self.handler.handle_process_video_thumbnail(instance).await?
556            }
557            FileProcessingFlowStep::StoreVideoThumbnail => {
558                self.handler.handle_store_video_thumbnail(instance).await?
559            }
560            FileProcessingFlowStep::UpdateFileVideoThumbnail => {
561                self.handler.handle_update_file_video_thumbnail(instance).await?
562            }
563            FileProcessingFlowStep::ProcessDocumentPreview => {
564                self.handler.handle_process_document_preview(instance).await?
565            }
566            FileProcessingFlowStep::StoreDocumentPreview => {
567                self.handler.handle_store_document_preview(instance).await?
568            }
569            FileProcessingFlowStep::UpdateFileDocumentPreview => {
570                self.handler.handle_update_file_document_preview(instance).await?
571            }
572            FileProcessingFlowStep::ProcessCompression => {
573                self.handler.handle_process_compression(instance).await?
574            }
575            FileProcessingFlowStep::StoreCompressedFile => {
576                self.handler.handle_store_compressed_file(instance).await?
577            }
578            FileProcessingFlowStep::UpdateFileCompression => {
579                self.handler.handle_update_file_compression(instance).await?
580            }
581            FileProcessingFlowStep::UpdateBucketCompressionStats => {
582                self.handler.handle_update_bucket_compression_stats(instance).await?
583            }
584            FileProcessingFlowStep::ProcessVirusScan => {
585                self.handler.handle_process_virus_scan(instance).await?
586            }
587            FileProcessingFlowStep::CheckThreatLevel => {
588                let next = self.handler.evaluate_check_threat_level(instance).await?;
589                Some(next)
590            }
591            FileProcessingFlowStep::UpdateScanSafe => {
592                self.handler.handle_update_scan_safe(instance).await?
593            }
594            FileProcessingFlowStep::UpdateScanWarning => {
595                self.handler.handle_update_scan_warning(instance).await?
596            }
597            FileProcessingFlowStep::NotifyScanWarning => {
598                self.handler.handle_notify_scan_warning(instance).await?
599            }
600            FileProcessingFlowStep::UpdateScanThreat => {
601                self.handler.handle_update_scan_threat(instance).await?
602            }
603            FileProcessingFlowStep::NotifyThreatDetected => {
604                self.handler.handle_notify_threat_detected(instance).await?
605            }
606            FileProcessingFlowStep::ProcessDeduplication => {
607                self.handler.handle_process_deduplication(instance).await?
608            }
609            FileProcessingFlowStep::CheckExistingHash => {
610                self.handler.handle_check_existing_hash(instance).await?
611            }
612            FileProcessingFlowStep::UpdateExistingHash => {
613                self.handler.handle_update_existing_hash(instance).await?
614            }
615            FileProcessingFlowStep::LinkFileHash => {
616                self.handler.handle_link_file_hash(instance).await?
617            }
618            FileProcessingFlowStep::CreateNewHash => {
619                self.handler.handle_create_new_hash(instance).await?
620            }
621            FileProcessingFlowStep::MarkComplete => {
622                self.handler.handle_mark_complete(instance).await?
623            }
624            FileProcessingFlowStep::EmitCompleteEvent => {
625                self.handler.handle_emit_complete_event(instance).await?
626            }
627            FileProcessingFlowStep::Complete => {
628                self.handler.handle_complete(instance).await?;
629                None // Terminal step
630            }
631            FileProcessingFlowStep::RetryOrFail => {
632                let next = self.handler.evaluate_retry_or_fail(instance).await?;
633                Some(next)
634            }
635            FileProcessingFlowStep::ScheduleRetry => {
636                self.handler.handle_schedule_retry(instance).await?
637            }
638            FileProcessingFlowStep::RetryTerminal => {
639                self.handler.handle_retry_terminal(instance).await?;
640                None // Terminal step
641            }
642            FileProcessingFlowStep::MarkFailed => {
643                self.handler.handle_mark_failed(instance).await?
644            }
645            FileProcessingFlowStep::EmitFailedEvent => {
646                self.handler.handle_emit_failed_event(instance).await?
647            }
648            FileProcessingFlowStep::FailedTerminal => {
649                self.handler.handle_failed_terminal(instance).await?;
650                None // Terminal step
651            }
652            FileProcessingFlowStep::JobNotFound => {
653                self.handler.handle_job_not_found(instance).await?;
654                None // Terminal step
655            }
656            FileProcessingFlowStep::FileNotFound => {
657                self.handler.handle_file_not_found(instance).await?;
658                None // Terminal step
659            }
660            FileProcessingFlowStep::UnknownJobType => {
661                self.handler.handle_unknown_job_type(instance).await?;
662                None // Terminal step
663            }
664        };
665
666        // Mark current step as completed
667        instance.completed_steps.push(current_step);
668
669        // Move to next step
670        match next_step {
671            Some(next) => {
672                instance.current_step = Some(next);
673            }
674            None => {
675                instance.current_step = None;
676                instance.status = FileProcessingFlowStatus::Completed;
677            }
678        }
679
680        instance.updated_at = chrono::Utc::now();
681        Ok(())
682    }
683
684    /// Run the flow to completion
685    pub async fn run(&self, instance: &mut FileProcessingFlowInstance) -> Result<(), FlowError> {
686        while !instance.is_complete() && instance.status != FileProcessingFlowStatus::Waiting {
687            self.execute_step(instance).await?;
688        }
689        Ok(())
690    }
691
692    /// Cancel the flow
693    pub fn cancel(&self, instance: &mut FileProcessingFlowInstance) {
694        instance.status = FileProcessingFlowStatus::Cancelled;
695        instance.updated_at = chrono::Utc::now();
696    }
697
698    /// Mark flow as failed
699    pub fn fail(&self, instance: &mut FileProcessingFlowInstance, error: impl Into<String>) {
700        instance.status = FileProcessingFlowStatus::Failed;
701        instance.error = Some(error.into());
702        instance.updated_at = chrono::Utc::now();
703    }
704}
705
706#[cfg(test)]
707mod tests {
708    use super::*;
709
710    #[test]
711    fn test_flow_instance_creation() {
712        let instance = FileProcessingFlowInstance::new("test-1");
713        assert_eq!(instance.id, "test-1");
714        assert_eq!(instance.status, FileProcessingFlowStatus::Pending);
715        assert!(!instance.is_complete());
716    }
717
718    #[test]
719    fn test_flow_context() {
720        let mut instance = FileProcessingFlowInstance::new("test-2");
721        instance.set_context("key", serde_json::json!("value"));
722        let value = instance.get_context("key");
723        assert_eq!(value, Some(&serde_json::json!("value")));
724    }
725
726    #[test]
727    fn test_flow_status_transitions() {
728        let mut instance = FileProcessingFlowInstance::new("test-3");
729        assert!(!instance.is_running());
730
731        instance.status = FileProcessingFlowStatus::Running;
732        assert!(instance.is_running());
733        assert!(!instance.is_complete());
734
735        instance.status = FileProcessingFlowStatus::Completed;
736        assert!(instance.is_complete());
737        assert!(!instance.is_running());
738    }
739}
backbone_bucket/application/workflows/file_processing_workflow.rs

backbone_bucket/application/workflows/
file_processing_workflow.rs
