oximedia_workflow/
lib.rs

1//! Comprehensive workflow orchestration engine for `OxiMedia`.
2//!
3//! This crate provides a complete workflow orchestration system with:
4//! - DAG-based workflow definition
5//! - Task dependencies and parallel execution
6//! - State persistence with `SQLite`
7//! - Cron-style scheduling
8//! - REST API for workflow management
9//! - Real-time monitoring and metrics
10//! - Multiple task types (transcode, QC, transfer, etc.)
11//!
12//! # Examples
13//!
14//! ## Creating a Simple Workflow
15//!
16//! ```rust
17//! use oximedia_workflow::{Workflow, Task, TaskType};
18//! use std::time::Duration;
19//!
20//! let mut workflow = Workflow::new("my-workflow");
21//!
22//! let task = Task::new("wait-task", TaskType::Wait {
23//!     duration: Duration::from_secs(5),
24//! });
25//!
26//! workflow.add_task(task);
27//! ```
28//!
29//! ## Creating a Multi-Pass Encoding Workflow
30//!
31//! ```rust
32//! use oximedia_workflow::patterns::multi_pass_encoding;
33//! use std::path::PathBuf;
34//!
35//! let workflow = multi_pass_encoding(
36//!     PathBuf::from("/source.mp4"),
37//!     PathBuf::from("/proxy.mp4"),
38//!     PathBuf::from("/output.mp4"),
39//!     "broadcast".to_string(),
40//! );
41//! ```
42
43#![forbid(unsafe_code)]
44#![warn(missing_docs)]
45
46#[cfg(all(not(target_arch = "wasm32"), feature = "sqlite"))]
47pub mod api;
48pub mod approval_gate;
49pub mod audit_log;
50pub mod batch_status;
51pub mod builder;
52pub mod checkpoint;
53pub mod circuit_breaker;
54#[cfg(all(not(target_arch = "wasm32"), feature = "sqlite"))]
55pub mod cli;
56pub mod cost_tracking;
57pub mod dag;
58pub mod error;
59pub mod event_bus;
60#[cfg(not(target_arch = "wasm32"))]
61pub mod executor;
62pub mod monitoring;
63pub mod notification_system;
64pub mod parallel_steps;
65pub mod patterns;
66#[cfg(all(not(target_arch = "wasm32"), feature = "sqlite"))]
67pub mod persistence;
68#[cfg(not(target_arch = "wasm32"))]
69pub mod queue;
70pub mod resource_pool;
71pub mod retry_policy;
72#[cfg(not(target_arch = "wasm32"))]
73pub mod scheduler;
74pub mod sla;
75pub mod sla_tracking;
76pub mod state_machine;
77pub mod step_condition;
78pub mod step_conditions;
79pub mod step_result;
80pub mod task;
81pub mod task_dependency;
82pub mod task_graph;
83pub mod task_priority_queue;
84pub mod task_template;
85pub mod templates;
86pub mod triggers;
87pub mod utils;
88pub mod validation;
89pub mod webhook;
90#[cfg(not(target_arch = "wasm32"))]
91pub mod websocket;
92pub mod workflow;
93pub mod workflow_audit;
94pub mod workflow_checkpoint;
95pub mod workflow_compose;
96pub mod workflow_diff;
97pub mod workflow_health_check;
98pub mod workflow_import_export;
99pub mod workflow_log;
100pub mod workflow_metrics;
101pub mod workflow_migration;
102pub mod workflow_simulation;
103pub mod workflow_snapshot;
104pub mod workflow_template;
105pub mod workflow_throttle;
106pub mod workflow_version;
107
108// Re-exports for convenience
109pub use batch_status::{BatchStatusWriter, FlushResult, StatusUpdate};
110pub use builder::{
111    QcTaskBuilder, TaskBuilder, TranscodeTaskBuilder, TransferTaskBuilder, WorkflowBuilder,
112};
113pub use circuit_breaker::{
114    CircuitBreaker, CircuitBreakerConfig, CircuitBreakerMetrics, CircuitBreakerRegistry,
115    CircuitBreakerSummary, CircuitDecision, CircuitState,
116};
117#[cfg(all(not(target_arch = "wasm32"), feature = "sqlite"))]
118pub use cli::Cli;
119pub use dag::{
120    audio_normalize, ingest_transcode, subtitle_burn, BranchEvaluator, BranchNode, BranchType,
121    DagError, DagRunStatus, DagWorkflowEngine, NodeId, NodeStatus, WorkflowDag, WorkflowEdge,
122    WorkflowNode, WorkflowTemplate,
123};
124pub use error::{Result, WorkflowError};
125pub use event_bus::{
126    BusEvent, EventBus, EventBusConfig, EventBusStats, EventFilter, SubscriptionId,
127};
128#[cfg(not(target_arch = "wasm32"))]
129pub use executor::{
130    parse_condition, DefaultTaskExecutor, ExecutionContext, TaskExecutor, WorkflowControl,
131    WorkflowExecutor,
132};
133pub use monitoring::{MonitoringService, SystemStatistics, TaskMetrics, WorkflowMetrics};
134pub use parallel_steps::{
135    execute_step, ParallelStepError, ParallelSteps, StepResult, StepType, WorkflowStage,
136    WorkflowStep,
137};
138pub use patterns::*;
139#[cfg(all(not(target_arch = "wasm32"), feature = "sqlite"))]
140pub use persistence::PersistenceManager;
141#[cfg(not(target_arch = "wasm32"))]
142pub use queue::{QueueStatistics, TaskQueue};
143pub use retry_policy::{ExponentialRetryPolicy, RetryDecision, RetryPolicyState};
144#[cfg(not(target_arch = "wasm32"))]
145pub use scheduler::{FileWatcher, ScheduledWorkflow, Trigger, WorkflowScheduler};
146pub use task::{
147    AnalysisType, HttpMethod, NotificationChannel, RetryPolicy, Task, TaskId, TaskPriority,
148    TaskResult, TaskState, TaskType, TransferProtocol,
149};
150pub use task_priority_queue::{PriorityEntry, PriorityLevel, TaskPriorityQueue};
151pub use utils::{
152    calculate_parallelism, clone_workflow, estimate_workflow_duration, expand_env_vars,
153    expand_template, find_critical_path, format_duration, generate_task_name,
154    get_workflow_statistics, merge_configs, normalize_paths, parse_duration, sanitize_task_name,
155    WorkflowStatistics,
156};
157pub use validation::{
158    ComplexityAnalyzer, ComplexityLevel, ComplexityMetrics, TaskValidator, ValidationReport,
159    ValidationRule, WorkflowValidator,
160};
161pub use webhook::{WebhookConfig, WebhookEvent, WebhookNotifier, WorkflowContext};
162#[cfg(not(target_arch = "wasm32"))]
163pub use websocket::{WebSocketManager, WebSocketState, WorkflowEvent};
164pub use workflow::{Edge, Workflow, WorkflowConfig, WorkflowId, WorkflowState};
165pub use workflow_import_export::{
166    deserialize_bundle, export_workflow, import_workflow, serialize_bundle, BundleFormat,
167    DagBundle, WorkflowBundle, WorkflowDefinition,
168};
169pub use workflow_migration::{
170    FieldChange, MigrationError, MigrationRegistry, MigrationStep, SchemaVersion,
171};
172pub use workflow_simulation::{
173    quick_simulate, ConditionSource, EvaluatedCondition, SimulatedOutcome, SimulationConfig,
174    SimulationResult, SimulationSummary, WorkflowSimulator,
175};
176
177/// Workflow engine - main entry point for the orchestration system.
178#[cfg(all(not(target_arch = "wasm32"), feature = "sqlite"))]
179pub struct WorkflowEngine {
180    persistence: std::sync::Arc<PersistenceManager>,
181    scheduler: std::sync::Arc<WorkflowScheduler>,
182    monitoring: std::sync::Arc<MonitoringService>,
183    executor: std::sync::Arc<dyn TaskExecutor>,
184}
185
186#[cfg(all(not(target_arch = "wasm32"), feature = "sqlite"))]
187impl WorkflowEngine {
188    /// Create a new workflow engine with the specified database path.
189    ///
190    /// # Errors
191    ///
192    /// Returns an error if the database cannot be initialized.
193    pub fn new(db_path: impl AsRef<std::path::Path>) -> Result<Self> {
194        let persistence = std::sync::Arc::new(PersistenceManager::new(db_path)?);
195        let scheduler = std::sync::Arc::new(WorkflowScheduler::new());
196        let monitoring = std::sync::Arc::new(MonitoringService::new());
197        let executor = std::sync::Arc::new(DefaultTaskExecutor);
198
199        Ok(Self {
200            persistence,
201            scheduler,
202            monitoring,
203            executor,
204        })
205    }
206
207    /// Create an in-memory workflow engine (useful for testing).
208    ///
209    /// # Errors
210    ///
211    /// Returns an error if the in-memory database cannot be initialized.
212    pub fn in_memory() -> Result<Self> {
213        let persistence = std::sync::Arc::new(PersistenceManager::in_memory()?);
214        let scheduler = std::sync::Arc::new(WorkflowScheduler::new());
215        let monitoring = std::sync::Arc::new(MonitoringService::new());
216        let executor = std::sync::Arc::new(DefaultTaskExecutor);
217
218        Ok(Self {
219            persistence,
220            scheduler,
221            monitoring,
222            executor,
223        })
224    }
225
226    /// Get persistence manager.
227    #[must_use]
228    pub fn persistence(&self) -> &std::sync::Arc<PersistenceManager> {
229        &self.persistence
230    }
231
232    /// Get scheduler.
233    #[must_use]
234    pub fn scheduler(&self) -> &std::sync::Arc<WorkflowScheduler> {
235        &self.scheduler
236    }
237
238    /// Get monitoring service.
239    #[must_use]
240    pub fn monitoring(&self) -> &std::sync::Arc<MonitoringService> {
241        &self.monitoring
242    }
243
244    /// Get task executor.
245    #[must_use]
246    pub fn executor(&self) -> &std::sync::Arc<dyn TaskExecutor> {
247        &self.executor
248    }
249
250    /// Submit a workflow for execution.
251    ///
252    /// # Errors
253    ///
254    /// Returns an error if the workflow cannot be saved or executed.
255    pub async fn submit_workflow(&self, workflow: &Workflow) -> Result<WorkflowId> {
256        self.persistence.save_workflow(workflow)?;
257        Ok(workflow.id)
258    }
259
260    /// Execute a workflow immediately.
261    ///
262    /// # Errors
263    ///
264    /// Returns an error if the workflow cannot be loaded or executed.
265    pub async fn execute_workflow(&self, workflow_id: WorkflowId) -> Result<()> {
266        let mut workflow = self.persistence.load_workflow(workflow_id)?;
267
268        let executor = WorkflowExecutor::new(self.executor.clone());
269
270        self.monitoring
271            .start_workflow(workflow_id, workflow.name.clone(), workflow.tasks.len());
272
273        let result = executor.execute(&mut workflow).await?;
274
275        self.monitoring
276            .complete_workflow(workflow_id, result.state == WorkflowState::Completed);
277
278        self.persistence.save_workflow(&workflow)?;
279
280        Ok(())
281    }
282
283    /// Schedule a workflow with a trigger.
284    ///
285    /// # Errors
286    ///
287    /// Returns an error if the workflow cannot be scheduled.
288    pub async fn schedule_workflow(
289        &self,
290        workflow: Workflow,
291        trigger: Trigger,
292    ) -> Result<WorkflowId> {
293        let workflow_id = workflow.id;
294        self.persistence.save_workflow(&workflow)?;
295        self.scheduler.add_schedule(workflow, trigger).await?;
296        Ok(workflow_id)
297    }
298
299    /// Start the workflow engine (scheduler and monitoring).
300    ///
301    /// # Errors
302    ///
303    /// Returns an error if the scheduler cannot be started.
304    pub async fn start(&self) -> Result<()> {
305        self.scheduler.start().await?;
306        tracing::info!("Workflow engine started");
307        Ok(())
308    }
309
310    /// Stop the workflow engine.
311    ///
312    /// # Errors
313    ///
314    /// Returns an error if the scheduler cannot be stopped.
315    pub async fn stop(&self) -> Result<()> {
316        self.scheduler.stop().await?;
317        tracing::info!("Workflow engine stopped");
318        Ok(())
319    }
320
321    /// Process scheduled workflows (should be called periodically).
322    pub async fn process_schedules(&self) -> Result<Vec<WorkflowId>> {
323        let ready_workflows = self.scheduler.check_schedules().await;
324        let mut executed = Vec::new();
325
326        for workflow in ready_workflows {
327            let workflow_id = workflow.id;
328            self.persistence.save_workflow(&workflow)?;
329
330            // Execute in background
331            let engine = Self {
332                persistence: self.persistence.clone(),
333                scheduler: self.scheduler.clone(),
334                monitoring: self.monitoring.clone(),
335                executor: self.executor.clone(),
336            };
337
338            tokio::spawn(async move {
339                if let Err(e) = engine.execute_workflow(workflow_id).await {
340                    tracing::error!("Scheduled workflow execution failed: {}", e);
341                }
342            });
343
344            executed.push(workflow_id);
345        }
346
347        Ok(executed)
348    }
349
350    /// Create API state for REST API server.
351    #[must_use]
352    pub fn api_state(&self) -> api::ApiState {
353        api::ApiState {
354            persistence: self.persistence.clone(),
355            scheduler: self.scheduler.clone(),
356            monitoring: self.monitoring.clone(),
357            executor: self.executor.clone(),
358            active_workflows: std::sync::Arc::new(tokio::sync::RwLock::new(
359                std::collections::HashMap::new(),
360            )),
361        }
362    }
363
364    /// Create API router for REST API server.
365    pub fn api_router(&self) -> axum::Router {
366        api::create_router(self.api_state())
367    }
368}
369
370#[cfg(all(test, feature = "sqlite"))]
371mod tests {
372    use super::*;
373    use std::time::Duration;
374
375    #[tokio::test]
376    async fn test_workflow_engine_creation() {
377        let engine = WorkflowEngine::in_memory();
378        assert!(engine.is_ok());
379    }
380
381    #[tokio::test]
382    async fn test_submit_workflow() {
383        let engine = WorkflowEngine::in_memory().expect("should succeed in test");
384        let workflow = Workflow::new("test-workflow");
385
386        let workflow_id = engine
387            .submit_workflow(&workflow)
388            .await
389            .expect("should succeed in test");
390        assert_eq!(workflow_id, workflow.id);
391    }
392
393    #[tokio::test]
394    async fn test_execute_workflow() {
395        let engine = WorkflowEngine::in_memory().expect("should succeed in test");
396        let mut workflow = Workflow::new("test-workflow");
397
398        let task = Task::new(
399            "wait-task",
400            TaskType::Wait {
401                duration: Duration::from_millis(10),
402            },
403        );
404        workflow.add_task(task);
405
406        let workflow_id = engine
407            .submit_workflow(&workflow)
408            .await
409            .expect("should succeed in test");
410        let result = engine.execute_workflow(workflow_id).await;
411        assert!(result.is_ok());
412    }
413
414    #[tokio::test]
415    async fn test_schedule_workflow() {
416        let engine = WorkflowEngine::in_memory().expect("should succeed in test");
417        let workflow = Workflow::new("test-workflow");
418        let trigger = Trigger::Manual;
419
420        let workflow_id = engine
421            .schedule_workflow(workflow, trigger)
422            .await
423            .expect("should succeed in test");
424
425        let schedules = engine.scheduler.list_schedules().await;
426        assert_eq!(schedules.len(), 1);
427        assert_eq!(schedules[0].0, workflow_id);
428    }
429
430    #[tokio::test]
431    async fn test_engine_start_stop() {
432        let engine = WorkflowEngine::in_memory().expect("should succeed in test");
433
434        engine.start().await.expect("should succeed in test");
435        engine.stop().await.expect("should succeed in test");
436    }
437}
oximedia_workflow/lib.rs

oximedia_workflow/
lib.rs
