peerspeak/src/audio/echo_cancel.rs

//! Acoustic echo cancellation via PipeWire's `module-echo-cancel`.
//!
//! Rather than running an echo canceller in-process (which would mean a C++ DSP
//! dependency plus plumbing the playout signal back to the capture path as a
//! delay-aligned reference), we delegate to the audio server. PipeWire already
//! sees both the real microphone and the speaker monitor, so loading
//! `module-echo-cancel` (WebRTC AEC + noise suppression + AGC) solves the
//! reference-alignment problem for free.
//!
//! When enabled, the module creates two virtual nodes:
//! - [`EC_SOURCE`] — the cleaned microphone (we capture from this).
//! - [`EC_SINK`] — playout goes here; it doubles as the AEC reference (we play to
//!   this, and the module forwards it to the real output while using it to cancel
//!   the echo from the mic).
//!
//! Binding to a specific mic/speaker (the user's chosen devices) is done with the
//! `source_master`/`sink_master` args; omitting them binds to the system defaults.
//! The loaded module is owned by an [`EchoCancelGuard`] that unloads it on drop, so
//! it never outlives a call (even if the join path bails out early).

use std::process::Command;
use std::time::{Duration, Instant};

/// node.name of the virtual (cleaned) capture source the module creates.
pub const EC_SOURCE: &str = "peerspeak_echocancel_source";
/// node.name of the virtual playback sink (also the AEC reference).
pub const EC_SINK: &str = "peerspeak_echocancel_sink";

/// How long to wait for the virtual nodes to appear after loading the module
/// before giving up — the nodes show up a beat after `load-module` returns.
const NODE_READY_TIMEOUT: Duration = Duration::from_secs(3);

/// Owns a loaded `module-echo-cancel` instance; unloads it on drop so the virtual
/// nodes never leak past the call that created them.
pub struct EchoCancelGuard {
    module_index: String,
    source_name: String,
    sink_name: String,
}

impl EchoCancelGuard {
    pub fn source_name(&self) -> &str {
        &self.source_name
    }

    pub fn sink_name(&self) -> &str {
        &self.sink_name
    }
}

impl Drop for EchoCancelGuard {
    fn drop(&mut self) {
        let _ = Command::new("pactl")
            .arg("unload-module")
            .arg(&self.module_index)
            .output();
        crate::log_msg(&format!(
            "Echo cancel: unloaded module {}",
            self.module_index
        ));
    }
}

/// Loads `module-echo-cancel` (WebRTC AEC) bound to the given real devices, waits
/// for its virtual nodes to come up, and returns a guard that unloads it on drop.
///
/// `real_source`/`real_sink` are the node.name of the chosen mic/speaker; pass
/// `None` (or an empty string) to bind to the system defaults. Returns `Err` with
/// a human-readable reason if `pactl` is missing, the load fails, or the nodes
/// don't appear — the caller should fall back to the direct devices.
pub fn enable(
    real_source: Option<&str>,
    real_sink: Option<&str>,
) -> Result<EchoCancelGuard, String> {
    // Best-effort: clear any stale instance left by a crashed prior run so we
    // don't stack duplicate modules / fight over the virtual node names.
    unload_stale();

    let owner_pid = std::process::id();
    let source_name = format!("{EC_SOURCE}.{owner_pid}");
    let sink_name = format!("{EC_SINK}.{owner_pid}");

    let mut cmd = Command::new("pactl");
    cmd.arg("load-module")
        .arg("module-echo-cancel")
        .arg("aec_method=webrtc")
        .arg(format!("source_name={source_name}"))
        .arg(format!("sink_name={sink_name}"));
    if let Some(src) = real_source.filter(|s| !s.is_empty()) {
        cmd.arg(format!("source_master={src}"));
    }
    if let Some(sink) = real_sink.filter(|s| !s.is_empty()) {
        cmd.arg(format!("sink_master={sink}"));
    }

    let out = cmd
        .output()
        .map_err(|e| format!("pactl not available: {e}"))?;
    if !out.status.success() {
        return Err(format!(
            "pactl load-module failed: {}",
            String::from_utf8_lossy(&out.stderr).trim()
        ));
    }

    let module_index = String::from_utf8_lossy(&out.stdout).trim().to_string();
    if module_index.parse::<u64>().is_err() {
        return Err(format!("unexpected pactl output: {module_index:?}"));
    }
    let guard = EchoCancelGuard {
        module_index,
        source_name,
        sink_name,
    };

    // The virtual nodes appear shortly after the module loads; wait for both so
    // the subsequent capture/playback streams can actually target them. If they
    // never show, drop the guard (unloads) and report failure.
    if !wait_for_nodes(guard.source_name(), guard.sink_name()) {
        return Err("echo-cancel virtual nodes did not appear in time".to_string());
    }

    crate::log_msg(&format!(
        "Echo cancel: loaded module {} (source_master={:?}, sink_master={:?})",
        guard.module_index, real_source, real_sink
    ));
    Ok(guard)
}

/// Polls until both virtual nodes exist or the timeout elapses.
fn wait_for_nodes(source_name: &str, sink_name: &str) -> bool {
    let deadline = Instant::now() + NODE_READY_TIMEOUT;
    loop {
        if node_present("sources", source_name) && node_present("sinks", sink_name) {
            return true;
        }
        if Instant::now() >= deadline {
            return false;
        }
        std::thread::sleep(Duration::from_millis(100));
    }
}

/// Whether `pactl list <kind> short` lists a node named `name`.
/// `kind` is "sources" or "sinks".
fn node_present(kind: &str, name: &str) -> bool {
    let Ok(out) = Command::new("pactl")
        .arg("list")
        .arg(kind)
        .arg("short")
        .output()
    else {
        return false;
    };
    String::from_utf8_lossy(&out.stdout)
        .lines()
        .any(|line| line.split('\t').nth(1) == Some(name))
}

fn pid_from_ec_args(args: &str) -> Option<u32> {
    let source_prefix = format!("source_name={EC_SOURCE}.");
    args.split_whitespace()
        .find_map(|arg| arg.strip_prefix(&source_prefix))?
        .parse()
        .ok()
}

fn ec_module_is_stale(args: &str, is_alive: impl Fn(u32) -> bool) -> bool {
    pid_from_ec_args(args).is_some_and(|pid| !is_alive(pid))
}

#[cfg(target_os = "linux")]
fn process_is_alive(pid: u32) -> bool {
    std::path::Path::new("/proc").join(pid.to_string()).exists()
}

#[cfg(not(target_os = "linux"))]
fn process_is_alive(_pid: u32) -> bool {
    true
}

/// Unloads leftover PeerSpeak `module-echo-cancel` instances only when their
/// owning process is gone. Best-effort and conservative on non-Linux platforms.
fn unload_stale() {
    let Ok(out) = Command::new("pactl")
        .arg("list")
        .arg("modules")
        .arg("short")
        .output()
    else {
        return;
    };
    for line in String::from_utf8_lossy(&out.stdout).lines() {
        let mut cols = line.split('\t');
        let index = cols.next().unwrap_or("");
        let name = cols.next().unwrap_or("");
        let args = cols.next().unwrap_or("");
        if name == "module-echo-cancel"
            && ec_module_is_stale(args, process_is_alive)
            && index.parse::<u64>().is_ok()
        {
            let _ = Command::new("pactl")
                .arg("unload-module")
                .arg(index)
                .output();
            crate::log_msg(&format!("Echo cancel: cleaned up stale module {index}"));
        }
    }
}

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

    /// Exercises the real load -> wait-for-nodes -> drop/unload path against the
    /// live PipeWire daemon. Side-effecting (briefly creates virtual audio nodes),
    /// so it's ignored by default; run with `cargo test -- --ignored echo_cancel`.
    #[test]
    #[ignore]
    fn enable_creates_and_unloads_nodes() {
        let guard = enable(None, None).expect("module-echo-cancel should load");
        let source_name = guard.source_name().to_string();
        let sink_name = guard.sink_name().to_string();
        assert!(
            node_present("sources", &source_name),
            "cleaned source must exist"
        );
        assert!(
            node_present("sinks", &sink_name),
            "reference sink must exist"
        );
        drop(guard);
        // Give pactl a moment to tear the nodes down.
        std::thread::sleep(Duration::from_millis(300));
        assert!(
            !node_present("sources", &source_name),
            "source must be gone after unload"
        );
        assert!(
            !node_present("sinks", &sink_name),
            "sink must be gone after unload"
        );
    }

    #[test]
    fn parses_owner_pid_only_from_our_source_name() {
        assert_eq!(
            pid_from_ec_args(
                "aec_method=webrtc source_name=peerspeak_echocancel_source.4242 sink_name=peerspeak_echocancel_sink.4242"
            ),
            Some(4242)
        );
        assert_eq!(pid_from_ec_args("aec_method=webrtc"), None);
        assert_eq!(
            pid_from_ec_args("source_name=peerspeak_echocancel_source.not-a-pid"),
            None
        );
        assert_eq!(
            pid_from_ec_args("source_name=someone_elses_source.4242"),
            None
        );
    }

    #[test]
    fn stale_decision_keeps_live_and_foreign_modules() {
        let ours = "source_name=peerspeak_echocancel_source.4242";
        assert!(!ec_module_is_stale(ours, |pid| pid == 4242));
        assert!(ec_module_is_stale(ours, |_| false));
        assert!(!ec_module_is_stale("source_name=foreign.4242", |_| false));
        assert!(!ec_module_is_stale(
            "source_name=peerspeak_echocancel_source.malformed",
            |_| false
        ));
    }
}