diff --git a/Cargo.lock b/Cargo.lock index a56b7d7..016eccc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1162,6 +1162,7 @@ dependencies = [ "directories", "doctate-common", "eframe", + "libc", "reqwest", "serde", "serde_json", diff --git a/client-desktop/Cargo.toml b/client-desktop/Cargo.toml index aad56a2..c1fdfbc 100644 --- a/client-desktop/Cargo.toml +++ b/client-desktop/Cargo.toml @@ -19,6 +19,9 @@ directories = "5" thiserror = "1" which = "6" +[target.'cfg(unix)'.dependencies] +libc = "0.2" + [dev-dependencies] tempfile = "3" tokio = { workspace = true, features = ["test-util"] } diff --git a/client-desktop/src/app.rs b/client-desktop/src/app.rs new file mode 100644 index 0000000..a453740 --- /dev/null +++ b/client-desktop/src/app.rs @@ -0,0 +1,381 @@ +//! The egui application: config panel, record/stop buttons, state +//! transitions driven by recorder + uploader channels. + +use std::path::PathBuf; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use doctate_common::timestamp::{now_rfc3339, recorded_at_to_filename_stem}; +use eframe::egui; +use tokio::runtime::Runtime; +use tokio::sync::mpsc; +use tracing::{error, info, warn}; +use uuid::Uuid; + +use crate::config::Config; +use crate::recorder::{Recorder, RecorderEvent}; +use crate::state::AppState; +use crate::uploader::{PendingUpload, UploadEvent, Uploader, write_sidecar}; + +pub struct DoctateApp { + runtime: Arc, + config: Option, + pending_dir: PathBuf, + state: AppState, + + server_url_input: String, + api_key_input: String, + + recorder: Option, + recorder_events: Option>, + + uploader: Option, + upload_events: Option>, +} + +enum UiAction { + SaveConfig, + StartNew, + Stop, + DismissError, +} + +impl DoctateApp { + pub fn new(runtime: Arc, pending_dir: PathBuf) -> Self { + let config = match Config::load() { + Ok(Some(c)) => Some(c), + Ok(None) => None, + Err(e) => { + warn!(error = %e, "config load failed"); + None + } + }; + + let server_url_input = config + .as_ref() + .map(|c| c.server_url.clone()) + .unwrap_or_default(); + let api_key_input = config + .as_ref() + .map(|c| c.api_key.clone()) + .unwrap_or_default(); + + let (state, uploader, upload_events) = if let Some(cfg) = &config { + let (up, rx) = spawn_uploader(&runtime, cfg.clone(), pending_dir.clone()); + (AppState::Idle, Some(up), Some(rx)) + } else { + (AppState::NotConfigured, None, None) + }; + + Self { + runtime, + config, + pending_dir, + state, + server_url_input, + api_key_input, + recorder: None, + recorder_events: None, + uploader, + upload_events, + } + } + + fn on_save_config(&mut self) { + let cfg = Config { + server_url: self.server_url_input.trim().to_string(), + api_key: self.api_key_input.trim().to_string(), + }; + match cfg.save() { + Ok(()) => { + info!("config saved"); + let (up, rx) = spawn_uploader(&self.runtime, cfg.clone(), self.pending_dir.clone()); + self.uploader = Some(up); + self.upload_events = Some(rx); + self.config = Some(cfg); + self.state = AppState::Idle; + } + Err(e) => { + error!(error = %e, "config save failed"); + self.state = AppState::Error { + message: format!("Speichern: {e}"), + }; + } + } + } + + fn on_start_new(&mut self) { + if let Err(e) = std::fs::create_dir_all(&self.pending_dir) { + self.state = AppState::Error { + message: format!("Pending-Verzeichnis: {e}"), + }; + return; + } + + let case_id = Uuid::new_v4(); + let recorded_at = now_rfc3339(); + let stem = format!("{}_{}", case_id, recorded_at_to_filename_stem(&recorded_at)); + let output_path = self.pending_dir.join(format!("{stem}.m4a")); + + let _guard = self.runtime.enter(); + match Recorder::start(output_path) { + Ok((recorder, events)) => { + self.recorder = Some(recorder); + self.recorder_events = Some(events); + self.state = AppState::Recording { + case_id, + recorded_at, + started_at: Instant::now(), + }; + } + Err(e) => { + self.state = AppState::Error { + message: format!("Aufnahme starten: {e}"), + }; + } + } + } + + fn on_stop(&mut self) { + let AppState::Recording { + case_id, + recorded_at, + .. + } = &self.state + else { + return; + }; + let case_id = *case_id; + let recorded_at = recorded_at.clone(); + if let Some(rec) = self.recorder.take() { + rec.stop(); + } + self.state = AppState::FinalizingRecording { + case_id, + recorded_at, + }; + } + + fn drain_recorder_events(&mut self) { + let Some(rx) = self.recorder_events.as_mut() else { + return; + }; + let mut events = Vec::new(); + while let Ok(event) = rx.try_recv() { + events.push(event); + } + for event in events { + match event { + RecorderEvent::Started => {} + RecorderEvent::Finished { output } => { + self.handle_recording_finished(output); + } + RecorderEvent::Failed { error: err } => { + error!(error = %err, "recorder failed"); + self.state = AppState::Error { + message: format!("Aufnahme: {err}"), + }; + self.recorder_events = None; + self.recorder = None; + } + } + } + } + + fn handle_recording_finished(&mut self, output: PathBuf) { + let (case_id, recorded_at) = match &self.state { + AppState::FinalizingRecording { + case_id, + recorded_at, + } => (*case_id, recorded_at.clone()), + AppState::Recording { + case_id, + recorded_at, + .. + } => (*case_id, recorded_at.clone()), + other => { + warn!(state = ?other, "recorder finished in unexpected state"); + self.recorder_events = None; + self.recorder = None; + return; + } + }; + + let upload = PendingUpload { + case_id, + recorded_at, + file: output, + }; + + if let Err(e) = write_sidecar(&upload) { + self.state = AppState::Error { + message: format!("Sidecar schreiben: {e}"), + }; + return; + } + + let Some(uploader) = &self.uploader else { + self.state = AppState::Error { + message: "Upload-Worker nicht verfügbar".into(), + }; + return; + }; + + if let Err(e) = uploader.submit(upload) { + self.state = AppState::Error { + message: format!("Upload-Submit: {e}"), + }; + return; + } + + self.state = AppState::Uploading { case_id }; + self.recorder_events = None; + self.recorder = None; + } + + fn drain_upload_events(&mut self) { + let Some(rx) = self.upload_events.as_mut() else { + return; + }; + let mut events = Vec::new(); + while let Ok(event) = rx.try_recv() { + events.push(event); + } + for event in events { + match event { + UploadEvent::Started(case_id) => { + info!(case_id = %case_id, "upload attempt started"); + } + UploadEvent::Succeeded { case_id, status } => { + info!(case_id = %case_id, status = ?status, "upload succeeded"); + if let AppState::Uploading { case_id: current } = &self.state + && *current == case_id + { + self.state = AppState::Idle; + } + } + UploadEvent::Failed { + case_id, + reason, + will_retry, + } => { + warn!(case_id = %case_id, reason = %reason, will_retry, "upload failed"); + if !will_retry + && let AppState::Uploading { case_id: current } = &self.state + && *current == case_id + { + self.state = AppState::Error { message: reason }; + } + } + } + } + } + + fn render_body(&mut self, ui: &mut egui::Ui) -> Option { + match &self.state { + AppState::NotConfigured => { + ui.label("Server URL:"); + ui.add( + egui::TextEdit::singleline(&mut self.server_url_input).desired_width(280.0), + ); + ui.add_space(4.0); + ui.label("API Key:"); + ui.add( + egui::TextEdit::singleline(&mut self.api_key_input) + .desired_width(280.0) + .password(true), + ); + ui.add_space(8.0); + if ui.button("Speichern").clicked() { + return Some(UiAction::SaveConfig); + } + None + } + AppState::Idle => { + if ui.button("● Neu").clicked() { + return Some(UiAction::StartNew); + } + None + } + AppState::Recording { started_at, .. } => { + let elapsed = started_at.elapsed(); + ui.label(format!( + "● REC {:02}:{:02}", + elapsed.as_secs() / 60, + elapsed.as_secs() % 60 + )); + ui.add_space(4.0); + if ui.button("■ Stop").clicked() { + return Some(UiAction::Stop); + } + None + } + AppState::FinalizingRecording { .. } => { + ui.spinner(); + ui.label("Speichere Aufnahme…"); + None + } + AppState::Uploading { .. } => { + ui.spinner(); + ui.label("Upload läuft…"); + None + } + AppState::Error { message } => { + ui.colored_label(egui::Color32::RED, message); + ui.add_space(8.0); + if ui.button("OK").clicked() { + return Some(UiAction::DismissError); + } + None + } + } + } + + fn handle_action(&mut self, action: UiAction) { + match action { + UiAction::SaveConfig => self.on_save_config(), + UiAction::StartNew => self.on_start_new(), + UiAction::Stop => self.on_stop(), + UiAction::DismissError => self.state = AppState::Idle, + } + } +} + +fn spawn_uploader( + runtime: &Runtime, + config: Config, + pending_dir: PathBuf, +) -> (Uploader, mpsc::UnboundedReceiver) { + let _guard = runtime.enter(); + Uploader::spawn(config, pending_dir) +} + +impl eframe::App for DoctateApp { + fn update(&mut self, ctx: &egui::Context, _frame: &mut eframe::Frame) { + self.drain_recorder_events(); + self.drain_upload_events(); + + egui::CentralPanel::default().show(ctx, |ui| { + ui.vertical_centered(|ui| { + ui.add_space(8.0); + ui.heading("doctate"); + ui.add_space(4.0); + ui.label(self.state.label()); + ui.add_space(12.0); + + let action = self.render_body(ui); + if let Some(action) = action { + self.handle_action(action); + } + }); + }); + + match self.state { + AppState::Recording { .. } => ctx.request_repaint_after(Duration::from_millis(250)), + AppState::FinalizingRecording { .. } | AppState::Uploading { .. } => { + ctx.request_repaint_after(Duration::from_millis(100)); + } + _ => {} + } + } +} diff --git a/client-desktop/src/main.rs b/client-desktop/src/main.rs index 2fb9d93..c0e622b 100644 --- a/client-desktop/src/main.rs +++ b/client-desktop/src/main.rs @@ -1,28 +1,48 @@ -#[allow(dead_code)] +mod app; mod config; -#[allow(dead_code)] mod paths; -#[allow(dead_code)] mod recorder; -#[allow(dead_code)] +mod state; mod uploader; +use std::sync::Arc; + use eframe::egui; use tracing::info; use tracing_subscriber::EnvFilter; +use crate::app::DoctateApp; +use crate::paths::pending_dir; + fn main() -> eframe::Result { tracing_subscriber::fmt() .with_env_filter( - EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("doctate_desktop=info")), + EnvFilter::try_from_default_env() + .unwrap_or_else(|_| EnvFilter::new("doctate_desktop=info")), ) .init(); info!("starting doctate-desktop"); + let runtime = Arc::new( + tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .enable_all() + .build() + .expect("build tokio runtime"), + ); + + let pending = match pending_dir() { + Ok(p) => p, + Err(e) => { + eprintln!("cannot determine pending directory: {e}"); + std::process::exit(1); + } + }; + let options = eframe::NativeOptions { viewport: egui::ViewportBuilder::default() - .with_inner_size([320.0, 200.0]) + .with_inner_size([320.0, 280.0]) .with_resizable(false), ..Default::default() }; @@ -30,22 +50,9 @@ fn main() -> eframe::Result { eframe::run_native( "doctate", options, - Box::new(|_cc| Ok(Box::::default())), + Box::new(move |_cc| { + let app = DoctateApp::new(runtime, pending); + Ok(Box::new(app) as Box) + }), ) } - -#[derive(Default)] -struct DoctateApp; - -impl eframe::App for DoctateApp { - fn update(&mut self, ctx: &egui::Context, _frame: &mut eframe::Frame) { - egui::CentralPanel::default().show(ctx, |ui| { - ui.vertical_centered(|ui| { - ui.add_space(20.0); - ui.heading("doctate"); - ui.add_space(20.0); - ui.label("Skelett — Buttons kommen in Schritt 4e."); - }); - }); - } -} diff --git a/client-desktop/src/recorder.rs b/client-desktop/src/recorder.rs index f870fc6..422d3d5 100644 --- a/client-desktop/src/recorder.rs +++ b/client-desktop/src/recorder.rs @@ -12,11 +12,10 @@ use std::process::Stdio; use std::time::Duration; use thiserror::Error; -use tokio::io::AsyncWriteExt; use tokio::process::{Child, Command}; use tokio::sync::{mpsc, oneshot}; use tokio::time::timeout; -use tracing::warn; +use tracing::{info, warn}; /// Max time to wait for ffmpeg to finalize the MOOV atom after stop; past /// this we hard-kill and emit `Failed`. @@ -52,6 +51,25 @@ impl Recorder { /// continues until `stop()` is called, then finalizes the m4a. pub fn start( output_path: PathBuf, + ) -> Result<(Self, mpsc::UnboundedReceiver), RecorderError> { + let input = audio_input_args(); + Self::spawn(output_path, &input) + } + + /// Test-only entry point: spawn with custom input args (e.g. `lavfi` + /// test sources instead of a real audio device). Lets integration + /// tests exercise the full lifecycle without depending on hardware. + #[cfg(test)] + fn start_with_input_args( + output_path: PathBuf, + input_args: &[&str], + ) -> Result<(Self, mpsc::UnboundedReceiver), RecorderError> { + Self::spawn(output_path, input_args) + } + + fn spawn( + output_path: PathBuf, + input_args: &[&str], ) -> Result<(Self, mpsc::UnboundedReceiver), RecorderError> { if which::which("ffmpeg").is_err() { return Err(RecorderError::FfmpegMissing); @@ -61,7 +79,7 @@ impl Recorder { let (stop_tx, stop_rx) = oneshot::channel(); let mut cmd = Command::new("ffmpeg"); - cmd.args(ffmpeg_args(&output_path)) + cmd.args(ffmpeg_args(&output_path, input_args)) .stdin(Stdio::piped()) .stdout(Stdio::null()) .stderr(Stdio::null()); @@ -108,23 +126,55 @@ async fn run_recorder_task( } } -/// Send 'q' to ffmpeg's stdin, wait up to `STOP_GRACEFUL_TIMEOUT` for a -/// clean exit; hard-kill on timeout. +/// Graceful stop. On Unix, ffmpeg does NOT poll stdin for keys when +/// stdin is a pipe (its `term_init()` calls `isatty()` and disables +/// `stdin_interaction` for non-TTY), so writing `q` is silently +/// ignored. SIGINT is honored regardless of TTY status — ffmpeg's +/// signal handler sets a flag, the main loop breaks out between +/// frames, MOOV atom is written, `exit(0)` fires. +/// +/// On Windows there is no equivalent via `tokio::process` without +/// launching the child in its own process group; we fall back to +/// writing `q` to stdin and rely on the timeout for corrupt-output +/// detection. Proper Windows graceful stop is future work. async fn finalize( child: &mut Child, output_path: PathBuf, events_tx: &mpsc::UnboundedSender, ) { - if let Some(stdin) = child.stdin.as_mut() - && let Err(e) = stdin.write_all(b"q\n").await + #[cfg(unix)] { - warn!(error = %e, "failed to write stop to ffmpeg stdin"); + if let Some(pid) = child.id() { + match send_sigint(pid) { + Ok(()) => info!(pid, "SIGINT sent to ffmpeg"), + Err(e) => warn!(pid, error = %e, "SIGINT failed"), + } + } } + #[cfg(windows)] + { + use tokio::io::AsyncWriteExt; + if let Some(mut stdin) = child.stdin.take() + && let Err(e) = stdin.write_all(b"q").await + { + warn!(error = %e, "failed to write stop to ffmpeg stdin"); + } + } + + // SIGINT causes ffmpeg to exit with a non-zero status (often 255) + // even when the MOOV atom was written cleanly. Don't second-guess + // the exit code — verify the output file was produced and is + // non-empty. Real corruption will surface server-side at decode. let event = match timeout(STOP_GRACEFUL_TIMEOUT, child.wait()).await { - Ok(Ok(status)) if status.success() => RecorderEvent::Finished { output: output_path }, - Ok(Ok(status)) => RecorderEvent::Failed { - error: format!("ffmpeg exited with {status}"), + Ok(Ok(status)) => match tokio::fs::metadata(&output_path).await { + Ok(meta) if meta.len() > 0 => RecorderEvent::Finished { output: output_path }, + Ok(_) => RecorderEvent::Failed { + error: format!("ffmpeg exit {status} and empty output"), + }, + Err(e) => RecorderEvent::Failed { + error: format!("ffmpeg exit {status} and no output: {e}"), + }, }, Ok(Err(e)) => RecorderEvent::Failed { error: format!("wait ffmpeg: {e}"), @@ -140,10 +190,24 @@ async fn finalize( let _ = events_tx.send(event); } -fn ffmpeg_args(output: &Path) -> Vec { +/// Send SIGINT to a child PID. Used only on Unix. +#[cfg(unix)] +fn send_sigint(pid: u32) -> std::io::Result<()> { + // SAFETY: libc::kill accepts any pid_t; SIGINT is a well-known + // signal. If the child already exited we get ESRCH, which we + // convert to io::Error for the caller to log. + let result = unsafe { libc::kill(pid as i32, libc::SIGINT) }; + if result == 0 { + Ok(()) + } else { + Err(std::io::Error::last_os_error()) + } +} + +fn ffmpeg_args(output: &Path, input_args: &[&str]) -> Vec { let mut args: Vec = vec!["-loglevel".into(), "error".into()]; - for s in audio_input_args() { - args.push(s.to_string()); + for s in input_args { + args.push((*s).to_string()); } args.extend([ "-c:a".into(), @@ -185,7 +249,8 @@ mod tests { #[test] fn ffmpeg_args_contain_output_path_and_encoder() { let path = Path::new("/tmp/test.m4a"); - let args = ffmpeg_args(path); + let input = audio_input_args(); + let args = ffmpeg_args(path, &input); assert!(args.contains(&"/tmp/test.m4a".to_string())); assert!(args.contains(&"-y".to_string())); assert!(args.contains(&"aac".to_string())); @@ -194,8 +259,56 @@ mod tests { #[test] fn ffmpeg_args_have_exactly_one_input() { - let args = ffmpeg_args(Path::new("/tmp/x.m4a")); + let input = audio_input_args(); + let args = ffmpeg_args(Path::new("/tmp/x.m4a"), &input); let input_flags = args.iter().filter(|s| *s == "-i").count(); assert_eq!(input_flags, 1); } + + /// End-to-end: spawn ffmpeg with a lavfi silence source, stop via + /// `Recorder::stop()`, verify the m4a is finalized (exists, has + /// non-trivial size, starts with a valid MP4 `ftyp` atom). + /// + /// `#[ignore]` because it requires ffmpeg in PATH and takes ~1s. + /// Run with: `cargo test -p doctate-desktop -- --ignored`. + #[tokio::test] + #[ignore = "requires ffmpeg in PATH"] + async fn records_lavfi_source_and_stops_cleanly() { + let tmp = tempfile::TempDir::new().unwrap(); + let output = tmp.path().join("lavfi.m4a"); + + let (recorder, mut events) = Recorder::start_with_input_args( + output.clone(), + &["-f", "lavfi", "-i", "anullsrc=r=48000:cl=mono"], + ) + .expect("start recorder"); + + match events.recv().await { + Some(RecorderEvent::Started) => {} + other => panic!("expected Started, got {other:?}"), + } + + tokio::time::sleep(Duration::from_millis(500)).await; + recorder.stop(); + + let event = tokio::time::timeout(Duration::from_secs(10), events.recv()) + .await + .expect("recorder event timed out") + .expect("event channel closed"); + + let output_file = match event { + RecorderEvent::Finished { output } => output, + other => panic!("expected Finished, got {other:?}"), + }; + + assert!(output_file.exists(), "output file should exist"); + let bytes = std::fs::read(&output_file).unwrap(); + assert!( + bytes.len() > 500, + "output should be > 500 bytes, got {}", + bytes.len() + ); + // MP4/m4a: "ftyp" box magic at offset 4. + assert_eq!(&bytes[4..8], b"ftyp", "should start with MP4 ftyp atom"); + } } diff --git a/client-desktop/src/state.rs b/client-desktop/src/state.rs new file mode 100644 index 0000000..bcea817 --- /dev/null +++ b/client-desktop/src/state.rs @@ -0,0 +1,46 @@ +//! UI-visible application state. Transitions are driven by user actions +//! (buttons) and async events from the recorder + uploader channels. + +use std::time::Instant; + +use uuid::Uuid; + +#[derive(Debug)] +pub enum AppState { + /// No valid config on disk — show settings panel. + NotConfigured, + /// Ready for a new recording. No ephemeral context. + Idle, + /// ffmpeg is actively capturing audio. + Recording { + case_id: Uuid, + recorded_at: String, + started_at: Instant, + }, + /// User pressed Stop; waiting for ffmpeg to finalize the MOOV atom. + FinalizingRecording { + case_id: Uuid, + recorded_at: String, + }, + /// File handed to uploader; waiting for server ACK. + Uploading { + case_id: Uuid, + }, + /// Terminal error; user must dismiss or fix config. + Error { + message: String, + }, +} + +impl AppState { + pub fn label(&self) -> &'static str { + match self { + Self::NotConfigured => "Konfiguration fehlt", + Self::Idle => "Bereit", + Self::Recording { .. } => "Aufnahme läuft", + Self::FinalizingRecording { .. } => "Speichere…", + Self::Uploading { .. } => "Upload läuft", + Self::Error { .. } => "Fehler", + } + } +}