audio-io-bsd 0.2.0

//! Concrete [`AudioBackend`] backed by [`cpal`].
//!
//! [`CpalBackend`] reaches the kernel audio layer through cpal's host API. On
//! FreeBSD this is the ALSA backend (FreeBSD's `alsa-lib` port translates ALSA
//! calls onto OSS `/dev/dsp`); on Linux it is native ALSA; on macOS `CoreAudio`.
//!
//! ## Real-time safety model
//!
//! Output follows the canonical RT-safe pattern for callback-driven audio:
//!
//! 1. [`CpalSink::write`] interleaves a planar [`AudioFrame`] into a
//!    **pre-allocated** scratch buffer, then pushes each sample into a
//!    lock-free [`rtrb`] ring. This is wait-free and alloc-free (after the first
//!    write stabilises the scratch capacity), so `write` is safe to call from
//!    the real-time audio thread.
//! 2. A dedicated **stream-owner thread** (spawned at open time) builds and
//!    plays the cpal stream; cpal's own audio-thread callback **pops** samples
//!    from the ring into the device buffer. The heavy device I/O happens there,
//!    never on the caller's RT thread.
//! 3. When the caller writes faster than the device consumes, the ring fills
//!    and [`CpalSink::write`] reports [`IoError::StreamSetup`] (transient
//!    back-pressure) rather than blocking or allocating.
//!
//! Capture ([`CpalSource::read`]) is intended for a **worker thread**; its
//! callback pushes captured samples into a ring that `read` drains.
//!
//! ### Why a stream-owner thread?
//!
//! `cpal::Stream` is `!Send` on most platforms (the underlying audio handle is
//! thread-affine). Since [`OutputSink`] / [`InputSource`] are `Send` (a host may
//! move a sink between its control and RT threads), the cpal stream cannot live
//! inside the sink. Instead the stream is owned by a dedicated thread that
//! outlives `build_*_stream` and is torn down when the sink/source is dropped.
//!
//! > **Version note:** this module pins cpal to **0.15** because cpal 0.18's
//! > ALSA backend does not compile on FreeBSD (`libc::ESTRPIPE` is Linux-only
//! > and `frames_to_duration` import fails — a cpal upstream issue). cpal 0.15
//! > builds cleanly on FreeBSD 15.1/14.2 with `alsa-lib` + `pkgconf`.
//!
//! [`cpal`]: https://crates.io/crates/cpal

use std::sync::mpsc;

use audio_core_bsd::AudioFrame;

use crate::backend::{AudioBackend, InputSource, OutputSink};
use crate::device::{BufferSize, DeviceDirection, DeviceInfo, StreamParams};
use crate::error::{IoError, Result};

// cpal's host/device/stream APIs are split across extension traits — bring them
// into scope so methods like `default_host`, `output_devices`, `name`,
// `supported_output_configs`, `build_output_stream` and `play` resolve.
use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};

/// Number of interleaved f32 samples the ring holds before back-pressure.
//  65536 samples ≈ 0.68 s @ 48 kHz stereo — ample jitter buffer for one cycle.
const RING_CAPACITY: usize = 1 << 16;

/// `cpal`-backed audio backend (ALSA on Linux/FreeBSD, `CoreAudio` on macOS).
///
/// Construct with [`CpalBackend::new`] (default host) or
/// [`CpalBackend::with_host`] (inject a specific cpal host, e.g. for tests).
pub struct CpalBackend {
    host: cpal::Host,
}

impl CpalBackend {
    /// Create a backend over the platform default cpal host.
    ///
    /// # Errors
    ///
    /// Returns [`IoError::Backend`] if cpal has no default host on this platform.
    pub fn new() -> Result<Self> {
        Ok(Self {
            host: cpal::default_host(),
        })
    }

    /// Create a backend over an explicit cpal host (useful for tests or when a
    /// non-default host — e.g. JACK — is desired).
    #[must_use]
    pub fn with_host(host: cpal::Host) -> Self {
        Self { host }
    }

    /// Borrow the underlying cpal host.
    #[must_use]
    pub fn host(&self) -> &cpal::Host {
        &self.host
    }
}

/// Build a cpal `StreamConfig` from validated [`StreamParams`].
fn stream_config(params: StreamParams) -> cpal::StreamConfig {
    cpal::StreamConfig {
        channels: params.channels,
        sample_rate: cpal::SampleRate(params.sample_rate),
        buffer_size: match params.buffer_size {
            BufferSize::Fixed(n) => cpal::BufferSize::Fixed(u32::try_from(n).unwrap_or(u32::MAX)),
            BufferSize::Default => cpal::BufferSize::Default,
        },
    }
}

/// Find a cpal device (input or output) whose `name()` matches `dev`.
fn find_device(host: &cpal::Host, dev: &str, output: bool) -> Result<cpal::Device> {
    let mut devices = if output {
        host.output_devices()
            .map_err(|e| IoError::Backend(e.to_string()))?
    } else {
        host.input_devices()
            .map_err(|e| IoError::Backend(e.to_string()))?
    };
    devices
        .find(|d| d.name().is_ok_and(|n| n == dev))
        .ok_or_else(|| IoError::DeviceNotFound(dev.into()))
}

/// Probe whether `device` supports `params` at the requested sample rate.
fn check_supported(device: &cpal::Device, params: StreamParams, output: bool) -> Result<()> {
    let target = cpal::SampleRate(params.sample_rate);
    let matches = |c: cpal::SupportedStreamConfigRange| {
        c.channels() == params.channels
            && c.min_sample_rate() <= target
            && target <= c.max_sample_rate()
    };
    let ok = if output {
        device
            .supported_output_configs()
            .map_err(|e| IoError::UnsupportedConfig(e.to_string()))?
            .any(matches)
    } else {
        device
            .supported_input_configs()
            .map_err(|e| IoError::UnsupportedConfig(e.to_string()))?
            .any(matches)
    };
    if ok {
        Ok(())
    } else {
        Err(IoError::UnsupportedConfig(format!(
            "no {} config for {}ch @ {} Hz",
            if output { "output" } else { "input" },
            params.channels,
            params.sample_rate
        )))
    }
}

/// Convert a cpal device to the platform-agnostic [`DeviceInfo`].
fn device_info(device: &cpal::Device, direction: DeviceDirection, is_default: bool) -> DeviceInfo {
    let name = device.name().unwrap_or_else(|_| "<unknown>".into());
    // Probe the device's channel count and sample-rate range from the first
    // supported config; fall back to sensible defaults if probing fails.
    let (channels, rates) = device
        .supported_output_configs()
        .ok()
        .and_then(|mut c| c.next())
        .map_or((2, vec![48_000]), |first| {
            (
                first.channels(),
                vec![first.min_sample_rate().0, first.max_sample_rate().0],
            )
        });
    DeviceInfo::new(name, direction, channels, rates, is_default)
}

impl AudioBackend for CpalBackend {
    fn enumerate_devices(&self) -> Vec<DeviceInfo> {
        let default_out = self.host.default_output_device();
        let default_in = self.host.default_input_device();
        let mut out = Vec::new();
        if let Ok(devs) = self.host.output_devices() {
            for d in devs {
                let is_default = default_out
                    .as_ref()
                    .is_some_and(|def| d.name().is_ok_and(|n| def.name().is_ok_and(|m| n == m)));
                out.push(device_info(&d, DeviceDirection::Output, is_default));
            }
        }
        if let Ok(devs) = self.host.input_devices() {
            for d in devs {
                let is_default = default_in
                    .as_ref()
                    .is_some_and(|def| d.name().is_ok_and(|n| def.name().is_ok_and(|m| n == m)));
                out.push(device_info(&d, DeviceDirection::Input, is_default));
            }
        }
        out
    }

    fn default_output(&self) -> Option<DeviceInfo> {
        self.host
            .default_output_device()
            .as_ref()
            .map(|d| device_info(d, DeviceDirection::Output, true))
    }

    fn default_input(&self) -> Option<DeviceInfo> {
        self.host
            .default_input_device()
            .as_ref()
            .map(|d| device_info(d, DeviceDirection::Input, true))
    }

    fn open_output(&self, dev: &str, params: StreamParams) -> Result<Box<dyn OutputSink>> {
        params.validate()?;
        let device = find_device(&self.host, dev, true)?;
        check_supported(&device, params, true)?;
        let cfg = stream_config(params);
        let channels = cfg.channels;

        let (producer, consumer) = rtrb::RingBuffer::<f32>::new(RING_CAPACITY);
        let (drop_tx, drop_rx) = mpsc::channel::<()>();
        let (ready_tx, ready_rx) = mpsc::channel::<Result<()>>();

        // Stream-owner thread: cpal::Stream is !Send, so it must live on one
        // thread for its entire lifetime. This thread builds, plays, and owns
        // the stream until the sink is dropped (drop_rx fires).
        std::thread::Builder::new()
            .name("audio-io-output".into())
            .spawn(move || {
                let mut consumer = consumer;
                let stream = device.build_output_stream::<f32, _, _>(
                    &cfg,
                    move |data: &mut [f32], _info| {
                        // RT audio thread: pop interleaved samples from the ring.
                        // A drained ring yields silence (0.0) — a benign dropout.
                        for slot in data.iter_mut() {
                            *slot = consumer.pop().unwrap_or(0.0);
                        }
                    },
                    |_err| {
                        // Stream errors are non-fatal dropouts; reported out-of-band.
                    },
                    None,
                );
                let stream = match stream {
                    Ok(s) => s,
                    Err(e) => {
                        let _ = ready_tx.send(Err(IoError::StreamSetup(e.to_string())));
                        return;
                    }
                };
                if let Err(e) = stream.play() {
                    let _ = ready_tx.send(Err(IoError::StreamSetup(e.to_string())));
                    return;
                }
                let _ = ready_tx.send(Ok(()));
                // Block until the sink is dropped, keeping the stream alive.
                let _ = drop_rx.recv();
                // `stream` dropped here → cpal stops the callback.
            })
            .map_err(|e| IoError::Backend(format!("spawn output thread: {e}")))?;

        // Wait for the stream-owner thread to confirm the stream is live.
        ready_rx
            .recv()
            .map_err(|e| IoError::StreamSetup(format!("output thread panicked: {e}")))??;

        Ok(Box::new(CpalSink {
            producer,
            scratch: Vec::with_capacity(4 * channels as usize),
            channels,
            drop_tx: Some(drop_tx),
        }))
    }

    fn open_input(&self, dev: &str, params: StreamParams) -> Result<Box<dyn InputSource>> {
        params.validate()?;
        let device = find_device(&self.host, dev, false)?;
        check_supported(&device, params, false)?;
        let cfg = stream_config(params);
        let channels = cfg.channels;
        let sample_rate = cfg.sample_rate.0;

        let (producer, consumer) = rtrb::RingBuffer::<f32>::new(RING_CAPACITY);
        let (drop_tx, drop_rx) = mpsc::channel::<()>();
        let (ready_tx, ready_rx) = mpsc::channel::<Result<()>>();

        std::thread::Builder::new()
            .name("audio-io-input".into())
            .spawn(move || {
                let mut producer = producer;
                let stream = device.build_input_stream::<f32, _, _>(
                    &cfg,
                    move |data: &[f32], _info| {
                        // Capture callback (worker side): push interleaved samples.
                        for &s in data {
                            let _ = producer.push(s);
                        }
                    },
                    |_err| {},
                    None,
                );
                let stream = match stream {
                    Ok(s) => s,
                    Err(e) => {
                        let _ = ready_tx.send(Err(IoError::StreamSetup(e.to_string())));
                        return;
                    }
                };
                if let Err(e) = stream.play() {
                    let _ = ready_tx.send(Err(IoError::StreamSetup(e.to_string())));
                    return;
                }
                let _ = ready_tx.send(Ok(()));
                let _ = drop_rx.recv();
            })
            .map_err(|e| IoError::Backend(format!("spawn input thread: {e}")))?;

        ready_rx
            .recv()
            .map_err(|e| IoError::StreamSetup(format!("input thread panicked: {e}")))??;

        Ok(Box::new(CpalSource {
            consumer,
            scratch: Vec::with_capacity(4 * channels as usize),
            channels,
            sample_rate,
            drop_tx: Some(drop_tx),
        }))
    }
}

/// Output sink backed by a cpal output stream (owned on a dedicated thread) and
/// a lock-free ring.
///
/// `write` interleaves the planar [`AudioFrame`] into a pre-allocated scratch
/// buffer and pushes samples into the ring; the cpal callback drains it.
/// Dropping the sink signals the stream-owner thread to tear the stream down.
pub struct CpalSink {
    producer: rtrb::Producer<f32>,
    /// Pre-allocated interleaving scratch — reused across writes (`clear()` keeps
    /// capacity, so no per-write allocation once stabilised).
    scratch: Vec<f32>,
    channels: cpal::ChannelCount,
    /// Stop signal for the stream-owner thread (`Some` while the stream is live).
    drop_tx: Option<mpsc::Sender<()>>,
}

impl OutputSink for CpalSink {
    fn write(&mut self, frame: &AudioFrame) -> Result<()> {
        let n = frame.num_frames();
        let ch = (self.channels.min(frame.channels)) as usize;
        if ch == 0 {
            return Ok(());
        }
        // Reuse the scratch buffer (clear keeps capacity → no realloc once warm).
        self.scratch.clear();
        self.scratch.reserve(n * ch);
        // Interleave planar [ch0.., ch1..] → interleaved [s0c0, s0c1, ...].
        for i in 0..n {
            for c in 0..ch {
                let v = frame.channel_slice(c).get(i).copied().unwrap_or(0.0);
                self.scratch.push(v);
            }
        }
        // Push to the ring (wait-free). A full ring is back-pressure, not a fault.
        for &s in &self.scratch {
            if self.producer.push(s).is_err() {
                return Err(IoError::StreamSetup(
                    "output ring full (back-pressure)".into(),
                ));
            }
        }
        Ok(())
    }
}

impl Drop for CpalSink {
    fn drop(&mut self) {
        // Signal the stream-owner thread to stop (drop the cpal::Stream).
        self.drop_tx.take();
    }
}

/// Input source backed by a cpal input stream (owned on a dedicated thread) and
/// a lock-free ring.
pub struct CpalSource {
    consumer: rtrb::Consumer<f32>,
    scratch: Vec<f32>,
    channels: cpal::ChannelCount,
    sample_rate: u32,
    drop_tx: Option<mpsc::Sender<()>>,
}

impl InputSource for CpalSource {
    fn read(&mut self) -> Result<AudioFrame> {
        // Drain available interleaved samples into the scratch buffer.
        self.scratch.clear();
        while let Ok(s) = self.consumer.pop() {
            self.scratch.push(s);
        }
        let ch = self.channels as usize;
        let n = self.scratch.len().checked_div(ch).unwrap_or(0);
        // De-interleave interleaved [s0c0, s0c1, ...] → planar [ch0.., ch1..].
        let mut planar = vec![0.0_f32; ch * n];
        for (i, &v) in self.scratch.iter().enumerate().take(ch * n) {
            let frame = i / ch;
            let channel = i % ch;
            planar[channel * n + frame] = v;
        }
        Ok(AudioFrame::from_planar(
            self.channels,
            self.sample_rate,
            planar,
        ))
    }
}

impl Drop for CpalSource {
    fn drop(&mut self) {
        self.drop_tx.take();
    }
}