lesavka/client/src/input/microphone.rs

// client/src/input/microphone.rs

#![forbid(unsafe_code)]

use anyhow::{Context, Result};
use gstreamer as gst;
use gstreamer_app as gst_app;
use gst::prelude::*;
use lesavka_common::lesavka::AudioPacket;
use tracing::{debug, error, info, warn, trace};
use shell_escape::unix::escape;
use std::sync::atomic::{AtomicU64, Ordering};

pub struct MicrophoneCapture {
    #[allow(dead_code)]  // kept alive to hold PLAYING state
    pipeline: gst::Pipeline,
    sink:     gst_app::AppSink,
}

impl MicrophoneCapture {
    pub fn new() -> Result<Self> {
        gst::init().ok();                        // idempotent

        /* pulsesrc (default mic) → AAC/ADTS → appsink  -------------------*/
        // Optional override: LESAVKA_MIC_SOURCE=<pulse‑device‑name>
        let device_arg = match std::env::var("LESAVKA_MIC_SOURCE") {
            Ok(s) if !s.is_empty() => {
                let full = Self::pulse_source_by_substr(&s).unwrap_or(s);
                format!("device={}", escape(full.into()))
            }
            _ => String::new(),
        };
        debug!("🎤 device: {device_arg}");
        let aac = ["avenc_aac", "fdkaacenc", "faac", "opusenc"]
            .into_iter()
            .find(|e| gst::ElementFactory::find(e).is_some())
            .unwrap_or("opusenc");
        let parser = if aac.contains("opus") {
            // opusenc already outputs raw Opus frames – just state the caps
            "capsfilter caps=audio/x-opus,rate=48000,channels=2"
        } else {
            // AAC → ADTS frames
            "aacparse ! capsfilter caps=audio/mpeg,stream-format=adts,rate=48000,channels=2"
        };
        let desc = format!(
            "pulsesrc {device_arg} do-timestamp=true ! \
             audio/x-raw,format=S16LE,channels=2,rate=48000 ! \
             audioconvert ! audioresample ! {aac} bitrate=128000 ! \
             {parser} ! \
             queue max-size-buffers=100 leaky=downstream ! \
             appsink name=asink emit-signals=true max-buffers=50 drop=true"
        );

        let pipeline: gst::Pipeline = gst::parse::launch(&desc)?
            .downcast()
            .expect("pipeline");
        let sink: gst_app::AppSink = pipeline
            .by_name("asink")
            .unwrap()
            .downcast()
            .unwrap();

        /* ─── bus for diagnostics ───────────────────────────────────────*/
        {
            let bus = pipeline.bus().unwrap();
            std::thread::spawn(move || {
                use gst::MessageView::*;
                for msg in bus.iter_timed(gst::ClockTime::NONE) {
                    match msg.view() {
                        StateChanged(s) if s.current() == gst::State::Playing
                            && msg.src().map(|s| s.is::<gst::Pipeline>()).unwrap_or(false) =>
                                info!("🎤 mic pipeline ▶️ (source=pulsesrc)"),
                        Error(e) =>
                            error!("🎤💥 mic: {} ({})", e.error(), e.debug().unwrap_or_default()),
                        Warning(w) =>
                            warn!("🎤⚠️ mic: {} ({})", w.error(), w.debug().unwrap_or_default()),
                        _ => {}
                    }
                }
            });
        }

        pipeline.set_state(gst::State::Playing)
            .context("start mic pipeline")?;

        Ok(Self { pipeline, sink })
    }

    /// Blocking pull; call from an async wrapper
    pub fn pull(&self) -> Option<AudioPacket> {
        match self.sink.pull_sample() {
            Ok(sample) => {
                let buf = sample.buffer().unwrap();
                let map = buf.map_readable().unwrap();
                let pts = buf.pts().unwrap_or(gst::ClockTime::ZERO).nseconds() / 1_000;
                static CNT: AtomicU64 = AtomicU64::new(0);
                let n = CNT.fetch_add(1, Ordering::Relaxed);
                if n < 10 || n % 300 == 0 {
                    trace!("🎤⇧ cli  pkt#{n}  {} bytes", map.len());
                }
                Some(AudioPacket { id: 0, pts, data: map.as_slice().to_vec() })

            }
            Err(_) => None,
        }
    }

    fn pulse_source_by_substr(fragment: &str) -> Option<String> {
        use std::process::Command;
        let out = Command::new("pactl").args(["list", "short", "sources"])
            .output().ok()?;
        let list = String::from_utf8_lossy(&out.stdout);
        list.lines()
            .find_map(|ln| {
                let mut cols = ln.split_whitespace();
                let _id   = cols.next()?;
                let name  = cols.next()?;             // column #1
                if name.contains(fragment) { Some(name.to_owned()) } else { None }
            })
    }
}