ubiquity_core/
command_unified.rs

1//! Unified command executor that provides parity across all platforms
2
3use async_trait::async_trait;
4use futures::Stream;
5use std::pin::Pin;
6use std::sync::Arc;
7use tracing::{debug, info};
8use uuid::Uuid;
9
10use crate::command::{CommandEvent, CommandExecutor, CommandRequest, CommandResult};
11use crate::error::UbiquityError;
12
13#[cfg(not(target_arch = "wasm32"))]
14use crate::command_local::LocalCommandExecutor;
15
16#[cfg(target_arch = "wasm32")]
17use crate::command_wasm::WasmCommandExecutor;
18
19use crate::command_cloud::CloudCommandExecutor;
20
21/// Execution mode for the unified executor
22#[derive(Debug, Clone, Copy, PartialEq, Eq)]
23pub enum ExecutionMode {
24    /// Use local process execution (not available in WASM)
25    Local,
26    /// Use WebAssembly Workers (browser only)
27    Wasm,
28    /// Use Cloudflare Workers Durable Objects
29    Cloud,
30    /// Automatically detect the best mode
31    Auto,
32}
33
34/// Configuration for unified command executor
35#[derive(Debug, Clone)]
36pub struct UnifiedExecutorConfig {
37    pub mode: ExecutionMode,
38    pub cloud_worker_url: Option<String>,
39    pub cloud_api_token: Option<String>,
40    pub cloud_namespace_id: Option<String>,
41    pub event_buffer_size: usize,
42    pub max_wasm_workers: usize,
43}
44
45impl Default for UnifiedExecutorConfig {
46    fn default() -> Self {
47        Self {
48            mode: ExecutionMode::Auto,
49            cloud_worker_url: None,
50            cloud_api_token: None,
51            cloud_namespace_id: None,
52            event_buffer_size: 1024,
53            max_wasm_workers: 4,
54        }
55    }
56}
57
58/// Unified command executor that provides parity across platforms
59pub struct UnifiedCommandExecutor {
60    config: UnifiedExecutorConfig,
61    executor: Arc<dyn CommandExecutor>,
62}
63
64impl UnifiedCommandExecutor {
65    /// Create a new unified executor with the given configuration
66    pub fn new(config: UnifiedExecutorConfig) -> Result<Self, UbiquityError> {
67        let executor = Self::create_executor(&config)?;
68        
69        Ok(Self {
70            config,
71            executor: Arc::from(executor),
72        })
73    }
74    
75    /// Create a new unified executor with automatic mode detection
76    pub fn auto() -> Result<Self, UbiquityError> {
77        Self::new(UnifiedExecutorConfig::default())
78    }
79    
80    /// Get the current execution mode
81    pub fn mode(&self) -> ExecutionMode {
82        if self.config.mode == ExecutionMode::Auto {
83            Self::detect_mode()
84        } else {
85            self.config.mode
86        }
87    }
88    
89    /// Detect the best execution mode for the current platform
90    fn detect_mode() -> ExecutionMode {
91        #[cfg(target_arch = "wasm32")]
92        {
93            ExecutionMode::Wasm
94        }
95        
96        #[cfg(not(target_arch = "wasm32"))]
97        {
98            ExecutionMode::Local
99        }
100    }
101    
102    /// Create the appropriate executor based on configuration
103    fn create_executor(config: &UnifiedExecutorConfig) -> Result<Box<dyn CommandExecutor>, UbiquityError> {
104        let mode = if config.mode == ExecutionMode::Auto {
105            Self::detect_mode()
106        } else {
107            config.mode
108        };
109        
110        info!("Creating command executor in {:?} mode", mode);
111        
112        match mode {
113            #[cfg(not(target_arch = "wasm32"))]
114            ExecutionMode::Local => {
115                let executor = LocalCommandExecutor::new()
116                    .with_event_buffer_size(config.event_buffer_size);
117                Ok(Box::new(executor))
118            }
119            
120            #[cfg(target_arch = "wasm32")]
121            ExecutionMode::Wasm => {
122                let executor = WasmCommandExecutor::new()?
123                    .with_max_workers(config.max_wasm_workers)?;
124                Ok(Box::new(executor))
125            }
126            
127            ExecutionMode::Cloud => {
128                let worker_url = config.cloud_worker_url.as_ref()
129                    .ok_or_else(|| UbiquityError::Configuration(
130                        "Cloud worker URL not configured".to_string()
131                    ))?;
132                
133                let api_token = config.cloud_api_token.as_ref()
134                    .ok_or_else(|| UbiquityError::Configuration(
135                        "Cloud API token not configured".to_string()
136                    ))?;
137                
138                let namespace_id = config.cloud_namespace_id.as_ref()
139                    .ok_or_else(|| UbiquityError::Configuration(
140                        "Cloud namespace ID not configured".to_string()
141                    ))?;
142                
143                let executor = CloudCommandExecutor::new(
144                    worker_url.clone(),
145                    api_token.clone(),
146                    namespace_id.clone(),
147                );
148                
149                Ok(Box::new(executor))
150            }
151            
152            _ => Err(UbiquityError::Configuration(format!(
153                "Execution mode {:?} not available on this platform",
154                mode
155            ))),
156        }
157    }
158    
159    /// Switch to a different execution mode
160    pub fn switch_mode(&mut self, mode: ExecutionMode) -> Result<(), UbiquityError> {
161        self.config.mode = mode;
162        self.executor = Arc::from(Self::create_executor(&self.config)?);
163        Ok(())
164    }
165    
166    /// Create a command builder
167    pub fn command(&self, command: impl Into<String>) -> CommandBuilder {
168        CommandBuilder::new(self, command)
169    }
170}
171
172#[async_trait]
173impl CommandExecutor for UnifiedCommandExecutor {
174    async fn execute(
175        &self,
176        request: CommandRequest,
177    ) -> Result<Pin<Box<dyn Stream<Item = CommandEvent> + Send>>, UbiquityError> {
178        debug!("Executing command {} in {:?} mode", request.command, self.mode());
179        self.executor.execute(request).await
180    }
181
182    async fn cancel(&self, command_id: Uuid) -> Result<(), UbiquityError> {
183        debug!("Cancelling command {} in {:?} mode", command_id, self.mode());
184        self.executor.cancel(command_id).await
185    }
186
187    async fn status(&self, command_id: Uuid) -> Result<Option<CommandResult>, UbiquityError> {
188        self.executor.status(command_id).await
189    }
190}
191
192/// Builder for creating and executing commands
193pub struct CommandBuilder<'a> {
194    executor: &'a UnifiedCommandExecutor,
195    request: CommandRequest,
196}
197
198impl<'a> CommandBuilder<'a> {
199    fn new(executor: &'a UnifiedCommandExecutor, command: impl Into<String>) -> Self {
200        Self {
201            executor,
202            request: CommandRequest::new(command),
203        }
204    }
205    
206    pub fn arg(mut self, arg: impl Into<String>) -> Self {
207        self.request.args.push(arg.into());
208        self
209    }
210    
211    pub fn args<I, S>(mut self, args: I) -> Self
212    where
213        I: IntoIterator<Item = S>,
214        S: Into<String>,
215    {
216        self.request.args.extend(args.into_iter().map(Into::into));
217        self
218    }
219    
220    pub fn env(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
221        self.request.env.insert(key.into(), value.into());
222        self
223    }
224    
225    pub fn envs<I, K, V>(mut self, vars: I) -> Self
226    where
227        I: IntoIterator<Item = (K, V)>,
228        K: Into<String>,
229        V: Into<String>,
230    {
231        for (k, v) in vars {
232            self.request.env.insert(k.into(), v.into());
233        }
234        self
235    }
236    
237    pub fn current_dir(mut self, dir: impl Into<String>) -> Self {
238        self.request.working_dir = Some(dir.into());
239        self
240    }
241    
242    pub fn timeout(mut self, timeout: std::time::Duration) -> Self {
243        self.request.timeout = Some(timeout);
244        self
245    }
246    
247    pub fn stdin(mut self, stdin: impl Into<String>) -> Self {
248        self.request.stdin = Some(stdin.into());
249        self
250    }
251    
252    pub async fn execute(self) -> Result<Pin<Box<dyn Stream<Item = CommandEvent> + Send>>, UbiquityError> {
253        self.executor.execute(self.request).await
254    }
255    
256    pub async fn output(self) -> Result<CommandOutput, UbiquityError> {
257        use futures::StreamExt;
258        
259        let request_id = self.request.id;
260        let mut stream = self.execute().await?;
261        let mut output = CommandOutput::new(request_id);
262        
263        while let Some(event) = stream.next().await {
264            output.process_event(event);
265        }
266        
267        Ok(output)
268    }
269}
270
271/// Collected output from a command execution
272#[derive(Debug, Clone)]
273pub struct CommandOutput {
274    pub id: Uuid,
275    pub stdout: String,
276    pub stderr: String,
277    pub exit_code: Option<i32>,
278    pub success: bool,
279    pub duration_ms: Option<u64>,
280    pub cancelled: bool,
281}
282
283impl CommandOutput {
284    fn new(id: Uuid) -> Self {
285        Self {
286            id,
287            stdout: String::new(),
288            stderr: String::new(),
289            exit_code: None,
290            success: false,
291            duration_ms: None,
292            cancelled: false,
293        }
294    }
295    
296    fn process_event(&mut self, event: CommandEvent) {
297        match event {
298            CommandEvent::Stdout { data, .. } => {
299                if !self.stdout.is_empty() {
300                    self.stdout.push('\n');
301                }
302                self.stdout.push_str(&data);
303            }
304            CommandEvent::Stderr { data, .. } => {
305                if !self.stderr.is_empty() {
306                    self.stderr.push('\n');
307                }
308                self.stderr.push_str(&data);
309            }
310            CommandEvent::Completed { exit_code, duration_ms, .. } => {
311                self.exit_code = Some(exit_code);
312                self.success = exit_code == 0;
313                self.duration_ms = Some(duration_ms);
314            }
315            CommandEvent::Failed { duration_ms, .. } => {
316                self.success = false;
317                self.duration_ms = Some(duration_ms);
318            }
319            CommandEvent::Cancelled { duration_ms, .. } => {
320                self.cancelled = true;
321                self.duration_ms = Some(duration_ms);
322            }
323            _ => {}
324        }
325    }
326}
327
328#[cfg(test)]
329mod tests {
330    use super::*;
331    use futures::StreamExt;
332
333    #[tokio::test]
334    async fn test_unified_executor_auto_mode() {
335        let executor = UnifiedCommandExecutor::auto().unwrap();
336        
337        #[cfg(not(target_arch = "wasm32"))]
338        assert_eq!(executor.mode(), ExecutionMode::Local);
339        
340        #[cfg(target_arch = "wasm32")]
341        assert_eq!(executor.mode(), ExecutionMode::Wasm);
342    }
343
344    #[tokio::test]
345    async fn test_command_builder() {
346        let executor = UnifiedCommandExecutor::auto().unwrap();
347        
348        let output = executor
349            .command("echo")
350            .arg("hello")
351            .arg("world")
352            .env("TEST", "value")
353            .timeout(std::time::Duration::from_secs(5))
354            .output()
355            .await
356            .unwrap();
357        
358        assert!(output.success);
359        assert_eq!(output.stdout.trim(), "hello world");
360    }
361
362    #[tokio::test]
363    async fn test_command_streaming() {
364        let executor = UnifiedCommandExecutor::auto().unwrap();
365        
366        let mut stream = executor
367            .command("echo")
368            .args(["line1", "line2"])
369            .execute()
370            .await
371            .unwrap();
372        
373        let mut events = Vec::new();
374        while let Some(event) = stream.next().await {
375            events.push(event);
376        }
377        
378        assert!(!events.is_empty());
379        assert!(events.iter().any(|e| matches!(e, CommandEvent::Started { .. })));
380    }
381}
ubiquity_core/command_unified.rs

ubiquity_core/
command_unified.rs
