doctate/client-desktop/src/recorder.rs

//! Audio recorder via an ffmpeg subprocess. Produces m4a/AAC files on disk.
//!
//! Portability note: this module is intentionally UI-agnostic and carries no
//! `egui`/`eframe` deps so it can be lifted into a shared client-core crate
//! once a second Rust client (Windows variant, etc.) arrives.
//!
//! Call `Recorder::start` inside a Tokio runtime context (entered with
//! `Runtime::enter()` or via `#[tokio::main]`).

use std::path::{Path, PathBuf};
use std::process::Stdio;
use std::time::Duration;

use thiserror::Error;
use tokio::process::{Child, Command};
use tokio::sync::{mpsc, oneshot};
use tokio::time::timeout;
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`.
const STOP_GRACEFUL_TIMEOUT: Duration = Duration::from_secs(5);

#[derive(Debug)]
pub enum RecorderEvent {
    /// ffmpeg spawned successfully; audio capture has begun.
    Started,
    /// ffmpeg exited cleanly; the m4a file is ready.
    Finished { output: PathBuf },
    /// ffmpeg or I/O failed; output file may be missing or corrupt.
    Failed { error: String },
}

#[derive(Debug, Error)]
pub enum RecorderError {
    #[error("ffmpeg not found in PATH")]
    FfmpegMissing,
    #[error("spawn ffmpeg failed: {0}")]
    Spawn(#[from] std::io::Error),
}

/// Handle to a running recorder. Dropping it does NOT stop the recorder —
/// call `stop()` explicitly (consumes self).
pub struct Recorder {
    stop_tx: oneshot::Sender<()>,
}

impl Recorder {
    /// Spawn a recorder task that captures the system default audio input
    /// to `output_path`. Returns a handle + event receiver. The task
    /// continues until `stop()` is called, then finalizes the m4a.
    pub fn start(
        output_path: PathBuf,
    ) -> Result<(Self, mpsc::UnboundedReceiver<RecorderEvent>), 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<RecorderEvent>), RecorderError> {
        Self::spawn(output_path, input_args)
    }

    fn spawn(
        output_path: PathBuf,
        input_args: &[&str],
    ) -> Result<(Self, mpsc::UnboundedReceiver<RecorderEvent>), RecorderError> {
        if which::which("ffmpeg").is_err() {
            return Err(RecorderError::FfmpegMissing);
        }

        let (events_tx, events_rx) = mpsc::unbounded_channel();
        let (stop_tx, stop_rx) = oneshot::channel();

        let mut cmd = Command::new("ffmpeg");
        cmd.args(ffmpeg_args(&output_path, input_args))
            .stdin(Stdio::piped())
            .stdout(Stdio::null())
            .stderr(Stdio::null());

        let child = cmd.spawn()?;

        tokio::spawn(run_recorder_task(child, output_path, stop_rx, events_tx));

        Ok((Self { stop_tx }, events_rx))
    }

    /// Request the recorder to finalize and exit. Consumes the handle —
    /// no double-stop. Completion arrives as `Finished` (or `Failed`) on
    /// the event receiver.
    pub fn stop(self) {
        let _ = self.stop_tx.send(());
    }
}

async fn run_recorder_task(
    mut child: Child,
    output_path: PathBuf,
    stop_rx: oneshot::Receiver<()>,
    events_tx: mpsc::UnboundedSender<RecorderEvent>,
) {
    let _ = events_tx.send(RecorderEvent::Started);

    tokio::select! {
        exit_result = child.wait() => {
            let event = match exit_result {
                Ok(status) if status.success() => RecorderEvent::Finished { output: output_path },
                Ok(status) => RecorderEvent::Failed {
                    error: format!("ffmpeg exited unexpectedly: {status}"),
                },
                Err(e) => RecorderEvent::Failed {
                    error: format!("wait ffmpeg: {e}"),
                },
            };
            let _ = events_tx.send(event);
        }
        _ = stop_rx => {
            finalize(&mut child, output_path, &events_tx).await;
        }
    }
}

/// 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<RecorderEvent>,
) {
    #[cfg(unix)]
    {
        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)) => 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}"),
        },
        Err(_elapsed) => {
            warn!("ffmpeg did not exit within timeout; killing");
            let _ = child.kill().await;
            RecorderEvent::Failed {
                error: "ffmpeg stop timeout — output may be corrupt".into(),
            }
        }
    };
    let _ = events_tx.send(event);
}

/// 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<String> {
    let mut args: Vec<String> = vec!["-loglevel".into(), "error".into()];
    for s in input_args {
        args.push((*s).to_string());
    }
    args.extend([
        "-c:a".into(),
        "aac".into(),
        "-b:a".into(),
        "96k".into(),
        "-ar".into(),
        "48000".into(),
        "-ac".into(),
        "1".into(),
        "-movflags".into(),
        "+faststart".into(),
        "-y".into(),
        output.to_string_lossy().into_owned(),
    ]);
    args
}

#[cfg(target_os = "linux")]
fn audio_input_args() -> Vec<&'static str> {
    // PulseAudio interface (also works under PipeWire via pipewire-pulse).
    vec!["-f", "pulse", "-i", "default"]
}

#[cfg(target_os = "windows")]
fn audio_input_args() -> Vec<&'static str> {
    // TODO(windows-client): enumerate dshow devices via
    // `ffmpeg -list_devices true -f dshow -i dummy` and cache the first
    // non-loopback capture device. "audio=default" below is a placeholder;
    // most Windows systems do not have a device literally named "default",
    // so the Windows client must resolve a real name before shipping.
    vec!["-f", "dshow", "-i", "audio=default"]
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn ffmpeg_args_contain_output_path_and_encoder() {
        let path = Path::new("/tmp/test.m4a");
        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()));
        assert!(args.contains(&"+faststart".to_string()));
    }

    #[test]
    fn ffmpeg_args_have_exactly_one_input() {
        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");
    }
}