vmic/src/pw/loopback.rs

use crate::error::{Result, VmicError};
use crate::pw::graph::{Port, PwGraph};
use crate::pw::text;
use std::process::{Child, Command, Stdio};
use std::time::{Duration, Instant};

pub const LATENCY_MS: u32 = 32;
pub const SAMPLE_RATE: u32 = 48_000;

const NODE_WAIT_TIMEOUT: Duration = Duration::from_secs(3);
const LINK_RETRY_TIMEOUT: Duration = Duration::from_secs(3);
const STAGE_STARTUP_CHECK: Duration = Duration::from_millis(500);

/// Node names derived for a vmic. Always ASCII, since they come from a name
/// that already passed `state::valid_name`.
pub struct NodeNames {
    pub sink: String,   // stage 1 capture (apps play into this)
    pub mid: String,    // stage 1 playback (internal)
    pub mix: String,    // stage 2 capture (internal, hw mic joins here)
    pub source: String, // stage 2 playback (recorders read from this)
    pub sink_desc: String,
    pub mid_desc: String,
    pub mix_desc: String,
    pub source_desc: String,
}

impl NodeNames {
    pub fn for_vmic(name: &str) -> Self {
        let lower = name.to_lowercase();
        let upper = name.to_uppercase();
        Self {
            sink: format!("vmic_{lower}_sink"),
            mid: format!("vmic_{lower}_mid"),
            mix: format!("vmic_{lower}_mix"),
            source: format!("vmic_{lower}_mic"),
            sink_desc: format!("VIRTUAL_MIC_{upper}_OUTPUT"),
            mid_desc: format!("VIRTUAL_MIC_{upper}_MID_INTERNAL"),
            mix_desc: format!("VIRTUAL_MIC_{upper}_MIX_INTERNAL"),
            source_desc: format!("VIRTUAL_MIC_{upper}_INPUT"),
        }
    }
}

/// OS pids and PipeWire node ids produced by [`create_stages`], persisted in
/// `VmicState` so later commands can terminate the processes and check
/// liveness against the live graph.
pub struct StageHandles {
    pub sink_pid: u32,
    pub mix_pid: u32,
    pub sink_node_id: u32,
    pub mix_node_id: u32,
}

/// OS pid and sink PipeWire node id produced by [`create_simple_stage`].
pub struct SingleStageHandles {
    pub pid: u32,
    pub sink_node_id: u32,
}

fn log_file_for(names: &NodeNames) -> Result<(std::path::PathBuf, std::fs::File)> {
    let log_path = std::env::temp_dir().join(format!(
        "vmic_{}.log",
        names.sink.trim_start_matches("vmic_").trim_end_matches("_sink")
    ));
    let log_file = std::fs::OpenOptions::new().create(true).append(true).open(&log_path)?;
    Ok((log_path, log_file))
}

/// Creates the baseline single-stage loopback: apps -> sink =(pw-loopback)=>
/// source_name -> recorders. Used whenever a 4-node split isn't needed (see
/// `commands::topology::desired_topology`) - the common case, and the reason
/// a fresh vmic only ever costs one process, not two.
///
/// `exclude_sink_ids`/`exclude_source_ids` matter only when this is called as
/// part of a topology migration (`commands::topology::migrate`), where a
/// same-named node from the topology being replaced may still be alive at
/// spawn time; see `PwGraph::wait_for_new_node`. Pass empty slices for a
/// brand-new vmic.
pub fn create_simple_stage(
    graph: &PwGraph,
    names: &NodeNames,
    exclude_sink_ids: &[u32],
    exclude_source_ids: &[u32],
) -> Result<SingleStageHandles> {
    let fmt = format!("audio.rate={SAMPLE_RATE} audio.position=[FL FR]");
    let (log_path, log_file) = log_file_for(names)?;

    let mut child = spawn_loopback_stage(
        &format!(
            "media.class=Audio/Sink node.name={} node.description={} node.virtual=false {fmt}",
            names.sink, names.sink_desc
        ),
        &format!(
            "media.class=Audio/Source node.name={} node.description={} node.virtual=false {fmt}",
            names.source, names.source_desc
        ),
        &log_file,
    )?;

    // Give the process a moment to fail fast (e.g. a name collision) before
    // trusting it.
    std::thread::sleep(STAGE_STARTUP_CHECK);
    if child.try_wait()?.is_some() {
        let _ = child.kill();
        return Err(VmicError::PipeWire(format!(
            "pw-loopback exited immediately, check '{}'.",
            log_path.display()
        )));
    }

    let pid = child.id();

    // Unlike create_stages, there's no link_stage retry loop to implicitly
    // wait through - both nodes must be waited for explicitly.
    let sink_node_id = (|| -> Result<u32> {
        let sink_node = graph.wait_for_new_node(&names.sink, exclude_sink_ids, NODE_WAIT_TIMEOUT)?;
        graph.wait_for_new_node(&names.source, exclude_source_ids, NODE_WAIT_TIMEOUT)?;
        Ok(sink_node.id)
    })();

    let sink_node_id = match sink_node_id {
        Ok(id) => id,
        Err(e) => {
            let _ = child.kill();
            return Err(e);
        }
    };

    Ok(SingleStageHandles { pid, sink_node_id })
}

/// Creates the two chained loopback stages as detached `pw-loopback`
/// subprocesses (stdin is `/dev/null`, never waited on, so the OS
/// reparents them to init on exit), then links stage 1 into stage 2
/// (mid -> mix). Used only when a mixed-in source needs to be kept out of
/// the self-monitor loopback (see `commands::topology`).
///
/// apps -> sink =(stage1)=> mid -> mix =(stage2)=> source -> recorders
///                                   ^ hw mic joins here (route --source)
///
/// `exclude_sink_ids` is the same migration-safety mechanism as
/// `create_simple_stage`'s - `mid`/`mix` never need it, since those names are
/// exclusive to this topology and can't already exist across a migration
/// boundary.
pub fn create_stages(graph: &PwGraph, names: &NodeNames, exclude_sink_ids: &[u32]) -> Result<StageHandles> {
    let fmt = format!("audio.rate={SAMPLE_RATE} audio.position=[FL FR]");
    let (log_path, log_file) = log_file_for(names)?;

    let mut sink_child = spawn_loopback_stage(
        &format!(
            "media.class=Audio/Sink node.name={} node.description={} node.virtual=false {fmt}",
            names.sink, names.sink_desc
        ),
        &format!(
            "media.class=Audio/Source node.name={} node.description={} node.virtual=false {fmt}",
            names.mid, names.mid_desc
        ),
        &log_file,
    )?;

    let mut mix_child = spawn_loopback_stage(
        &format!(
            "media.class=Audio/Sink node.name={} node.description={} node.virtual=false {fmt}",
            names.mix, names.mix_desc
        ),
        &format!(
            "media.class=Audio/Source node.name={} node.description={} node.virtual=false {fmt}",
            names.source, names.source_desc
        ),
        &log_file,
    )?;

    // Give both processes a moment to fail fast (e.g. a name collision)
    // before trusting them.
    std::thread::sleep(STAGE_STARTUP_CHECK);
    if sink_child.try_wait()?.is_some() || mix_child.try_wait()?.is_some() {
        let _ = sink_child.kill();
        let _ = mix_child.kill();
        return Err(VmicError::PipeWire(format!(
            "pw-loopback exited immediately, check '{}'.",
            log_path.display()
        )));
    }

    let sink_pid = sink_child.id();
    let mix_pid = mix_child.id();

    let stage_ids = (|| -> Result<(u32, u32)> {
        let sink_node = graph.wait_for_new_node(&names.sink, exclude_sink_ids, NODE_WAIT_TIMEOUT)?;
        let mix_node = graph.wait_for_node(&names.mix, NODE_WAIT_TIMEOUT)?;
        link_stage(graph, &names.mid, &names.mix, LINK_RETRY_TIMEOUT)?;
        Ok((sink_node.id, mix_node.id))
    })();

    let (sink_node_id, mix_node_id) = match stage_ids {
        Ok(ids) => ids,
        Err(e) => {
            let _ = sink_child.kill();
            let _ = mix_child.kill();
            return Err(e);
        }
    };

    Ok(StageHandles { sink_pid, mix_pid, sink_node_id, mix_node_id })
}

fn spawn_loopback_stage(capture_props: &str, playback_props: &str, log_file: &std::fs::File) -> Result<Child> {
    Command::new("pw-loopback")
        .arg(format!("--latency={LATENCY_MS}"))
        .args(["-m", "[ FL FR ]"])
        .arg(format!("--capture-props={capture_props}"))
        .arg(format!("--playback-props={playback_props}"))
        .stdin(Stdio::null())
        .stdout(log_file.try_clone()?)
        .stderr(log_file.try_clone()?)
        .spawn()
        .map_err(VmicError::from)
}

/// Resolves and links `from_node`'s output ports into `target_node`'s input
/// ports, retrying briefly while ports may still be appearing. Shared by
/// stage creation and `route -s`.
pub fn link_stage(graph: &PwGraph, from_node: &str, target_node: &str, timeout: Duration) -> Result<u32> {
    let deadline = Instant::now() + timeout;
    loop {
        let outs: Vec<Port> = graph.ports_for(from_node).into_iter().filter(|p| !p.is_input).collect();
        let ins: Vec<Port> = graph.ports_for(target_node).into_iter().filter(|p| p.is_input).collect();

        if let Some(pairs) = resolve_link_pairs(&outs, &ins) {
            let linked = pairs.iter().filter(|(o, i)| graph.link_ports(*o, *i).is_ok()).count() as u32;
            if linked > 0 {
                return Ok(linked);
            }
        }

        if Instant::now() >= deadline {
            return Err(VmicError::PortResolution {
                from: from_node.to_string(),
                to: target_node.to_string(),
            });
        }
        graph.poll_tick();
    }
}

/// Resolves the same port pairs `link_stage` would use and unlinks them,
/// best-effort. Used to tear down a mixed-in hardware source.
pub fn unlink_stage(graph: &PwGraph, from_node: &str, target_node: &str) -> Result<u32> {
    let outs: Vec<Port> = graph.ports_for(from_node).into_iter().filter(|p| !p.is_input).collect();
    let ins: Vec<Port> = graph.ports_for(target_node).into_iter().filter(|p| p.is_input).collect();

    let Some(pairs) = resolve_link_pairs(&outs, &ins) else {
        return Ok(0);
    };

    let mut removed = 0;
    for (o, i) in pairs {
        if graph.unlink_ports(o, i).is_ok() {
            removed += 1;
        }
    }
    Ok(removed)
}

fn find_port<'a>(ports: &'a [Port], name: &str) -> Option<&'a Port> {
    ports.iter().find(|p| p.name == name)
}

/// Resolves link pairs for feeding `outs` into `ins`. A mono source is
/// linked into both input channels.
fn resolve_link_pairs(outs: &[Port], ins: &[Port]) -> Option<Vec<(u32, u32)>> {
    let mut fl = find_port(outs, "capture_FL");
    let mut fr = find_port(outs, "capture_FR");
    // Sinks name their input ports playback_*; some name them input_*.
    let mut ifl = find_port(ins, "input_FL").or_else(|| find_port(ins, "playback_FL"));
    let mut ifr = find_port(ins, "input_FR").or_else(|| find_port(ins, "playback_FR"));

    // Fall back to first/second port for exotic port names.
    if fl.is_none() {
        fl = outs.first();
    }
    if fr.is_none() {
        fr = outs.get(1);
    }
    if ifl.is_none() {
        ifl = ins.first();
    }
    if ifr.is_none() {
        ifr = ins.get(1);
    }

    let fl = fl?;
    let ifl = ifl?;

    let mut pairs = vec![(fl.id, ifl.id)];
    match (fr, ifr) {
        (Some(fr), Some(ifr)) => pairs.push((fr.id, ifr.id)),
        (None, Some(ifr)) => pairs.push((fl.id, ifr.id)), // mono source: feed both channels
        _ => {}
    }
    Some(pairs)
}

/// SIGTERM a verified pw-loopback pid, escalating to SIGKILL if it lingers.
/// `expected_name_fragment` must appear in the process's cmdline first, as
/// insurance against a stale/reused pid.
pub fn terminate_stage(pid: i64, expected_name_fragment: &str) -> Result<()> {
    let pid = pid as libc::pid_t;

    if !pid_matches(pid, expected_name_fragment) {
        return Ok(());
    }

    unsafe { libc::kill(pid, libc::SIGTERM) };

    for _ in 0..10 {
        if !process_alive(pid) {
            return Ok(());
        }
        std::thread::sleep(Duration::from_millis(100));
    }

    if process_alive(pid) && pid_matches(pid, expected_name_fragment) {
        unsafe { libc::kill(pid, libc::SIGKILL) };
    }
    Ok(())
}

fn process_alive(pid: libc::pid_t) -> bool {
    unsafe { libc::kill(pid, 0) == 0 }
}

/// True only if `pid` is a live `pw-loopback` process whose cmdline
/// mentions `expected_name_fragment`.
fn pid_matches(pid: libc::pid_t, expected_name_fragment: &str) -> bool {
    let Ok(cmdline) = std::fs::read(format!("/proc/{pid}/cmdline")) else {
        return false;
    };
    let text = String::from_utf8_lossy(&cmdline);
    text.contains("pw-loopback") && text.contains(expected_name_fragment)
}

/// Unloads a pactl-hosted module (the self-monitor loopback).
pub fn unload_pulse_module(module_id: u32) -> Result<()> {
    let status = Command::new("pactl")
        .args(["unload-module", &module_id.to_string()])
        .status()?;
    if !status.success() {
        return Err(VmicError::PipeWire(format!("failed to unload module {module_id}")));
    }
    Ok(())
}

/// Loads a self-monitor loopback: sink.monitor -> default sink. `graph` is
/// unused (module loading goes through `pactl`, see `pw/mod.rs`) but kept
/// so the signature matches its call sites.
pub fn enable_self_monitor(_graph: &PwGraph, sink_name: &str) -> Result<u32> {
    let default_sink = String::from_utf8_lossy(&Command::new("pactl").arg("get-default-sink").output()?.stdout)
        .trim()
        .to_string();

    if default_sink == sink_name {
        return Err(VmicError::WouldFeedback);
    }

    let output = Command::new("pactl")
        .args([
            "load-module",
            "module-loopback",
            &format!("source={sink_name}.monitor"),
            "sink=@DEFAULT_SINK@",
        ])
        .output()?;

    if !output.status.success() {
        return Err(VmicError::PipeWire("failed to create self-monitor loopback".into()));
    }
    String::from_utf8_lossy(&output.stdout)
        .trim()
        .parse()
        .map_err(|_| VmicError::PipeWire("failed to create self-monitor loopback".into()))
}

/// Sets the volume of the self-monitor loopback's playback stream,
/// retrying briefly while it may still be appearing.
pub fn set_loopback_volume(_graph: &PwGraph, loopback_id: u32, pct: u8) -> Result<()> {
    let mut sink_input_id = None;
    for _ in 0..5 {
        if let Some(found) = text::find_sink_input_for_module(loopback_id)? {
            sink_input_id = Some(found);
            break;
        }
        std::thread::sleep(Duration::from_millis(200));
    }
    let Some(sink_input_id) = sink_input_id else {
        return Err(VmicError::PipeWire("loopback playback stream not found".into()));
    };

    let status = Command::new("pactl")
        .args(["set-sink-input-volume", &sink_input_id.to_string(), &format!("{pct}%")])
        .status()?;
    if !status.success() {
        return Err(VmicError::PipeWire("pactl set-sink-input-volume failed".into()));
    }
    Ok(())
}