zk_audio/

native.rs

use crate::builder::{NativePipelineBuilder, PipelineBuildRequest};
use crate::contracts::{CaptureBackend, DeviceEnumerator, ProcessorBuildRequest, ProcessorFactory};
use crate::core::{
    ActiveListening, ActiveRecording, AudioError, AudioResult, CaptureDiagnostics, InputDeviceInfo,
    NativeCaptureConfig,
};
use crate::factory::ProfileProcessorFactory;
use crate::metrics::LevelMetrics;
use crate::mic_sim::MicrophoneSimulatorFactory;
use crate::pipeline::{AudioPipeline, PlaybackBuffer, PlaybackSink};
use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};
use cpal::{SampleFormat, StreamConfig};
use serde_json::Value;
use std::process::Command;
use std::sync::mpsc;
use std::sync::{Arc, Mutex};
use std::thread::{self, JoinHandle};
use std::time::Instant;

pub struct NativeCaptureBackend;

impl DeviceEnumerator for NativeCaptureBackend {
    fn list_input_devices(&self) -> AudioResult<Vec<InputDeviceInfo>> {
        list_input_devices()
    }

    fn set_default_input_device(&self, device_id: &str) -> AudioResult<()> {
        set_default_input_device(device_id)
    }
}

pub fn list_input_devices() -> AudioResult<Vec<InputDeviceInfo>> {
    #[cfg(target_os = "linux")]
    if let Ok(devices) = list_pipewire_input_devices() {
        if !devices.is_empty() {
            return Ok(devices);
        }
    }

    list_cpal_input_devices()
}

pub fn set_default_input_device(device_id: &str) -> AudioResult<()> {
    #[cfg(target_os = "linux")]
    {
        let output = Command::new("pactl")
            .args(["set-default-source", device_id])
            .output()
            .map_err(|err| {
                AudioError::new(format!(
                    "Failed to invoke pactl set-default-source for '{}': {}",
                    device_id, err
                ))
            })?;

        if !output.status.success() {
            let stderr = String::from_utf8_lossy(&output.stderr);
            return Err(AudioError::new(format!(
                "pactl set-default-source '{}' failed: {}",
                device_id,
                stderr.trim()
            )));
        }

        return Ok(());
    }

    #[allow(unreachable_code)]
    Err(AudioError::new(
        "Setting the default input device is only supported on Linux",
    ))
}

fn list_cpal_input_devices() -> AudioResult<Vec<InputDeviceInfo>> {
    let host = cpal::default_host();
    let default_name = host
        .default_input_device()
        .and_then(|device| device.description().ok())
        .map(|d| d.to_string());
    let devices = host
        .input_devices()
        .map_err(|err| AudioError::new(format!("Failed to enumerate input devices: {}", err)))?;

    let mut result = Vec::new();
    for device in devices {
        let name = device
            .description()
            .map_err(|err| AudioError::new(format!("Failed to read input device name: {}", err)))?
            .to_string();
        let is_default = default_name.as_deref() == Some(name.as_str());
        result.push(InputDeviceInfo {
            id: name.clone(),
            name,
            is_default,
        });
    }

    result.sort_by(|a, b| a.name.cmp(&b.name));
    Ok(result)
}

#[cfg(target_os = "linux")]
fn list_pipewire_input_devices() -> AudioResult<Vec<InputDeviceInfo>> {
    let default_source = pactl_default_source().ok().flatten();
    let output = Command::new("pactl")
        .args(["-f", "json", "list", "sources"])
        .output()
        .map_err(|err| AudioError::new(format!("Failed to invoke pactl list sources: {}", err)))?;

    if !output.status.success() {
        let stderr = String::from_utf8_lossy(&output.stderr);
        return Err(AudioError::new(format!(
            "pactl list sources failed: {}",
            stderr.trim()
        )));
    }

    let value: Value = serde_json::from_slice(&output.stdout)
        .map_err(|err| AudioError::new(format!("Failed to parse pactl source list: {}", err)))?;
    let sources = value
        .as_array()
        .ok_or_else(|| AudioError::new("Unexpected pactl source list payload"))?;

    let mut result = Vec::new();

    for source in sources {
        if !is_selectable_pipewire_source(source) {
            continue;
        }

        let Some(id) = source.get("name").and_then(Value::as_str) else {
            continue;
        };

        let display_name = friendly_pipewire_source_name(source).unwrap_or_else(|| id.to_string());
        let is_default = default_source.as_deref() == Some(id);

        result.push(InputDeviceInfo {
            id: id.to_string(),
            name: display_name,
            is_default,
        });
    }

    result.sort_by(|a, b| a.name.cmp(&b.name));
    Ok(result)
}

#[cfg(target_os = "linux")]
fn pactl_default_source() -> AudioResult<Option<String>> {
    let output = Command::new("pactl")
        .arg("info")
        .output()
        .map_err(|err| AudioError::new(format!("Failed to invoke pactl info: {}", err)))?;

    if !output.status.success() {
        let stderr = String::from_utf8_lossy(&output.stderr);
        return Err(AudioError::new(format!(
            "pactl info failed: {}",
            stderr.trim()
        )));
    }

    let stdout = String::from_utf8_lossy(&output.stdout);
    Ok(stdout
        .lines()
        .find_map(|line| line.strip_prefix("Default Source: "))
        .map(|value| value.trim().to_string()))
}

#[cfg(target_os = "linux")]
fn is_selectable_pipewire_source(source: &Value) -> bool {
    let Some(id) = source.get("name").and_then(Value::as_str) else {
        return false;
    };
    if id.ends_with(".monitor") {
        return false;
    }

    if source
        .get("monitor_source")
        .and_then(Value::as_str)
        .is_some_and(|value| !value.is_empty())
    {
        return false;
    }

    let properties = source.get("properties").and_then(Value::as_object);
    if properties
        .and_then(|props| props.get("media.class"))
        .and_then(Value::as_str)
        .is_some_and(|class| class != "Audio/Source")
    {
        return false;
    }

    let description = source
        .get("description")
        .and_then(Value::as_str)
        .unwrap_or_default();
    !description.starts_with("Monitor of ")
}

#[cfg(target_os = "linux")]
fn friendly_pipewire_source_name(source: &Value) -> Option<String> {
    let properties = source.get("properties").and_then(Value::as_object);
    let device_description = properties
        .and_then(|props| props.get("device.description"))
        .and_then(Value::as_str)
        .or_else(|| source.get("description").and_then(Value::as_str))?;

    let active_port_name = source.get("active_port").and_then(Value::as_str);
    let active_port_description = source
        .get("ports")
        .and_then(Value::as_array)
        .and_then(|ports| {
            ports.iter().find_map(|port| {
                let name = port.get("name").and_then(Value::as_str)?;
                if Some(name) == active_port_name {
                    port.get("description").and_then(Value::as_str)
                } else {
                    None
                }
            })
        });

    if let Some(port_description) = active_port_description {
        return Some(format!("{} - {}", port_description, device_description));
    }

    Some(device_description.to_string())
}

impl NativeCaptureBackend {
    pub fn new() -> Self {
        Self
    }

    fn build_pipeline(
        config: &NativeCaptureConfig,
        sample_rate: u32,
        device_channels: u16,
        device_name: Option<String>,
    ) -> AudioResult<AudioPipeline> {
        NativePipelineBuilder::new().build(PipelineBuildRequest {
            output_path: config.output_path.clone(),
            profile: config.profile,
            sample_rate,
            device_channels,
            device_name,
            gain_db: config.input_gain_db,
            limiter_threshold: config.limiter_threshold,
            high_pass_hz: config.high_pass_hz,
            noise_suppression_amount: config.noise_suppression_amount,
            noise_calibration_ms: config.noise_calibration_ms,
            delay_effect: config.delay_effect,
            stage_overrides: config.stage_overrides.clone(),
            microphone_sim: config.microphone_sim,
        })
    }

    fn build_live_pipeline(
        config: &NativeCaptureConfig,
        sample_rate: u32,
        device_channels: u16,
        device_name: Option<String>,
        playback_buffer: Arc<Mutex<PlaybackBuffer>>,
    ) -> AudioResult<AudioPipeline> {
        let output_spec = crate::core::AudioSpec {
            sample_rate,
            channels: 1,
        };
        let processors =
            ProfileProcessorFactory::new().build_processors(ProcessorBuildRequest {
                profile: config.profile,
                gain_db: config.input_gain_db,
                limiter_threshold: config.limiter_threshold,
                high_pass_hz: config.high_pass_hz,
                noise_suppression_amount: config.noise_suppression_amount,
                noise_calibration_ms: config.noise_calibration_ms,
                delay_effect: config.delay_effect,
                stage_overrides: config.stage_overrides.clone(),
            })?;
        let processor_names = processors
            .iter()
            .map(|processor| processor.name().to_string())
            .collect::<Vec<_>>();
        let microphone_sim_processors =
            MicrophoneSimulatorFactory::new().build_processors(config.microphone_sim)?;
        let microphone_sim_processor_names = microphone_sim_processors
            .iter()
            .map(|processor| processor.name().to_string())
            .collect::<Vec<_>>();
        let microphone_sim_model = config
            .microphone_sim
            .active_model()
            .map(|model| model.as_str().to_string());

        let mut notes = config
            .delay_effect
            .map(|effect| vec![format!("delay_effect={}", effect.preset.as_str())])
            .unwrap_or_default();
        notes.push("mode=live_listening".to_string());
        notes.push("output=default".to_string());
        notes.push("pipeline_order=microphone_sim->voice_processing->playback".to_string());
        if let Some(model) = &microphone_sim_model {
            notes.push(format!("microphone_sim={}", model));
        }

        AudioPipeline::new(
            output_spec,
            CaptureDiagnostics {
                backend: "native_listening".to_string(),
                profile: config.profile.as_str().to_string(),
                profile_base: Some(config.profile.as_str().to_string()),
                device_name,
                sample_rate: Some(sample_rate),
                channels: Some(device_channels),
                processor_names,
                processor_stage_overrides: config
                    .stage_overrides
                    .iter()
                    .map(|(stage, mode)| format!("{}={}", stage, mode.as_str()))
                    .collect(),
                resolved_delay_preset: config
                    .delay_effect
                    .map(|effect| effect.preset.as_str().to_string()),
                microphone_sim_model,
                microphone_sim_processor_names,
                notes,
                ..CaptureDiagnostics::default()
            },
            microphone_sim_processors,
            processors,
            Box::new(PlaybackSink::from_buffer(playback_buffer)),
            Box::new(LevelMetrics::default()),
        )
    }

    pub fn start_live(&self, config: NativeCaptureConfig) -> AudioResult<Box<dyn ActiveListening>> {
        let (ready_tx, ready_rx) = mpsc::channel();
        let (cmd_tx, cmd_rx) = mpsc::channel();

        let handle = thread::spawn(move || {
            native_listening_thread(config, cmd_rx, ready_tx);
        });

        match ready_rx
            .recv()
            .map_err(|_| AudioError::new("Native listening thread failed to report readiness"))?
        {
            Ok(()) => Ok(Box::new(NativeListeningHandle {
                cmd_tx,
                join_handle: Some(handle),
            })),
            Err(err) => {
                let _ = handle.join();
                Err(err)
            }
        }
    }
}

impl CaptureBackend for NativeCaptureBackend {
    fn start(&self, config: NativeCaptureConfig) -> AudioResult<Box<dyn ActiveRecording>> {
        let (ready_tx, ready_rx) = mpsc::channel();
        let (cmd_tx, cmd_rx) = mpsc::channel();

        let handle = thread::spawn(move || {
            native_capture_thread(config, cmd_rx, ready_tx);
        });

        match ready_rx
            .recv()
            .map_err(|_| AudioError::new("Native capture thread failed to report readiness"))?
        {
            Ok(()) => Ok(Box::new(NativeRecordingHandle {
                cmd_tx,
                join_handle: Some(handle),
            })),
            Err(err) => {
                let _ = handle.join();
                Err(err)
            }
        }
    }
}

enum NativeCommand {
    Stop(mpsc::Sender<AudioResult<CaptureDiagnostics>>),
}

enum NativeListeningCommand {
    Stop(mpsc::Sender<AudioResult<CaptureDiagnostics>>),
    UpdateConfig(NativeCaptureConfig, mpsc::Sender<AudioResult<()>>),
}

fn native_capture_thread(
    config: NativeCaptureConfig,
    cmd_rx: mpsc::Receiver<NativeCommand>,
    ready_tx: mpsc::Sender<AudioResult<()>>,
) {
    let result = start_native_stream(config);
    let (stream, pipeline, start_time, notes) = match result {
        Ok(values) => {
            let _ = ready_tx.send(Ok(()));
            values
        }
        Err(err) => {
            let _ = ready_tx.send(Err(err));
            return;
        }
    };

    let _keep_stream_alive = stream;

    match cmd_rx.recv() {
        Ok(NativeCommand::Stop(reply_tx)) => {
            drop(_keep_stream_alive);
            let duration_ms = Some(start_time.elapsed().as_millis() as i64);
            let diagnostics = pipeline
                .lock()
                .map_err(|_| AudioError::new("Audio pipeline lock poisoned"))
                .and_then(|mut pipeline| pipeline.finalize(duration_ms))
                .map(|mut diagnostics| {
                    diagnostics.notes.extend(notes);
                    diagnostics
                });
            let _ = reply_tx.send(diagnostics);
        }
        Err(_) => {}
    }
}

fn native_listening_thread(
    config: NativeCaptureConfig,
    cmd_rx: mpsc::Receiver<NativeListeningCommand>,
    ready_tx: mpsc::Sender<AudioResult<()>>,
) {
    let result = start_live_monitor_stream(config);
    let (input_stream, output_stream, pipeline, playback_buffer, start_time, notes) = match result {
        Ok(values) => {
            let _ = ready_tx.send(Ok(()));
            values
        }
        Err(err) => {
            let _ = ready_tx.send(Err(err));
            return;
        }
    };

    let _keep_input_stream_alive = input_stream;
    let _keep_output_stream_alive = output_stream;

    loop {
        match cmd_rx.recv() {
            Ok(NativeListeningCommand::Stop(reply_tx)) => {
                drop(_keep_input_stream_alive);
                drop(_keep_output_stream_alive);
                let duration_ms = Some(start_time.elapsed().as_millis() as i64);
                let diagnostics = pipeline
                    .lock()
                    .map_err(|_| AudioError::new("Audio pipeline lock poisoned"))
                    .and_then(|mut pipeline| pipeline.finalize(duration_ms))
                    .map(|mut diagnostics| {
                        diagnostics.notes.extend(notes.clone());
                        diagnostics
                    });
                let _ = reply_tx.send(diagnostics);
                break;
            }
            Ok(NativeListeningCommand::UpdateConfig(config, reply_tx)) => {
                let result = rebuild_live_pipeline(&pipeline, &playback_buffer, config);
                let _ = reply_tx.send(result);
            }
            Err(_) => break,
        }
    }
}

type NativeThreadState = (
    cpal::Stream,
    Arc<Mutex<AudioPipeline>>,
    Instant,
    Vec<String>,
);

type NativeListeningThreadState = (
    cpal::Stream,
    cpal::Stream,
    Arc<Mutex<AudioPipeline>>,
    Arc<Mutex<PlaybackBuffer>>,
    Instant,
    Vec<String>,
);

fn start_native_stream(config: NativeCaptureConfig) -> AudioResult<NativeThreadState> {
    let host = cpal::default_host();
    let device = select_input_device(&host, config.preferred_input_device.as_deref())?;
    let device_name = device.description().ok().map(|d| d.to_string());
    let supported_config = device
        .default_input_config()
        .map_err(|err| AudioError::new(format!("Failed to read input device config: {}", err)))?;

    let (sample_rate, channels) =
        select_stream_config(&device, config.target_sample_rate, config.target_channels).unwrap_or(
            (
                supported_config.sample_rate(),
                supported_config.channels(),
            ),
        );

    let pipeline = Arc::new(Mutex::new(NativeCaptureBackend::build_pipeline(
        &config,
        sample_rate,
        channels,
        device_name.clone(),
    )?));

    let cpal_config = StreamConfig {
        channels,
        sample_rate: sample_rate,
        buffer_size: cpal::BufferSize::Default,
    };
    let stream = build_input_stream(
        &device,
        supported_config.sample_format(),
        &cpal_config,
        channels,
        Arc::clone(&pipeline),
        native_input_stream_error,
    )?;

    stream
        .play()
        .map_err(|err| AudioError::new(format!("Failed to start input stream: {}", err)))?;

    Ok((
        stream,
        pipeline,
        Instant::now(),
        vec![
            format!("device={}", device_name.unwrap_or_else(|| "unknown".into())),
            format!(
                "configured_device={}",
                config
                    .preferred_input_device
                    .unwrap_or_else(|| "system_default".to_string())
            ),
            format!("channels={}", channels),
            format!("sample_rate={}", sample_rate),
            format!("noise_calibration_ms={}", config.noise_calibration_ms),
        ],
    ))
}

fn native_input_stream_error(err: cpal::StreamError) {
    log::error!("Native audio stream error: {}", err);
}

fn native_listening_input_stream_error(err: cpal::StreamError) {
    log::error!("Native listening input stream error: {}", err);
}

fn native_listening_output_stream_error(err: cpal::StreamError) {
    log::error!("Native listening output stream error: {}", err);
}

fn start_live_monitor_stream(
    config: NativeCaptureConfig,
) -> AudioResult<NativeListeningThreadState> {
    let host = cpal::default_host();
    let input_device = select_input_device(&host, config.preferred_input_device.as_deref())?;
    let input_device_name = input_device.description().ok().map(|d| d.to_string());
    let input_config = input_device
        .default_input_config()
        .map_err(|err| AudioError::new(format!("Failed to read input device config: {}", err)))?;
    let output_device = host
        .default_output_device()
        .ok_or_else(|| AudioError::new("No output device available"))?;
    let output_device_name = output_device.description().ok().map(|d| d.to_string());
    let output_config = output_device
        .default_output_config()
        .map_err(|err| AudioError::new(format!("Failed to read output device config: {}", err)))?;

    let output_sample_rate = output_config.sample_rate();
    let output_channels = output_config.channels().max(1);
    let input_channels = select_stream_config(
        &input_device,
        output_sample_rate,
        config.target_channels.max(1),
    )
    .map(|(_, channels)| channels)
    .ok_or_else(|| {
        AudioError::new(format!(
            "Input device does not support {} Hz for live listening",
            output_sample_rate
        ))
    })?;

    let cpal_input_config = StreamConfig {
        channels: input_channels,
        sample_rate: output_sample_rate,
        buffer_size: cpal::BufferSize::Default,
    };
    let cpal_output_config = StreamConfig {
        channels: output_channels,
        sample_rate: output_sample_rate,
        buffer_size: cpal::BufferSize::Default,
    };

    let max_buffer_samples = (output_sample_rate as usize * 2).max(4096);
    let (_, playback_buffer) = PlaybackSink::with_capacity(max_buffer_samples);
    let pipeline = Arc::new(Mutex::new(NativeCaptureBackend::build_live_pipeline(
        &config,
        output_sample_rate,
        input_channels,
        input_device_name.clone(),
        Arc::clone(&playback_buffer),
    )?));

    let input_stream = build_input_stream(
        &input_device,
        input_config.sample_format(),
        &cpal_input_config,
        input_channels,
        Arc::clone(&pipeline),
        native_listening_input_stream_error,
    )?;
    let output_stream = build_output_stream(
        &output_device,
        output_config.sample_format(),
        &cpal_output_config,
        Arc::clone(&playback_buffer),
        native_listening_output_stream_error,
    )?;

    input_stream
        .play()
        .map_err(|err| AudioError::new(format!("Failed to start input stream: {}", err)))?;
    output_stream
        .play()
        .map_err(|err| AudioError::new(format!("Failed to start output stream: {}", err)))?;

    Ok((
        input_stream,
        output_stream,
        pipeline,
        playback_buffer,
        Instant::now(),
        vec![
            format!(
                "input_device={}",
                input_device_name.unwrap_or_else(|| "unknown".into())
            ),
            format!(
                "output_device={}",
                output_device_name.unwrap_or_else(|| "unknown".into())
            ),
            format!("input_channels={}", input_channels),
            format!("output_channels={}", output_channels),
            format!("sample_rate={}", output_sample_rate),
        ],
    ))
}

fn rebuild_live_pipeline(
    pipeline: &Arc<Mutex<AudioPipeline>>,
    playback_buffer: &Arc<Mutex<PlaybackBuffer>>,
    config: NativeCaptureConfig,
) -> AudioResult<()> {
    if let Ok(mut buffer) = playback_buffer.lock() {
        buffer.clear();
    }

    let (pipeline_sample_rate, pipeline_channels, device_name) = {
        let current = pipeline
            .lock()
            .map_err(|_| AudioError::new("Audio pipeline lock poisoned"))?;
        (
            current.sample_rate(),
            current.input_channels(),
            current.device_name().map(ToOwned::to_owned),
        )
    };

    let rebuilt = NativeCaptureBackend::build_live_pipeline(
        &config,
        pipeline_sample_rate,
        pipeline_channels,
        device_name,
        Arc::clone(playback_buffer),
    )?;

    let mut current = pipeline
        .lock()
        .map_err(|_| AudioError::new("Audio pipeline lock poisoned"))?;
    *current = rebuilt;
    Ok(())
}

fn build_input_stream(
    device: &cpal::Device,
    sample_format: SampleFormat,
    config: &StreamConfig,
    input_channels: u16,
    pipeline: Arc<Mutex<AudioPipeline>>,
    err_fn: fn(cpal::StreamError),
) -> AudioResult<cpal::Stream> {
    let stream = match sample_format {
        SampleFormat::F32 => {
            let pipeline_for_stream = Arc::clone(&pipeline);
            device.build_input_stream(
                config,
                move |data: &[f32], _| {
                    process_stream_buffer(&pipeline_for_stream, data, input_channels)
                },
                err_fn,
                None,
            )
        }
        SampleFormat::I16 => {
            let pipeline_for_stream = Arc::clone(&pipeline);
            device.build_input_stream(
                config,
                move |data: &[i16], _| {
                    let converted = data
                        .iter()
                        .map(|sample| *sample as f32 / i16::MAX as f32)
                        .collect::<Vec<_>>();
                    process_stream_buffer(&pipeline_for_stream, &converted, input_channels)
                },
                err_fn,
                None,
            )
        }
        SampleFormat::U16 => {
            let pipeline_for_stream = Arc::clone(&pipeline);
            device.build_input_stream(
                config,
                move |data: &[u16], _| {
                    let converted = data
                        .iter()
                        .map(|sample| (*sample as f32 - 32768.0) / 32768.0)
                        .collect::<Vec<_>>();
                    process_stream_buffer(&pipeline_for_stream, &converted, input_channels)
                },
                err_fn,
                None,
            )
        }
        other => {
            return Err(AudioError::new(format!(
                "Unsupported input sample format: {:?}",
                other
            )))
        }
    }
    .map_err(|err| AudioError::new(format!("Failed to build input stream: {}", err)))?;

    Ok(stream)
}

fn build_output_stream(
    device: &cpal::Device,
    sample_format: SampleFormat,
    config: &StreamConfig,
    playback_buffer: Arc<Mutex<PlaybackBuffer>>,
    err_fn: fn(cpal::StreamError),
) -> AudioResult<cpal::Stream> {
    let output_channels = config.channels;
    let stream = match sample_format {
        SampleFormat::F32 => {
            let buffer = Arc::clone(&playback_buffer);
            device.build_output_stream(
                config,
                move |data: &mut [f32], _| fill_output_buffer_f32(data, output_channels, &buffer),
                err_fn,
                None,
            )
        }
        SampleFormat::I16 => {
            let buffer = Arc::clone(&playback_buffer);
            device.build_output_stream(
                config,
                move |data: &mut [i16], _| fill_output_buffer_i16(data, output_channels, &buffer),
                err_fn,
                None,
            )
        }
        SampleFormat::U16 => {
            let buffer = Arc::clone(&playback_buffer);
            device.build_output_stream(
                config,
                move |data: &mut [u16], _| fill_output_buffer_u16(data, output_channels, &buffer),
                err_fn,
                None,
            )
        }
        other => {
            return Err(AudioError::new(format!(
                "Unsupported output sample format: {:?}",
                other
            )))
        }
    }
    .map_err(|err| AudioError::new(format!("Failed to build output stream: {}", err)))?;

    Ok(stream)
}

fn fill_output_buffer_f32(
    data: &mut [f32],
    channels: u16,
    playback_buffer: &Arc<Mutex<PlaybackBuffer>>,
) {
    let channels = channels.max(1) as usize;
    if let Ok(mut buffer) = playback_buffer.lock() {
        for frame in data.chunks_mut(channels) {
            let sample = buffer.pop_mono_sample().unwrap_or(0.0).clamp(-1.0, 1.0);
            for out in frame {
                *out = sample;
            }
        }
    } else {
        data.fill(0.0);
    }
}

fn fill_output_buffer_i16(
    data: &mut [i16],
    channels: u16,
    playback_buffer: &Arc<Mutex<PlaybackBuffer>>,
) {
    let channels = channels.max(1) as usize;
    if let Ok(mut buffer) = playback_buffer.lock() {
        for frame in data.chunks_mut(channels) {
            let sample =
                (buffer.pop_mono_sample().unwrap_or(0.0).clamp(-1.0, 1.0) * i16::MAX as f32) as i16;
            for out in frame {
                *out = sample;
            }
        }
    } else {
        data.fill(0);
    }
}

fn fill_output_buffer_u16(
    data: &mut [u16],
    channels: u16,
    playback_buffer: &Arc<Mutex<PlaybackBuffer>>,
) {
    let channels = channels.max(1) as usize;
    if let Ok(mut buffer) = playback_buffer.lock() {
        for frame in data.chunks_mut(channels) {
            let sample = ((buffer.pop_mono_sample().unwrap_or(0.0).clamp(-1.0, 1.0) * 32767.0)
                + 32768.0)
                .round()
                .clamp(0.0, u16::MAX as f32) as u16;
            for out in frame {
                *out = sample;
            }
        }
    } else {
        data.fill(u16::MIN);
    }
}

fn select_input_device(
    host: &cpal::Host,
    preferred_name: Option<&str>,
) -> AudioResult<cpal::Device> {
    #[cfg(target_os = "linux")]
    {
        if let Some(preferred_name) = preferred_name {
            if let Err(err) = set_default_input_device(preferred_name) {
                log::warn!(
                    "Failed to set preferred PipeWire source '{}' as default: {}",
                    preferred_name,
                    err
                );
            }
        }

        if let Some(device) = find_named_input_device(host, "pipewire")? {
            return Ok(device);
        }
    }

    if let Some(preferred_name) = preferred_name {
        let devices = host.input_devices().map_err(|err| {
            AudioError::new(format!("Failed to enumerate input devices: {}", err))
        })?;
        for device in devices {
            if device.description().ok().map(|d| d.to_string()).as_deref() == Some(preferred_name) {
                return Ok(device);
            }
        }
        log::warn!(
            "Preferred input device '{}' not found, falling back to system default",
            preferred_name
        );
    }

    host.default_input_device()
        .ok_or_else(|| AudioError::new("No input device available"))
}

fn find_named_input_device(
    host: &cpal::Host,
    target_name: &str,
) -> AudioResult<Option<cpal::Device>> {
    let devices = host
        .input_devices()
        .map_err(|err| AudioError::new(format!("Failed to enumerate input devices: {}", err)))?;
    for device in devices {
        if device
            .description()
            .ok()
            .is_some_and(|name| name.to_string().eq_ignore_ascii_case(target_name))
        {
            return Ok(Some(device));
        }
    }
    Ok(None)
}

fn process_stream_buffer(pipeline: &Arc<Mutex<AudioPipeline>>, data: &[f32], channels: u16) {
    if let Ok(mut pipeline) = pipeline.lock() {
        if let Err(err) = pipeline.process_input(data, channels) {
            log::error!("Audio pipeline processing failed: {}", err);
        }
    }
}

fn select_stream_config(
    device: &cpal::Device,
    target_sample_rate: u32,
    target_channels: u16,
) -> Option<(u32, u16)> {
    let supported_configs = device.supported_input_configs().ok()?;
    let mut best_match = None;
    for cfg in supported_configs {
        let min_rate = cfg.min_sample_rate();
        let max_rate = cfg.max_sample_rate();
        let channels = cfg.channels();
        if target_sample_rate >= min_rate
            && target_sample_rate <= max_rate
            && channels == target_channels
        {
            return Some((target_sample_rate, target_channels));
        }
        if best_match.is_none() && target_sample_rate >= min_rate && target_sample_rate <= max_rate
        {
            best_match = Some((target_sample_rate, channels));
        }
    }
    best_match
}

struct NativeRecordingHandle {
    cmd_tx: mpsc::Sender<NativeCommand>,
    join_handle: Option<JoinHandle<()>>,
}

struct NativeListeningHandle {
    cmd_tx: mpsc::Sender<NativeListeningCommand>,
    join_handle: Option<JoinHandle<()>>,
}

impl ActiveRecording for NativeRecordingHandle {
    fn stop(&mut self) -> AudioResult<CaptureDiagnostics> {
        let (reply_tx, reply_rx) = mpsc::channel();
        self.cmd_tx
            .send(NativeCommand::Stop(reply_tx))
            .map_err(|_| AudioError::new("Native capture thread is no longer running"))?;
        let diagnostics = reply_rx
            .recv()
            .map_err(|_| AudioError::new("No response from native capture thread"))??;
        if let Some(handle) = self.join_handle.take() {
            let _ = handle.join();
        }
        Ok(diagnostics)
    }
}

impl ActiveListening for NativeListeningHandle {
    fn stop(&mut self) -> AudioResult<CaptureDiagnostics> {
        let (reply_tx, reply_rx) = mpsc::channel();
        self.cmd_tx
            .send(NativeListeningCommand::Stop(reply_tx))
            .map_err(|_| AudioError::new("Native listening thread is no longer running"))?;
        let diagnostics = reply_rx
            .recv()
            .map_err(|_| AudioError::new("No response from native listening thread"))??;
        if let Some(handle) = self.join_handle.take() {
            let _ = handle.join();
        }
        Ok(diagnostics)
    }

    fn update_config(&mut self, config: NativeCaptureConfig) -> AudioResult<()> {
        let (reply_tx, reply_rx) = mpsc::channel();
        self.cmd_tx
            .send(NativeListeningCommand::UpdateConfig(config, reply_tx))
            .map_err(|_| AudioError::new("Native listening thread is no longer running"))?;
        reply_rx
            .recv()
            .map_err(|_| AudioError::new("No response from native listening thread"))??;
        Ok(())
    }
}