oximedia_workflow/
task.rs

1//! Task definitions and types.
2
3use crate::error::{Result, WorkflowError};
4use serde::{Deserialize, Serialize};
5use std::collections::HashMap;
6use std::path::PathBuf;
7use std::time::Duration;
8use uuid::Uuid;
9
10/// Unique task identifier.
11#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
12pub struct TaskId(Uuid);
13
14impl TaskId {
15    /// Create a new random task ID.
16    #[must_use]
17    pub fn new() -> Self {
18        Self(Uuid::new_v4())
19    }
20
21    /// Get the underlying UUID.
22    #[must_use]
23    pub const fn as_uuid(&self) -> &Uuid {
24        &self.0
25    }
26}
27
28impl Default for TaskId {
29    fn default() -> Self {
30        Self::new()
31    }
32}
33
34impl std::fmt::Display for TaskId {
35    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
36        write!(f, "{}", self.0)
37    }
38}
39
40impl From<Uuid> for TaskId {
41    fn from(uuid: Uuid) -> Self {
42        Self(uuid)
43    }
44}
45
46/// Task state.
47#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
48#[serde(rename_all = "lowercase")]
49#[derive(Default)]
50pub enum TaskState {
51    /// Task is pending execution.
52    #[default]
53    Pending,
54    /// Task is queued for execution.
55    Queued,
56    /// Task is currently running.
57    Running,
58    /// Task completed successfully.
59    Completed,
60    /// Task failed.
61    Failed,
62    /// Task was cancelled.
63    Cancelled,
64    /// Task is waiting for dependencies.
65    Waiting,
66    /// Task is retrying after failure.
67    Retrying,
68    /// Task is skipped due to conditions.
69    Skipped,
70}
71
72impl TaskState {
73    /// Check if task is in a terminal state.
74    #[must_use]
75    pub const fn is_terminal(&self) -> bool {
76        matches!(
77            self,
78            Self::Completed | Self::Failed | Self::Cancelled | Self::Skipped
79        )
80    }
81
82    /// Check if task is active.
83    #[must_use]
84    pub const fn is_active(&self) -> bool {
85        matches!(self, Self::Running | Self::Retrying)
86    }
87
88    /// Check if task can be started.
89    #[must_use]
90    pub const fn can_start(&self) -> bool {
91        matches!(self, Self::Pending | Self::Queued)
92    }
93}
94
95/// Task priority.
96#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, Default)]
97pub enum TaskPriority {
98    /// Low priority.
99    Low = 0,
100    /// Normal priority.
101    #[default]
102    Normal = 1,
103    /// High priority.
104    High = 2,
105    /// Critical priority.
106    Critical = 3,
107}
108
109/// Retry policy for tasks.
110#[derive(Debug, Clone, Serialize, Deserialize)]
111pub struct RetryPolicy {
112    /// Maximum number of retry attempts.
113    pub max_attempts: u32,
114    /// Initial delay before retry.
115    pub initial_delay: Duration,
116    /// Maximum delay between retries.
117    pub max_delay: Duration,
118    /// Backoff multiplier.
119    pub backoff_multiplier: f64,
120    /// Whether to use exponential backoff.
121    pub exponential_backoff: bool,
122}
123
124impl Default for RetryPolicy {
125    fn default() -> Self {
126        Self {
127            max_attempts: 3,
128            initial_delay: Duration::from_secs(1),
129            max_delay: Duration::from_secs(60),
130            backoff_multiplier: 2.0,
131            exponential_backoff: true,
132        }
133    }
134}
135
136impl RetryPolicy {
137    /// Calculate delay for a specific attempt.
138    #[must_use]
139    pub fn delay_for_attempt(&self, attempt: u32) -> Duration {
140        if !self.exponential_backoff {
141            return self.initial_delay;
142        }
143
144        let delay = self.initial_delay.as_secs_f64()
145            * self
146                .backoff_multiplier
147                .powi(i32::try_from(attempt).unwrap_or(10));
148        let delay = delay.min(self.max_delay.as_secs_f64());
149        Duration::from_secs_f64(delay)
150    }
151
152    /// Check if should retry given the attempt count.
153    #[must_use]
154    pub const fn should_retry(&self, attempt: u32) -> bool {
155        attempt < self.max_attempts
156    }
157}
158
159/// Task type enumeration.
160#[derive(Debug, Clone, Serialize, Deserialize)]
161#[serde(tag = "type", rename_all = "snake_case")]
162pub enum TaskType {
163    /// Transcode media file.
164    Transcode {
165        /// Input file path.
166        input: PathBuf,
167        /// Output file path.
168        output: PathBuf,
169        /// Preset name.
170        preset: String,
171        /// Additional parameters.
172        #[serde(default)]
173        params: HashMap<String, serde_json::Value>,
174    },
175
176    /// Quality control validation.
177    QualityControl {
178        /// Input file path.
179        input: PathBuf,
180        /// QC profile name.
181        profile: String,
182        /// Validation rules.
183        #[serde(default)]
184        rules: Vec<String>,
185    },
186
187    /// File transfer operation.
188    Transfer {
189        /// Source path or URL.
190        source: String,
191        /// Destination path or URL.
192        destination: String,
193        /// Transfer protocol.
194        protocol: TransferProtocol,
195        /// Transfer options.
196        #[serde(default)]
197        options: HashMap<String, String>,
198    },
199
200    /// Send notification.
201    Notification {
202        /// Notification channel.
203        channel: NotificationChannel,
204        /// Message content.
205        message: String,
206        /// Additional metadata.
207        #[serde(default)]
208        metadata: HashMap<String, String>,
209    },
210
211    /// Execute custom script.
212    CustomScript {
213        /// Script path.
214        script: PathBuf,
215        /// Script arguments.
216        #[serde(default)]
217        args: Vec<String>,
218        /// Environment variables.
219        #[serde(default)]
220        env: HashMap<String, String>,
221    },
222
223    /// Media analysis task.
224    Analysis {
225        /// Input file path.
226        input: PathBuf,
227        /// Analysis types to perform.
228        analyses: Vec<AnalysisType>,
229        /// Output path for results.
230        output: Option<PathBuf>,
231    },
232
233    /// Conditional decision task.
234    Conditional {
235        /// Condition expression.
236        condition: String,
237        /// Task to execute if true.
238        true_task: Option<Box<Task>>,
239        /// Task to execute if false.
240        false_task: Option<Box<Task>>,
241    },
242
243    /// Wait for duration.
244    Wait {
245        /// Duration to wait.
246        duration: Duration,
247    },
248
249    /// HTTP request task.
250    HttpRequest {
251        /// Request URL.
252        url: String,
253        /// HTTP method.
254        method: HttpMethod,
255        /// Request headers.
256        #[serde(default)]
257        headers: HashMap<String, String>,
258        /// Request body.
259        body: Option<String>,
260    },
261}
262
263/// Transfer protocol types.
264#[derive(Debug, Clone, Serialize, Deserialize)]
265#[serde(rename_all = "lowercase")]
266pub enum TransferProtocol {
267    /// Local file system.
268    Local,
269    /// FTP transfer.
270    Ftp,
271    /// SFTP transfer.
272    Sftp,
273    /// Amazon S3.
274    S3,
275    /// HTTP/HTTPS.
276    Http,
277    /// Rsync.
278    Rsync,
279}
280
281/// Notification channel types.
282#[derive(Debug, Clone, Serialize, Deserialize)]
283#[serde(rename_all = "lowercase")]
284pub enum NotificationChannel {
285    /// Email notification.
286    Email {
287        /// Recipient addresses.
288        to: Vec<String>,
289        /// Subject line.
290        subject: String,
291    },
292    /// Webhook notification.
293    Webhook {
294        /// Webhook URL.
295        url: String,
296    },
297    /// Slack notification.
298    Slack {
299        /// Slack channel.
300        channel: String,
301        /// Webhook URL.
302        webhook_url: String,
303    },
304    /// Discord notification.
305    Discord {
306        /// Webhook URL.
307        webhook_url: String,
308    },
309}
310
311/// Media analysis types.
312#[derive(Debug, Clone, Serialize, Deserialize)]
313#[serde(rename_all = "snake_case")]
314pub enum AnalysisType {
315    /// Audio level analysis.
316    AudioLevels,
317    /// Video quality analysis.
318    VideoQuality,
319    /// Scene detection.
320    SceneDetection,
321    /// Black frame detection.
322    BlackFrames,
323    /// Silence detection.
324    Silence,
325    /// Color analysis.
326    Color,
327    /// Motion analysis.
328    Motion,
329}
330
331/// HTTP methods.
332#[derive(Debug, Clone, Serialize, Deserialize)]
333#[serde(rename_all = "UPPERCASE")]
334pub enum HttpMethod {
335    /// GET method.
336    Get,
337    /// POST method.
338    Post,
339    /// PUT method.
340    Put,
341    /// DELETE method.
342    Delete,
343    /// PATCH method.
344    Patch,
345}
346
347/// Task definition.
348#[derive(Debug, Clone, Serialize, Deserialize)]
349pub struct Task {
350    /// Unique task identifier.
351    pub id: TaskId,
352    /// Task name.
353    pub name: String,
354    /// Task type and configuration.
355    pub task_type: TaskType,
356    /// Current state.
357    #[serde(default)]
358    pub state: TaskState,
359    /// Task priority.
360    #[serde(default)]
361    pub priority: TaskPriority,
362    /// Retry policy.
363    #[serde(default)]
364    pub retry: RetryPolicy,
365    /// Execution timeout.
366    #[serde(default = "default_timeout")]
367    pub timeout: Duration,
368    /// Task dependencies.
369    #[serde(default)]
370    pub dependencies: Vec<TaskId>,
371    /// Task metadata.
372    #[serde(default)]
373    pub metadata: HashMap<String, String>,
374    /// Number of retry attempts so far.
375    #[serde(default)]
376    pub retry_count: u32,
377    /// Task conditions (must evaluate to true to run).
378    #[serde(default)]
379    pub conditions: Vec<String>,
380}
381
382fn default_timeout() -> Duration {
383    Duration::from_secs(3600) // 1 hour
384}
385
386impl Task {
387    /// Create a new task.
388    #[must_use]
389    pub fn new(name: impl Into<String>, task_type: TaskType) -> Self {
390        Self {
391            id: TaskId::new(),
392            name: name.into(),
393            task_type,
394            state: TaskState::Pending,
395            priority: TaskPriority::default(),
396            retry: RetryPolicy::default(),
397            timeout: default_timeout(),
398            dependencies: Vec::new(),
399            metadata: HashMap::new(),
400            retry_count: 0,
401            conditions: Vec::new(),
402        }
403    }
404
405    /// Add a dependency to this task.
406    pub fn add_dependency(&mut self, task_id: TaskId) {
407        if !self.dependencies.contains(&task_id) {
408            self.dependencies.push(task_id);
409        }
410    }
411
412    /// Set task priority.
413    #[must_use]
414    pub fn with_priority(mut self, priority: TaskPriority) -> Self {
415        self.priority = priority;
416        self
417    }
418
419    /// Set retry policy.
420    #[must_use]
421    pub fn with_retry(mut self, retry: RetryPolicy) -> Self {
422        self.retry = retry;
423        self
424    }
425
426    /// Set timeout.
427    #[must_use]
428    pub fn with_timeout(mut self, timeout: Duration) -> Self {
429        self.timeout = timeout;
430        self
431    }
432
433    /// Add metadata.
434    pub fn with_metadata(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
435        self.metadata.insert(key.into(), value.into());
436        self
437    }
438
439    /// Add condition.
440    pub fn with_condition(mut self, condition: impl Into<String>) -> Self {
441        self.conditions.push(condition.into());
442        self
443    }
444
445    /// Check if task can be executed.
446    #[must_use]
447    pub const fn can_execute(&self) -> bool {
448        self.state.can_start()
449    }
450
451    /// Update task state.
452    pub fn set_state(&mut self, state: TaskState) -> Result<()> {
453        // Validate state transition
454        match (&self.state, &state) {
455            (TaskState::Completed | TaskState::Failed | TaskState::Cancelled, _)
456                if !matches!(state, TaskState::Pending) =>
457            {
458                return Err(WorkflowError::InvalidStateTransition {
459                    from: format!("{:?}", self.state),
460                    to: format!("{state:?}"),
461                });
462            }
463            _ => {}
464        }
465
466        self.state = state;
467        Ok(())
468    }
469
470    /// Increment retry count.
471    pub fn increment_retry(&mut self) {
472        self.retry_count += 1;
473    }
474
475    /// Check if should retry.
476    #[must_use]
477    pub fn should_retry(&self) -> bool {
478        self.retry.should_retry(self.retry_count)
479    }
480
481    /// Get retry delay.
482    #[must_use]
483    pub fn retry_delay(&self) -> Duration {
484        self.retry.delay_for_attempt(self.retry_count)
485    }
486}
487
488/// Task execution result.
489#[derive(Debug, Clone, Serialize, Deserialize)]
490pub struct TaskResult {
491    /// Task identifier.
492    pub task_id: TaskId,
493    /// Execution status.
494    pub status: TaskState,
495    /// Result data.
496    pub data: Option<serde_json::Value>,
497    /// Error message if failed.
498    pub error: Option<String>,
499    /// Execution duration.
500    pub duration: Duration,
501    /// Output files produced.
502    #[serde(default)]
503    pub outputs: Vec<PathBuf>,
504}
505
506#[cfg(test)]
507mod tests {
508    use super::*;
509
510    #[test]
511    fn test_task_id_creation() {
512        let id1 = TaskId::new();
513        let id2 = TaskId::new();
514        assert_ne!(id1, id2);
515    }
516
517    #[test]
518    fn test_task_state_terminal() {
519        assert!(TaskState::Completed.is_terminal());
520        assert!(TaskState::Failed.is_terminal());
521        assert!(!TaskState::Running.is_terminal());
522    }
523
524    #[test]
525    fn test_task_state_active() {
526        assert!(TaskState::Running.is_active());
527        assert!(!TaskState::Pending.is_active());
528    }
529
530    #[test]
531    fn test_retry_policy_delay() {
532        let policy = RetryPolicy::default();
533        let delay1 = policy.delay_for_attempt(0);
534        let delay2 = policy.delay_for_attempt(1);
535        assert!(delay2 > delay1);
536    }
537
538    #[test]
539    fn test_retry_policy_max_attempts() {
540        let policy = RetryPolicy {
541            max_attempts: 3,
542            ..Default::default()
543        };
544        assert!(policy.should_retry(0));
545        assert!(policy.should_retry(2));
546        assert!(!policy.should_retry(3));
547    }
548
549    #[test]
550    fn test_task_creation() {
551        let task = Task::new(
552            "test-task",
553            TaskType::Wait {
554                duration: Duration::from_secs(10),
555            },
556        );
557        assert_eq!(task.name, "test-task");
558        assert_eq!(task.state, TaskState::Pending);
559    }
560
561    #[test]
562    fn test_task_with_priority() {
563        let task = Task::new(
564            "test",
565            TaskType::Wait {
566                duration: Duration::from_secs(1),
567            },
568        )
569        .with_priority(TaskPriority::High);
570        assert_eq!(task.priority, TaskPriority::High);
571    }
572
573    #[test]
574    fn test_task_add_dependency() {
575        let mut task = Task::new(
576            "test",
577            TaskType::Wait {
578                duration: Duration::from_secs(1),
579            },
580        );
581        let dep_id = TaskId::new();
582        task.add_dependency(dep_id);
583        assert_eq!(task.dependencies.len(), 1);
584        assert_eq!(task.dependencies[0], dep_id);
585    }
586
587    #[test]
588    fn test_task_state_transition() {
589        let mut task = Task::new(
590            "test",
591            TaskType::Wait {
592                duration: Duration::from_secs(1),
593            },
594        );
595        assert!(task.set_state(TaskState::Running).is_ok());
596        assert!(task.set_state(TaskState::Completed).is_ok());
597        assert!(task.set_state(TaskState::Running).is_err());
598    }
599
600    #[test]
601    fn test_task_retry_logic() {
602        let mut task = Task::new(
603            "test",
604            TaskType::Wait {
605                duration: Duration::from_secs(1),
606            },
607        )
608        .with_retry(RetryPolicy {
609            max_attempts: 2,
610            ..Default::default()
611        });
612
613        assert!(task.should_retry());
614        task.increment_retry();
615        assert!(task.should_retry());
616        task.increment_retry();
617        assert!(!task.should_retry());
618    }
619
620    #[test]
621    fn test_task_priority_ordering() {
622        assert!(TaskPriority::Critical > TaskPriority::High);
623        assert!(TaskPriority::High > TaskPriority::Normal);
624        assert!(TaskPriority::Normal > TaskPriority::Low);
625    }
626
627    #[test]
628    fn test_retry_policy_exponential_backoff() {
629        let policy = RetryPolicy {
630            max_attempts: 5,
631            initial_delay: Duration::from_secs(1),
632            max_delay: Duration::from_secs(60),
633            backoff_multiplier: 2.0,
634            exponential_backoff: true,
635        };
636
637        let delay0 = policy.delay_for_attempt(0);
638        let delay1 = policy.delay_for_attempt(1);
639        let delay2 = policy.delay_for_attempt(2);
640
641        assert_eq!(delay0.as_secs(), 1);
642        assert_eq!(delay1.as_secs(), 2);
643        assert_eq!(delay2.as_secs(), 4);
644    }
645}
oximedia_workflow/task.rs

oximedia_workflow/
task.rs
