phoenix_cli/
dev.rs

1use std::{
2    ffi::OsString,
3    future::Future,
4    path::{Path, PathBuf},
5    process::{ExitStatus, Stdio},
6    sync::mpsc,
7    time::Duration,
8};
9
10use notify::{EventKind, RecommendedWatcher, RecursiveMode, Watcher};
11use thiserror::Error;
12use tokio::process::{Child, Command};
13use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender, unbounded_channel};
14
15#[derive(Clone, Debug, Eq, PartialEq)]
16pub struct CommandSpec {
17    pub program: PathBuf,
18    pub args: Vec<OsString>,
19}
20
21impl CommandSpec {
22    #[must_use]
23    pub fn new(program: impl Into<PathBuf>) -> Self {
24        Self {
25            program: program.into(),
26            args: Vec::new(),
27        }
28    }
29
30    #[must_use]
31    pub fn arg(mut self, argument: impl Into<OsString>) -> Self {
32        self.args.push(argument.into());
33        self
34    }
35
36    #[must_use]
37    pub fn args(mut self, arguments: impl IntoIterator<Item = impl Into<OsString>>) -> Self {
38        self.args.extend(arguments.into_iter().map(Into::into));
39        self
40    }
41}
42
43#[derive(Clone, Debug)]
44pub struct DevConfig {
45    pub working_directory: PathBuf,
46    pub rust: CommandSpec,
47    pub vite: CommandSpec,
48    /// When true (default), watch Rust sources and restart `cargo run` on change.
49    pub watch_rust: bool,
50}
51
52impl Default for DevConfig {
53    fn default() -> Self {
54        Self {
55            working_directory: PathBuf::from("."),
56            rust: CommandSpec::new("cargo").args(["run", "--", "serve"]),
57            vite: CommandSpec::new("npm").args(["run", "dev", "--", "--strictPort"]),
58            watch_rust: true,
59        }
60    }
61}
62
63impl DevConfig {
64    #[must_use]
65    pub fn working_directory(mut self, directory: impl Into<PathBuf>) -> Self {
66        self.working_directory = directory.into();
67        self
68    }
69
70    #[must_use]
71    pub fn rust(mut self, command: CommandSpec) -> Self {
72        self.rust = command;
73        self
74    }
75
76    #[must_use]
77    pub fn vite(mut self, command: CommandSpec) -> Self {
78        self.vite = command;
79        self
80    }
81
82    #[must_use]
83    pub const fn watch_rust(mut self, enabled: bool) -> Self {
84        self.watch_rust = enabled;
85        self
86    }
87}
88
89pub struct DevSupervisor {
90    config: DevConfig,
91}
92
93impl DevSupervisor {
94    #[must_use]
95    pub fn new(config: DevConfig) -> Self {
96        Self { config }
97    }
98
99    /// Run Rust and Vite together until Ctrl-C, or until Vite exits on its own.
100    /// When `watch_rust` is enabled, Rust source changes restart the backend;
101    /// compile/run failures wait for the next change instead of tearing down Vite.
102    ///
103    /// # Errors
104    ///
105    /// Returns an error when a child cannot start, signal registration fails,
106    /// or the Vite process exits on its own.
107    pub async fn run(self) -> Result<(), DevError> {
108        self.run_with_shutdown(async { tokio::signal::ctrl_c().await.map_err(DevError::Signal) })
109            .await
110    }
111
112    /// Run with a caller-provided shutdown future, primarily for integration
113    /// with another application lifecycle or deterministic tests.
114    ///
115    /// # Errors
116    ///
117    /// Returns a spawn, wait, shutdown, watch, or early-child-exit error.
118    pub async fn run_with_shutdown<F>(self, shutdown: F) -> Result<(), DevError>
119    where
120        F: Future<Output = Result<(), DevError>> + Send,
121    {
122        let cwd = self.config.working_directory.clone();
123        let mut vite = spawn("Vite", &self.config.vite, &cwd)?;
124        let mut changes = if self.config.watch_rust {
125            Some(start_rust_watcher(&cwd)?)
126        } else {
127            None
128        };
129        tokio::pin!(shutdown);
130
131        let result = async {
132            loop {
133                let mut rust = spawn("Rust", &self.config.rust, &cwd)?;
134                if self.config.watch_rust {
135                    eprintln!(
136                        "px dev: watching Rust (app/, src/, routes/, config/, database/, Cargo.toml)"
137                    );
138                }
139
140                let event = tokio::select! {
141                    result = rust.wait() => Event::Rust(result.map_err(DevError::Wait)?),
142                    result = vite.wait() => Event::Vite(result.map_err(DevError::Wait)?),
143                    result = &mut shutdown => Event::Shutdown(result),
144                    changed = recv_change(&mut changes) => {
145                        changed?;
146                        Event::RustChanged
147                    }
148                };
149
150                match event {
151                    Event::Shutdown(result) => {
152                        terminate(&mut rust).await?;
153                        return result;
154                    }
155                    Event::Vite(status) => {
156                        terminate(&mut rust).await?;
157                        return Err(DevError::Exited {
158                            process: "Vite",
159                            status,
160                        });
161                    }
162                    Event::RustChanged => {
163                        eprintln!("px dev: Rust source changed — rebuilding…");
164                        terminate(&mut rust).await?;
165                        drain_changes(&mut changes, Duration::from_millis(400)).await;
166                        continue;
167                    }
168                    Event::Rust(status) => {
169                        if !self.config.watch_rust {
170                            return Err(DevError::Exited {
171                                process: "Rust",
172                                status,
173                            });
174                        }
175                        eprintln!(
176                            "px dev: Rust process exited ({status}); waiting for source changes…"
177                        );
178                        let wait = tokio::select! {
179                            result = vite.wait() => WaitWhileDown::Vite(result.map_err(DevError::Wait)?),
180                            result = &mut shutdown => WaitWhileDown::Shutdown(result),
181                            changed = recv_change(&mut changes) => {
182                                changed?;
183                                WaitWhileDown::Changed
184                            }
185                        };
186                        match wait {
187                            WaitWhileDown::Shutdown(result) => return result,
188                            WaitWhileDown::Vite(status) => {
189                                return Err(DevError::Exited {
190                                    process: "Vite",
191                                    status,
192                                });
193                            }
194                            WaitWhileDown::Changed => {
195                                drain_changes(&mut changes, Duration::from_millis(400)).await;
196                                continue;
197                            }
198                        }
199                    }
200                }
201            }
202        }
203        .await;
204
205        terminate(&mut vite).await?;
206        result
207    }
208}
209
210enum Event {
211    Rust(ExitStatus),
212    Vite(ExitStatus),
213    Shutdown(Result<(), DevError>),
214    RustChanged,
215}
216
217enum WaitWhileDown {
218    Vite(ExitStatus),
219    Shutdown(Result<(), DevError>),
220    Changed,
221}
222
223struct RustWatcher {
224    _watcher: RecommendedWatcher,
225    rx: UnboundedReceiver<()>,
226}
227
228fn start_rust_watcher(cwd: &Path) -> Result<RustWatcher, DevError> {
229    let (tx, rx) = unbounded_channel();
230    let (notify_tx, notify_rx) = mpsc::channel();
231    let mut watcher = notify::recommended_watcher(move |result| {
232        let _ = notify_tx.send(result);
233    })
234    .map_err(DevError::Watch)?;
235
236    for relative in ["app", "src", "routes", "config", "database"] {
237        let path = cwd.join(relative);
238        if path.is_dir() {
239            watcher
240                .watch(&path, RecursiveMode::Recursive)
241                .map_err(DevError::Watch)?;
242        }
243    }
244    let cargo_toml = cwd.join("Cargo.toml");
245    if cargo_toml.is_file() {
246        watcher
247            .watch(&cargo_toml, RecursiveMode::NonRecursive)
248            .map_err(DevError::Watch)?;
249    }
250
251    let cwd = cwd.to_path_buf();
252    std::thread::Builder::new()
253        .name("px-dev-rust-watch".into())
254        .spawn(move || watch_loop(cwd, notify_rx, tx))
255        .map_err(|source| DevError::Spawn {
256            process: "RustWatcher",
257            source,
258        })?;
259
260    Ok(RustWatcher {
261        _watcher: watcher,
262        rx,
263    })
264}
265
266fn watch_loop(
267    cwd: PathBuf,
268    notify_rx: mpsc::Receiver<Result<notify::Event, notify::Error>>,
269    tx: UnboundedSender<()>,
270) {
271    while let Ok(result) = notify_rx.recv() {
272        let Ok(event) = result else {
273            continue;
274        };
275        if is_relevant_event(&cwd, &event) {
276            let _ = tx.send(());
277        }
278    }
279}
280
281fn is_relevant_event(cwd: &Path, event: &notify::Event) -> bool {
282    match event.kind {
283        EventKind::Modify(_) | EventKind::Create(_) | EventKind::Remove(_) => {}
284        _ => return false,
285    }
286    event.paths.iter().any(|path| is_watched_rust_path(cwd, path))
287}
288
289fn is_watched_rust_path(cwd: &Path, path: &Path) -> bool {
290    let Ok(relative) = path.strip_prefix(cwd) else {
291        return false;
292    };
293    let mut components = relative.components();
294    let Some(std::path::Component::Normal(first)) = components.next() else {
295        return path.file_name().is_some_and(|name| name == "Cargo.toml");
296    };
297    let first = first.to_string_lossy();
298    matches!(
299        first.as_ref(),
300        "app" | "src" | "routes" | "config" | "database" | "Cargo.toml"
301    ) && !relative
302        .components()
303        .any(|component| matches!(component, std::path::Component::Normal(name) if name == "target"))
304}
305
306async fn recv_change(changes: &mut Option<RustWatcher>) -> Result<(), DevError> {
307    match changes.as_mut() {
308        Some(watcher) => watcher.rx.recv().await.ok_or(DevError::WatchClosed),
309        None => std::future::pending().await,
310    }
311}
312
313async fn drain_changes(changes: &mut Option<RustWatcher>, window: Duration) {
314    let Some(watcher) = changes.as_mut() else {
315        return;
316    };
317    let deadline = tokio::time::Instant::now() + window;
318    loop {
319        let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
320        if remaining.is_zero() {
321            break;
322        }
323        match tokio::time::timeout(remaining, watcher.rx.recv()).await {
324            Ok(Some(())) => continue,
325            _ => break,
326        }
327    }
328}
329
330fn spawn(label: &'static str, spec: &CommandSpec, cwd: &Path) -> Result<Child, DevError> {
331    let mut command = Command::new(&spec.program);
332    command
333        .args(&spec.args)
334        .current_dir(cwd)
335        .stdin(Stdio::inherit())
336        .stdout(Stdio::inherit())
337        .stderr(Stdio::inherit())
338        .kill_on_drop(true);
339    #[cfg(unix)]
340    command.process_group(0);
341    command.spawn().map_err(|source| DevError::Spawn {
342        process: label,
343        source,
344    })
345}
346
347#[cfg(unix)]
348async fn terminate(child: &mut Child) -> Result<(), DevError> {
349    use nix::{
350        sys::signal::{Signal, killpg},
351        unistd::Pid,
352    };
353
354    let Some(id) = child.id() else {
355        let _ = child.wait().await.map_err(DevError::Wait)?;
356        return Ok(());
357    };
358    let process_group = Pid::from_raw(i32::try_from(id).map_err(|_| DevError::InvalidProcessId)?);
359    if let Err(error) = killpg(process_group, Signal::SIGTERM)
360        && error != nix::errno::Errno::ESRCH
361    {
362        return Err(DevError::SignalProcess(error));
363    }
364    if tokio::time::timeout(Duration::from_secs(3), child.wait())
365        .await
366        .is_err()
367    {
368        if let Err(error) = killpg(process_group, Signal::SIGKILL)
369            && error != nix::errno::Errno::ESRCH
370        {
371            return Err(DevError::SignalProcess(error));
372        }
373        let _ = child.wait().await.map_err(DevError::Wait)?;
374    }
375    Ok(())
376}
377
378#[cfg(not(unix))]
379async fn terminate(child: &mut Child) -> Result<(), DevError> {
380    if child.id().is_some() {
381        child.start_kill().map_err(DevError::Shutdown)?;
382    }
383    let _ = child.wait().await.map_err(DevError::Wait)?;
384    Ok(())
385}
386
387#[derive(Debug, Error)]
388pub enum DevError {
389    #[error("failed to start the {process} development process: {source}")]
390    Spawn {
391        process: &'static str,
392        source: std::io::Error,
393    },
394    #[error("failed while waiting for a development process: {0}")]
395    Wait(std::io::Error),
396    #[error("failed to stop a development process: {0}")]
397    Shutdown(std::io::Error),
398    #[cfg(unix)]
399    #[error("failed to signal a development process group: {0}")]
400    SignalProcess(nix::errno::Errno),
401    #[error("a development process returned an invalid process id")]
402    InvalidProcessId,
403    #[error("failed to listen for Ctrl-C: {0}")]
404    Signal(std::io::Error),
405    #[error("failed to watch Rust sources for reload: {0}")]
406    Watch(#[from] notify::Error),
407    #[error("Rust file watcher stopped unexpectedly")]
408    WatchClosed,
409    #[error("the {process} development process exited early with {status}")]
410    Exited {
411        process: &'static str,
412        status: ExitStatus,
413    },
414}
415
416#[cfg(test)]
417mod tests {
418    use super::*;
419    use std::time::Duration;
420
421    fn shell(script: &str) -> CommandSpec {
422        CommandSpec::new("sh").args(["-c", script])
423    }
424
425    #[test]
426    fn default_dev_config_runs_serve_with_watch() {
427        let config = DevConfig::default();
428        assert_eq!(config.rust.program, PathBuf::from("cargo"));
429        assert!(config.watch_rust);
430        assert_eq!(
431            config.rust.args,
432            vec![
433                OsString::from("run"),
434                OsString::from("--"),
435                OsString::from("serve")
436            ]
437        );
438    }
439
440    #[test]
441    fn watched_paths_cover_app_sources_not_views() {
442        let cwd = PathBuf::from("/app");
443        assert!(is_watched_rust_path(
444            &cwd,
445            &cwd.join("app/controllers/home.rs")
446        ));
447        assert!(is_watched_rust_path(&cwd, &cwd.join("Cargo.toml")));
448        assert!(!is_watched_rust_path(
449            &cwd,
450            &cwd.join("views/pages/home.tsx")
451        ));
452        assert!(!is_watched_rust_path(
453            &cwd,
454            &cwd.join("target/debug/my-app")
455        ));
456    }
457
458    #[tokio::test]
459    async fn shutdown_stops_and_reaps_both_processes() {
460        let supervisor = DevSupervisor::new(
461            DevConfig::default()
462                .watch_rust(false)
463                .rust(shell("sleep 10 & wait"))
464                .vite(shell("sleep 10 & wait")),
465        );
466        supervisor
467            .run_with_shutdown(async {
468                tokio::time::sleep(Duration::from_millis(50)).await;
469                Ok(())
470            })
471            .await
472            .unwrap();
473    }
474
475    #[tokio::test]
476    async fn one_failed_process_stops_the_other_without_watch() {
477        let supervisor = DevSupervisor::new(
478            DevConfig::default()
479                .watch_rust(false)
480                .rust(shell("exit 7"))
481                .vite(shell("sleep 10")),
482        );
483        let result = supervisor.run_with_shutdown(std::future::pending()).await;
484
485        assert!(matches!(
486            result,
487            Err(DevError::Exited {
488                process: "Rust",
489                status,
490            }) if status.code() == Some(7)
491        ));
492    }
493
494    #[tokio::test]
495    async fn rust_exit_with_watch_waits_for_shutdown_without_failing() {
496        let supervisor = DevSupervisor::new(
497            DevConfig::default()
498                .watch_rust(true)
499                .rust(shell("exit 1"))
500                .vite(shell("sleep 10 & wait")),
501        );
502        supervisor
503            .run_with_shutdown(async {
504                tokio::time::sleep(Duration::from_millis(80)).await;
505                Ok(())
506            })
507            .await
508            .unwrap();
509    }
510}
phoenix_cli/dev.rs

phoenix_cli/
dev.rs
