vmic/src/pw/text.rs

//! Byte-safe text scraping for anything that has to match against
//! human-authored device names/descriptions.
//!
//! PipeWire's structured JSON dump (`pw-dump --json`) has had real problems
//! serializing certain multibyte UTF-8 sequences in node/device properties -
//! a hardware mic whose description contains a "TM"-style trademark symbol
//! is a documented case that can produce JSON a strict deserializer refuses
//! to parse. This module shells out to `pactl`'s plain-text listings
//! instead, and reads output with `String::from_utf8_lossy`, so a stray or
//! unexpected byte sequence degrades to `\u{FFFD}` instead of aborting the
//! whole match or corrupting a JSON parse.
//!
//! This module owns ONLY name/description matching. Actually creating and
//! linking nodes goes through `pw::graph`'s numeric-id API once a match is
//! resolved to a concrete node name.

use crate::error::Result;
use std::process::Command;

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SourceMatch {
    pub name: String,
    pub description: String,
}

/// Runs `pactl list sources` and parses Name/Description pairs.
pub fn list_sources() -> Result<Vec<SourceMatch>> {
    let output = Command::new("pactl").args(["list", "sources"]).output()?;
    Ok(parse_pactl_sources(&String::from_utf8_lossy(&output.stdout)))
}

fn parse_pactl_sources(text: &str) -> Vec<SourceMatch> {
    let mut sources = Vec::new();
    let mut current_name: Option<String> = None;

    for line in text.lines() {
        let trimmed = line.trim_start();
        if let Some(rest) = trimmed.strip_prefix("Name: ") {
            current_name = Some(rest.trim().to_string());
        } else if let Some(rest) = trimmed.strip_prefix("Description: ") {
            if let Some(name) = current_name.take() {
                sources.push(SourceMatch {
                    name,
                    description: rest.trim().to_string(),
                });
            }
        }
    }
    sources
}

/// Excludes monitor sources and vmic's own nodes (a vmic node routed as its
/// own hardware source would be an unguarded feedback loop - see
/// `PwGraph::break_feedback_rings`, which doesn't catch this case since it
/// only watches for rings through a sink's monitor port). Case-insensitive
/// substring match against "name description", or an explicit numeric
/// index into `pactl list short sources` (same exclusions apply there too).
pub fn match_source(filter: &str) -> Result<Vec<SourceMatch>> {
    if let Ok(idx) = filter.parse::<u32>() {
        let output = Command::new("pactl")
            .args(["list", "short", "sources"])
            .output()?;
        let text = String::from_utf8_lossy(&output.stdout);
        return Ok(text
            .lines()
            .filter_map(|l| {
                let mut cols = l.split_whitespace();
                let id = cols.next()?;
                let name = cols.next()?;
                (id == idx.to_string()).then(|| SourceMatch {
                    name: name.to_string(),
                    description: String::new(),
                })
            })
            .filter(|s| !s.name.ends_with(".monitor") && !s.name.starts_with("vmic_"))
            .collect());
    }

    let needle = filter.to_lowercase();
    Ok(list_sources()?
        .into_iter()
        .filter(|s| !s.name.ends_with(".monitor") && !s.name.starts_with("vmic_"))
        .filter(|s| format!("{} {}", s.name, s.description).to_lowercase().contains(&needle))
        .collect())
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StreamMatch {
    pub id: String,
    pub application: String,
    pub media_name: String,
}

/// Matches a numeric id, "app", or "app:media" (case-insensitive
/// substring) against sink-inputs or source-outputs.
pub fn match_streams(kind: &str, filter: &str) -> Result<Vec<StreamMatch>> {
    let output = Command::new("pactl").args(["list", kind]).output()?;
    let text = String::from_utf8_lossy(&output.stdout);
    let streams = parse_pactl_streams(&text, kind);

    if let Ok(_) = filter.parse::<u32>() {
        return Ok(streams.into_iter().filter(|s| s.id == filter).collect());
    }

    let mut parts = filter.splitn(2, ':');
    let app = parts.next().unwrap_or("").to_lowercase();
    let media = parts.next().unwrap_or("").to_lowercase();

    Ok(streams
        .into_iter()
        .filter(|s| app.is_empty() || s.application.to_lowercase().contains(&app))
        .filter(|s| media.is_empty() || s.media_name.to_lowercase().contains(&media))
        .collect())
}

fn parse_pactl_streams(text: &str, kind: &str) -> Vec<StreamMatch> {
    let header_prefix = if kind == "sink-inputs" {
        "Sink Input #"
    } else {
        "Source Output #"
    };

    let mut streams = Vec::new();
    let mut id = String::new();
    let mut application = String::new();
    let mut media_name = String::new();
    let mut started = false;

    let flush = |streams: &mut Vec<StreamMatch>, id: &str, application: &str, media_name: &str| {
        if !id.is_empty() {
            streams.push(StreamMatch {
                id: id.to_string(),
                application: application.to_string(),
                media_name: media_name.to_string(),
            });
        }
    };

    for line in text.lines() {
        if let Some(rest) = line.strip_prefix(header_prefix) {
            flush(&mut streams, &id, &application, &media_name);
            id = rest.trim_start_matches('#').trim().to_string();
            application.clear();
            media_name.clear();
            started = true;
            continue;
        }
        if !started {
            continue;
        }
        let trimmed = line.trim_start();
        if let Some(rest) = trimmed.strip_prefix("application.name = ") {
            application = unquote(rest);
        } else if let Some(rest) = trimmed.strip_prefix("media.name = ") {
            media_name = unquote(rest);
        }
    }
    flush(&mut streams, &id, &application, &media_name);
    streams
}

fn unquote(s: &str) -> String {
    s.trim_matches('"').to_string()
}

/// The sink-input (playback stream) owned by the given `module-loopback`
/// id, if it has appeared yet. Callers retry this briefly - see
/// `pw::loopback::set_loopback_volume`.
pub fn find_sink_input_for_module(module_id: u32) -> Result<Option<u32>> {
    let output = Command::new("pactl").args(["list", "sink-inputs"]).output()?;
    Ok(parse_sink_input_for_module(&String::from_utf8_lossy(&output.stdout), module_id))
}

fn parse_sink_input_for_module(text: &str, module_id: u32) -> Option<u32> {
    let mut current_id: Option<u32> = None;
    for line in text.lines() {
        if let Some(rest) = line.strip_prefix("Sink Input #") {
            current_id = rest.trim().parse().ok();
            continue;
        }
        let trimmed = line.trim_start();
        if let Some(rest) = trimmed.strip_prefix("Owner Module: ") {
            if rest.trim().parse::<u32>().ok() == Some(module_id) {
                return current_id;
            }
        }
    }
    None
}

/// Numeric pulse-side id for `name` within `kind` ("sinks" or "sources"),
/// via `pactl list short <kind>` - same id/name column layout used by
/// `match_source`'s numeric branch and `wipe::first_non_vmic`.
fn resolve_pulse_id(kind: &str, name: &str) -> Result<Option<u32>> {
    let output = Command::new("pactl").args(["list", "short", kind]).output()?;
    Ok(String::from_utf8_lossy(&output.stdout)
        .lines()
        .find_map(|l| {
            let mut cols = l.split_whitespace();
            let id = cols.next()?.parse::<u32>().ok()?;
            (cols.next()? == name).then_some(id)
        }))
}

/// Ids of every sink-input currently targeting `sink_name`. Used by
/// `commands::topology::migrate` to snapshot stream routing before a
/// migration recreates the sink node, so they can be moved back afterward.
pub fn sink_inputs_on(sink_name: &str) -> Result<Vec<u32>> {
    let Some(id) = resolve_pulse_id("sinks", sink_name)? else {
        return Ok(Vec::new());
    };
    let output = Command::new("pactl").args(["list", "sink-inputs"]).output()?;
    Ok(parse_ids_matching(&String::from_utf8_lossy(&output.stdout), "Sink Input #", "Sink: ", id))
}

/// Ids of every source-output currently reading from `source_name`. See
/// `sink_inputs_on`.
pub fn source_outputs_on(source_name: &str) -> Result<Vec<u32>> {
    let Some(id) = resolve_pulse_id("sources", source_name)? else {
        return Ok(Vec::new());
    };
    let output = Command::new("pactl").args(["list", "source-outputs"]).output()?;
    Ok(parse_ids_matching(&String::from_utf8_lossy(&output.stdout), "Source Output #", "Source: ", id))
}

/// Ids of every block (headed by `header_prefix<id>`) whose `key_prefix<id>`
/// property line equals `target_id`.
fn parse_ids_matching(text: &str, header_prefix: &str, key_prefix: &str, target_id: u32) -> Vec<u32> {
    let mut ids = Vec::new();
    let mut current_id: Option<u32> = None;

    for line in text.lines() {
        if let Some(rest) = line.strip_prefix(header_prefix) {
            current_id = rest.trim().parse().ok();
            continue;
        }
        let trimmed = line.trim_start();
        if let Some(rest) = trimmed.strip_prefix(key_prefix) {
            if rest.trim().parse::<u32>().ok() == Some(target_id) {
                if let Some(id) = current_id {
                    ids.push(id);
                }
            }
        }
    }
    ids
}

/// Every `module-loopback` whose `Argument:` line mentions a `vmic_` node,
/// tracked in state or not.
pub fn list_vmic_loopback_module_ids() -> Result<Vec<u32>> {
    let output = Command::new("pactl").args(["list", "modules"]).output()?;
    Ok(parse_vmic_loopback_module_ids(&String::from_utf8_lossy(&output.stdout)))
}

fn parse_vmic_loopback_module_ids(text: &str) -> Vec<u32> {
    let mut ids = Vec::new();
    let mut current_id: Option<u32> = None;
    let mut is_loopback = false;

    for line in text.lines() {
        if let Some(rest) = line.strip_prefix("Module #") {
            current_id = rest.trim().parse().ok();
            is_loopback = false;
            continue;
        }
        let trimmed = line.trim_start();
        if let Some(rest) = trimmed.strip_prefix("Name: ") {
            is_loopback = rest.trim() == "module-loopback";
        } else if let Some(rest) = trimmed.strip_prefix("Argument:") {
            if is_loopback && rest.contains("vmic_") {
                if let Some(id) = current_id {
                    ids.push(id);
                }
            }
        }
    }
    ids
}

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

    #[test]
    fn finds_sink_input_owned_by_module() {
        let text = "\
Sink Input #10
\tOwner Module: n/a
Sink Input #11
\tOwner Module: 42
Sink Input #12
\tOwner Module: 7
";
        assert_eq!(parse_sink_input_for_module(text, 42), Some(11));
        assert_eq!(parse_sink_input_for_module(text, 99), None);
    }

    #[test]
    fn finds_sink_inputs_targeting_sink() {
        let text = "\
Sink Input #10
\tSink: 65
Sink Input #11
\tSink: 99
Sink Input #12
\tSink: 65
";
        assert_eq!(parse_ids_matching(text, "Sink Input #", "Sink: ", 65), vec![10, 12]);
        assert_eq!(parse_ids_matching(text, "Sink Input #", "Sink: ", 1), Vec::<u32>::new());
    }

    #[test]
    fn finds_source_outputs_targeting_source() {
        let text = "\
Source Output #20
\tSource: 3
Source Output #21
\tSource: 4
";
        assert_eq!(parse_ids_matching(text, "Source Output #", "Source: ", 3), vec![20]);
    }

    #[test]
    fn lists_only_vmic_loopback_modules() {
        let text = "\
Module #1
\tName: module-loopback
\tArgument: source=vmic_test_sink.monitor sink=@DEFAULT_SINK@
Module #2
\tName: module-loopback
\tArgument: source=alsa_input.usb-mic sink=@DEFAULT_SINK@
Module #3
\tName: libpipewire-module-rt
\tArgument: vmic_should_not_match
";
        assert_eq!(parse_vmic_loopback_module_ids(text), vec![1]);
    }
}