backbone_bucket/application/workflows/
multipart_upload_workflow.rs

1//! MultipartUpload flow implementation
2//!
3//! Generated by metaphor-schema
4//!
5//! Multipart upload workflow for large files including:
6//! - Session initialization
7//! - Chunk upload tracking
8//! - Progress updates
9//! - File assembly
10//! - Validation and processing
11
12use serde::{Deserialize, Serialize};
13use std::sync::Arc;
14use chrono;
15use backbone_core::flow::{WorkflowStep, WorkflowContext};
16
17/// Error type for flow execution
18#[derive(Debug, Clone, thiserror::Error)]
19pub enum FlowError {
20    #[error("No current step to execute")]
21    NoCurrentStep,
22
23    #[error("Step execution failed: {0}")]
24    StepFailed(String),
25
26    #[error("Condition evaluation failed: {0}")]
27    ConditionFailed(String),
28
29    #[error("Compensation failed: {0}")]
30    CompensationFailed(String),
31
32    #[error("Flow timed out")]
33    Timeout,
34
35    #[error("Flow cancelled")]
36    Cancelled,
37
38    #[error("Invalid state transition: {from} -> {to}")]
39    InvalidTransition { from: String, to: String },
40}
41
42/// Execution status for MultipartUpload flow
43#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
44#[serde(rename_all = "snake_case")]
45pub enum MultipartUploadFlowStatus {
46    /// Flow is pending execution
47    Pending,
48    /// Flow is currently running
49    Running,
50    /// Flow is waiting for an event or condition
51    Waiting,
52    /// Flow completed successfully
53    Completed,
54    /// Flow failed
55    Failed,
56    /// Flow was cancelled
57    Cancelled,
58    /// Flow is compensating (rolling back)
59    Compensating,
60}
61
62/// Steps in MultipartUpload flow
63#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
64#[serde(rename_all = "snake_case")]
65pub enum MultipartUploadFlowStep {
66    /// load_bucket
67    LoadBucket,
68    /// check_bucket_accessible
69    CheckBucketAccessible,
70    /// validate_file_size
71    ValidateFileSize,
72    /// check_max_file_size
73    CheckMaxFileSize,
74    /// calculate_chunk_params
75    CalculateChunkParams,
76    /// validate_mime_type
77    ValidateMimeType,
78    /// load_quota
79    LoadQuota,
80    /// create_default_quota
81    CreateDefaultQuota,
82    /// check_quota_available
83    CheckQuotaAvailable,
84    /// create_upload_session
85    CreateUploadSession,
86    /// reserve_quota
87    ReserveQuota,
88    /// emit_session_created
89    EmitSessionCreated,
90    /// complete
91    Complete,
92    /// bucket_not_found
93    BucketNotFound,
94    /// bucket_not_active
95    BucketNotActive,
96    /// invalid_file_size
97    InvalidFileSize,
98    /// file_too_large
99    FileTooLarge,
100    /// mime_type_not_allowed
101    MimeTypeNotAllowed,
102    /// quota_exceeded
103    QuotaExceeded,
104}
105
106/// Instance of MultipartUpload flow execution
107#[derive(Debug, Clone, Serialize, Deserialize)]
108pub struct MultipartUploadFlowInstance {
109    /// Unique instance ID
110    pub id: String,
111    /// Current status
112    pub status: MultipartUploadFlowStatus,
113    /// Current step
114    pub current_step: Option<MultipartUploadFlowStep>,
115    /// Flow context (variables)
116    pub context: serde_json::Value,
117    /// Completed steps
118    pub completed_steps: Vec<MultipartUploadFlowStep>,
119    /// Error if failed
120    pub error: Option<String>,
121    /// Created timestamp
122    pub created_at: chrono::DateTime<chrono::Utc>,
123    /// Updated timestamp
124    pub updated_at: chrono::DateTime<chrono::Utc>,
125}
126
127impl MultipartUploadFlowInstance {
128    /// Create a new flow instance
129    pub fn new(id: impl Into<String>) -> Self {
130        let now = chrono::Utc::now();
131        Self {
132            id: id.into(),
133            status: MultipartUploadFlowStatus::Pending,
134            current_step: None,
135            context: serde_json::json!({}),
136            completed_steps: Vec::new(),
137            error: None,
138            created_at: now,
139            updated_at: now,
140        }
141    }
142
143    /// Check if flow is complete
144    pub fn is_complete(&self) -> bool {
145        matches!(
146            self.status,
147            MultipartUploadFlowStatus::Completed | MultipartUploadFlowStatus::Failed | MultipartUploadFlowStatus::Cancelled
148        )
149    }
150
151    /// Check if flow is running
152    pub fn is_running(&self) -> bool {
153        matches!(
154            self.status,
155            MultipartUploadFlowStatus::Running | MultipartUploadFlowStatus::Waiting
156        )
157    }
158
159    /// Set a context variable
160    pub fn set_context(&mut self, key: &str, value: serde_json::Value) {
161        if let serde_json::Value::Object(ref mut map) = self.context {
162            map.insert(key.to_string(), value);
163        }
164        self.updated_at = chrono::Utc::now();
165    }
166
167    /// Get a context variable
168    pub fn get_context(&self, key: &str) -> Option<&serde_json::Value> {
169        if let serde_json::Value::Object(ref map) = self.context {
170            map.get(key)
171        } else {
172            None
173        }
174    }
175}
176
177// Phase 1: FlowInstance satisfies backbone_core::flow::WorkflowContext.
178// The entity() accessor requires the entity to be carried in the instance;
179// add `pub entity: MultipartUpload` in the // <<< CUSTOM section and remove the todo!.
180#[allow(unused_variables)]
181impl WorkflowContext<MultipartUploadFlowInstance> for MultipartUploadFlowInstance {
182    fn entity(&self) -> &MultipartUploadFlowInstance { self }
183    fn set_var(&mut self, key: &str, value: serde_json::Value) {
184        self.set_context(key, value);
185    }
186    fn get_var(&self, key: &str) -> Option<&serde_json::Value> {
187        self.get_context(key)
188    }
189}
190
191/// Step handler trait for MultipartUpload flow
192#[async_trait::async_trait]
193pub trait MultipartUploadStepHandler: Send + Sync {
194    /// Handle load_bucket step
195    async fn handle_load_bucket(
196        &self,
197        instance: &mut MultipartUploadFlowInstance,
198    ) -> Result<Option<MultipartUploadFlowStep>, FlowError>;
199
200    /// Evaluate condition for check_bucket_accessible step
201    async fn evaluate_check_bucket_accessible(
202        &self,
203        instance: &MultipartUploadFlowInstance,
204    ) -> Result<MultipartUploadFlowStep, FlowError>;
205
206    /// Evaluate condition for validate_file_size step
207    async fn evaluate_validate_file_size(
208        &self,
209        instance: &MultipartUploadFlowInstance,
210    ) -> Result<MultipartUploadFlowStep, FlowError>;
211
212    /// Evaluate condition for check_max_file_size step
213    async fn evaluate_check_max_file_size(
214        &self,
215        instance: &MultipartUploadFlowInstance,
216    ) -> Result<MultipartUploadFlowStep, FlowError>;
217
218    /// Handle calculate_chunk_params step
219    async fn handle_calculate_chunk_params(
220        &self,
221        instance: &mut MultipartUploadFlowInstance,
222    ) -> Result<Option<MultipartUploadFlowStep>, FlowError>;
223
224    /// Evaluate condition for validate_mime_type step
225    async fn evaluate_validate_mime_type(
226        &self,
227        instance: &MultipartUploadFlowInstance,
228    ) -> Result<MultipartUploadFlowStep, FlowError>;
229
230    /// Handle load_quota step
231    async fn handle_load_quota(
232        &self,
233        instance: &mut MultipartUploadFlowInstance,
234    ) -> Result<Option<MultipartUploadFlowStep>, FlowError>;
235
236    /// Handle create_default_quota step
237    async fn handle_create_default_quota(
238        &self,
239        instance: &mut MultipartUploadFlowInstance,
240    ) -> Result<Option<MultipartUploadFlowStep>, FlowError>;
241
242    /// Evaluate condition for check_quota_available step
243    async fn evaluate_check_quota_available(
244        &self,
245        instance: &MultipartUploadFlowInstance,
246    ) -> Result<MultipartUploadFlowStep, FlowError>;
247
248    /// Handle create_upload_session step
249    async fn handle_create_upload_session(
250        &self,
251        instance: &mut MultipartUploadFlowInstance,
252    ) -> Result<Option<MultipartUploadFlowStep>, FlowError>;
253
254    /// Handle reserve_quota step
255    async fn handle_reserve_quota(
256        &self,
257        instance: &mut MultipartUploadFlowInstance,
258    ) -> Result<Option<MultipartUploadFlowStep>, FlowError>;
259
260    /// Handle emit_session_created step
261    async fn handle_emit_session_created(
262        &self,
263        instance: &mut MultipartUploadFlowInstance,
264    ) -> Result<Option<MultipartUploadFlowStep>, FlowError>;
265
266    /// Handle terminal step complete
267    async fn handle_complete(
268        &self,
269        instance: &mut MultipartUploadFlowInstance,
270    ) -> Result<(), FlowError>;
271
272    /// Handle terminal step bucket_not_found
273    async fn handle_bucket_not_found(
274        &self,
275        instance: &mut MultipartUploadFlowInstance,
276    ) -> Result<(), FlowError>;
277
278    /// Handle terminal step bucket_not_active
279    async fn handle_bucket_not_active(
280        &self,
281        instance: &mut MultipartUploadFlowInstance,
282    ) -> Result<(), FlowError>;
283
284    /// Handle terminal step invalid_file_size
285    async fn handle_invalid_file_size(
286        &self,
287        instance: &mut MultipartUploadFlowInstance,
288    ) -> Result<(), FlowError>;
289
290    /// Handle terminal step file_too_large
291    async fn handle_file_too_large(
292        &self,
293        instance: &mut MultipartUploadFlowInstance,
294    ) -> Result<(), FlowError>;
295
296    /// Handle terminal step mime_type_not_allowed
297    async fn handle_mime_type_not_allowed(
298        &self,
299        instance: &mut MultipartUploadFlowInstance,
300    ) -> Result<(), FlowError>;
301
302    /// Handle terminal step quota_exceeded
303    async fn handle_quota_exceeded(
304        &self,
305        instance: &mut MultipartUploadFlowInstance,
306    ) -> Result<(), FlowError>;
307
308}
309
310/// Executor for MultipartUpload flow
311pub struct MultipartUploadFlowExecutor<H: MultipartUploadStepHandler> {
312    handler: Arc<H>,
313}
314
315impl<H: MultipartUploadStepHandler> MultipartUploadFlowExecutor<H> {
316    /// Create a new flow executor
317    pub fn new(handler: Arc<H>) -> Self {
318        Self { handler }
319    }
320
321    /// Start a new flow instance
322    pub async fn start(&self, instance_id: impl Into<String>) -> Result<MultipartUploadFlowInstance, FlowError> {
323        let mut instance = MultipartUploadFlowInstance::new(instance_id);
324        instance.status = MultipartUploadFlowStatus::Running;
325        instance.current_step = Some(MultipartUploadFlowStep::LoadBucket);
326        Ok(instance)
327    }
328
329    /// Execute the current step
330    pub async fn execute_step(&self, instance: &mut MultipartUploadFlowInstance) -> Result<(), FlowError> {
331        let current_step = match instance.current_step {
332            Some(step) => step,
333            None => return Err(FlowError::NoCurrentStep),
334        };
335
336        let next_step = match current_step {
337            MultipartUploadFlowStep::LoadBucket => {
338                self.handler.handle_load_bucket(instance).await?
339            }
340            MultipartUploadFlowStep::CheckBucketAccessible => {
341                let next = self.handler.evaluate_check_bucket_accessible(instance).await?;
342                Some(next)
343            }
344            MultipartUploadFlowStep::ValidateFileSize => {
345                let next = self.handler.evaluate_validate_file_size(instance).await?;
346                Some(next)
347            }
348            MultipartUploadFlowStep::CheckMaxFileSize => {
349                let next = self.handler.evaluate_check_max_file_size(instance).await?;
350                Some(next)
351            }
352            MultipartUploadFlowStep::CalculateChunkParams => {
353                self.handler.handle_calculate_chunk_params(instance).await?
354            }
355            MultipartUploadFlowStep::ValidateMimeType => {
356                let next = self.handler.evaluate_validate_mime_type(instance).await?;
357                Some(next)
358            }
359            MultipartUploadFlowStep::LoadQuota => {
360                self.handler.handle_load_quota(instance).await?
361            }
362            MultipartUploadFlowStep::CreateDefaultQuota => {
363                self.handler.handle_create_default_quota(instance).await?
364            }
365            MultipartUploadFlowStep::CheckQuotaAvailable => {
366                let next = self.handler.evaluate_check_quota_available(instance).await?;
367                Some(next)
368            }
369            MultipartUploadFlowStep::CreateUploadSession => {
370                self.handler.handle_create_upload_session(instance).await?
371            }
372            MultipartUploadFlowStep::ReserveQuota => {
373                self.handler.handle_reserve_quota(instance).await?
374            }
375            MultipartUploadFlowStep::EmitSessionCreated => {
376                self.handler.handle_emit_session_created(instance).await?
377            }
378            MultipartUploadFlowStep::Complete => {
379                self.handler.handle_complete(instance).await?;
380                None // Terminal step
381            }
382            MultipartUploadFlowStep::BucketNotFound => {
383                self.handler.handle_bucket_not_found(instance).await?;
384                None // Terminal step
385            }
386            MultipartUploadFlowStep::BucketNotActive => {
387                self.handler.handle_bucket_not_active(instance).await?;
388                None // Terminal step
389            }
390            MultipartUploadFlowStep::InvalidFileSize => {
391                self.handler.handle_invalid_file_size(instance).await?;
392                None // Terminal step
393            }
394            MultipartUploadFlowStep::FileTooLarge => {
395                self.handler.handle_file_too_large(instance).await?;
396                None // Terminal step
397            }
398            MultipartUploadFlowStep::MimeTypeNotAllowed => {
399                self.handler.handle_mime_type_not_allowed(instance).await?;
400                None // Terminal step
401            }
402            MultipartUploadFlowStep::QuotaExceeded => {
403                self.handler.handle_quota_exceeded(instance).await?;
404                None // Terminal step
405            }
406        };
407
408        // Mark current step as completed
409        instance.completed_steps.push(current_step);
410
411        // Move to next step
412        match next_step {
413            Some(next) => {
414                instance.current_step = Some(next);
415            }
416            None => {
417                instance.current_step = None;
418                instance.status = MultipartUploadFlowStatus::Completed;
419            }
420        }
421
422        instance.updated_at = chrono::Utc::now();
423        Ok(())
424    }
425
426    /// Run the flow to completion
427    pub async fn run(&self, instance: &mut MultipartUploadFlowInstance) -> Result<(), FlowError> {
428        while !instance.is_complete() && instance.status != MultipartUploadFlowStatus::Waiting {
429            self.execute_step(instance).await?;
430        }
431        Ok(())
432    }
433
434    /// Cancel the flow
435    pub fn cancel(&self, instance: &mut MultipartUploadFlowInstance) {
436        instance.status = MultipartUploadFlowStatus::Cancelled;
437        instance.updated_at = chrono::Utc::now();
438    }
439
440    /// Mark flow as failed
441    pub fn fail(&self, instance: &mut MultipartUploadFlowInstance, error: impl Into<String>) {
442        instance.status = MultipartUploadFlowStatus::Failed;
443        instance.error = Some(error.into());
444        instance.updated_at = chrono::Utc::now();
445    }
446}
447
448#[cfg(test)]
449mod tests {
450    use super::*;
451
452    #[test]
453    fn test_flow_instance_creation() {
454        let instance = MultipartUploadFlowInstance::new("test-1");
455        assert_eq!(instance.id, "test-1");
456        assert_eq!(instance.status, MultipartUploadFlowStatus::Pending);
457        assert!(!instance.is_complete());
458    }
459
460    #[test]
461    fn test_flow_context() {
462        let mut instance = MultipartUploadFlowInstance::new("test-2");
463        instance.set_context("key", serde_json::json!("value"));
464        let value = instance.get_context("key");
465        assert_eq!(value, Some(&serde_json::json!("value")));
466    }
467
468    #[test]
469    fn test_flow_status_transitions() {
470        let mut instance = MultipartUploadFlowInstance::new("test-3");
471        assert!(!instance.is_running());
472
473        instance.status = MultipartUploadFlowStatus::Running;
474        assert!(instance.is_running());
475        assert!(!instance.is_complete());
476
477        instance.status = MultipartUploadFlowStatus::Completed;
478        assert!(instance.is_complete());
479        assert!(!instance.is_running());
480    }
481}
backbone_bucket/application/workflows/multipart_upload_workflow.rs

backbone_bucket/application/workflows/
multipart_upload_workflow.rs
