//! ALSA backend implementation.
//!
//! Default backend on Linux and BSD systems.

extern crate alsa;
extern crate libc;

use std::{
    cmp,
    sync::{
        atomic::{AtomicBool, AtomicUsize, Ordering},
        Arc,
    },
    thread::{self, JoinHandle},
    time::Duration,
    vec::IntoIter as VecIntoIter,
};

use self::alsa::poll::Descriptors;
pub use self::enumerate::Devices;

use crate::{
    host::fill_with_equilibrium,
    iter::{SupportedInputConfigs, SupportedOutputConfigs},
    traits::{DeviceTrait, HostTrait, StreamTrait},
    BackendSpecificError, BufferSize, BuildStreamError, ChannelCount, Data,
    DefaultStreamConfigError, DeviceDescription, DeviceDescriptionBuilder, DeviceDirection,
    DeviceId, DeviceIdError, DeviceNameError, DevicesError, FrameCount, InputCallbackInfo,
    OutputCallbackInfo, PauseStreamError, PlayStreamError, SampleFormat, SampleRate, StreamConfig,
    StreamError, SupportedBufferSize, SupportedStreamConfig, SupportedStreamConfigRange,
    SupportedStreamConfigsError,
};

mod enumerate;

// ALSA Buffer Size Behavior
// =========================
//
// ## ALSA Latency Model
//
// **Hardware vs Software Buffer**: ALSA maintains a software buffer in memory that feeds
// a hardware buffer in the audio device. Audio latency is determined by how much data
// sits in the software buffer before being transferred to hardware.
//
// **Period-Based Transfer**: ALSA transfers data in chunks called "periods". When one
// period worth of data has been consumed by hardware, ALSA triggers a callback to refill
// that period in the software buffer.
//
// ## BufferSize::Fixed Behavior
//
// When `BufferSize::Fixed(x)` is specified, cpal attempts to configure the period size
// to approximately `x` frames to achieve the requested callback size. However, the
// actual callback size may differ from the request:
//
// - ALSA may round the period size to hardware-supported values
// - Different devices have different period size constraints
// - The callback size is not guaranteed to exactly match the request
// - If the requested size cannot be accommodated, ALSA will choose the nearest
//   supported configuration
//
// This mirrors the behavior documented in the cpal API where `BufferSize::Fixed(x)`
// requests but does not guarantee a specific callback size.
//
// ## BufferSize::Default Behavior
//
// When `BufferSize::Default` is specified, cpal does NOT set explicit period size or
// period count constraints, allowing the device/driver to choose sensible defaults.
//
// **Why not set defaults?** Different audio systems have different behaviors:
//
// - **Native ALSA hardware**: Typically chooses reasonable defaults (e.g., 512-2048
//   frame periods with 2-4 periods)
//
// - **PipeWire-ALSA plugin**: Allocates a large ring buffer (~1M frames at 48kHz) but
//   uses small periods (512-1024 frames). Critically, if you request `set_periods(2)`
//   without specifying period size, PipeWire calculates period = buffer/2, resulting
//   in pathologically large periods (~524K frames = 10 seconds). See issues #1029 and
//   #1036.
//
// By not constraining period configuration, PipeWire-ALSA can use its optimized defaults
// (small periods with many-period buffer), while native ALSA hardware uses its own defaults.
//
// **Startup latency**: Regardless of buffer size, cpal uses double-buffering for startup
// (start_threshold = 2 periods), ensuring low latency even with large multi-period ring
// buffers.

const DEFAULT_DEVICE: &str = "default";

// Some ALSA plugins (e.g. alsaequal, certain USB drivers) are not reentrant.
static ALSA_OPEN_MUTEX: std::sync::Mutex<()> = std::sync::Mutex::new(());

// TODO: Not yet defined in rust-lang/libc crate
const LIBC_ENOTSUPP: libc::c_int = 524;

/// The default Linux and BSD host type.
#[derive(Debug, Clone)]
pub struct Host {
    inner: Arc<AlsaContext>,
}

impl Host {
    pub fn new() -> Result<Self, crate::HostUnavailable> {
        let inner = AlsaContext::new().map_err(|_| crate::HostUnavailable)?;
        Ok(Host {
            inner: Arc::new(inner),
        })
    }
}

impl HostTrait for Host {
    type Devices = Devices;
    type Device = Device;

    fn is_available() -> bool {
        // Assume ALSA is always available on Linux and BSD.
        true
    }

    fn devices(&self) -> Result<Self::Devices, DevicesError> {
        self.enumerate_devices()
    }

    fn device_by_id(&self, id: &crate::DeviceId) -> Option<Self::Device> {
        let canonical_id = crate::DeviceId(id.0, canonical_pcm_id(&id.1));
        self.devices()
            .ok()?
            .find(|d| d.id().ok().as_ref() == Some(&canonical_id))
    }

    fn default_input_device(&self) -> Option<Self::Device> {
        Some(Device::default())
    }

    fn default_output_device(&self) -> Option<Self::Device> {
        Some(Device::default())
    }
}

/// Global count of active ALSA context instances.
static ALSA_CONTEXT_COUNT: AtomicUsize = AtomicUsize::new(0);

/// ALSA backend context shared between `Host`, `Device`, and `Stream` via `Arc`.
#[derive(Debug)]
pub(super) struct AlsaContext;

impl AlsaContext {
    fn new() -> Result<Self, alsa::Error> {
        // Initialize global ALSA config cache on first context creation.
        if ALSA_CONTEXT_COUNT.fetch_add(1, Ordering::SeqCst) == 0 {
            alsa::config::update()?;
        }
        Ok(Self)
    }
}

impl Drop for AlsaContext {
    fn drop(&mut self) {
        // Free the global ALSA config cache when the last context is dropped.
        if ALSA_CONTEXT_COUNT.fetch_sub(1, Ordering::SeqCst) == 1 {
            let _ = alsa::config::update_free_global();
        }
    }
}

impl DeviceTrait for Device {
    type SupportedInputConfigs = SupportedInputConfigs;
    type SupportedOutputConfigs = SupportedOutputConfigs;
    type Stream = Stream;

    // ALSA overrides name() to return pcm_id directly instead of from description
    fn name(&self) -> Result<String, DeviceNameError> {
        Device::name(self)
    }

    fn description(&self) -> Result<DeviceDescription, DeviceNameError> {
        Device::description(self)
    }

    fn id(&self) -> Result<DeviceId, DeviceIdError> {
        Device::id(self)
    }

    // Override trait defaults to avoid opening devices during enumeration.
    //
    // ALSA does not guarantee transactional cleanup on failed snd_pcm_open(). Opening plugins like
    // alsaequal that fail with EPERM can leak FDs, poisoning the ALSA backend for the process
    // lifetime (subsequent device opens fail with EBUSY until process exit).
    fn supports_input(&self) -> bool {
        matches!(
            self.direction,
            DeviceDirection::Input | DeviceDirection::Duplex
        )
    }

    fn supports_output(&self) -> bool {
        matches!(
            self.direction,
            DeviceDirection::Output | DeviceDirection::Duplex
        )
    }

    fn supported_input_configs(
        &self,
    ) -> Result<Self::SupportedInputConfigs, SupportedStreamConfigsError> {
        Device::supported_input_configs(self)
    }

    fn supported_output_configs(
        &self,
    ) -> Result<Self::SupportedOutputConfigs, SupportedStreamConfigsError> {
        Device::supported_output_configs(self)
    }

    fn default_input_config(&self) -> Result<SupportedStreamConfig, DefaultStreamConfigError> {
        Device::default_input_config(self)
    }

    fn default_output_config(&self) -> Result<SupportedStreamConfig, DefaultStreamConfigError> {
        Device::default_output_config(self)
    }

    fn build_input_stream_raw<D, E>(
        &self,
        conf: StreamConfig,
        sample_format: SampleFormat,
        data_callback: D,
        error_callback: E,
        timeout: Option<Duration>,
    ) -> Result<Self::Stream, BuildStreamError>
    where
        D: FnMut(&Data, &InputCallbackInfo) + Send + 'static,
        E: FnMut(StreamError) + Send + 'static,
    {
        let stream_inner =
            self.build_stream_inner(conf, sample_format, alsa::Direction::Capture)?;
        let stream = Self::Stream::new_input(
            Arc::new(stream_inner),
            data_callback,
            error_callback,
            timeout,
        );
        Ok(stream)
    }

    fn build_output_stream_raw<D, E>(
        &self,
        conf: StreamConfig,
        sample_format: SampleFormat,
        data_callback: D,
        error_callback: E,
        timeout: Option<Duration>,
    ) -> Result<Self::Stream, BuildStreamError>
    where
        D: FnMut(&mut Data, &OutputCallbackInfo) + Send + 'static,
        E: FnMut(StreamError) + Send + 'static,
    {
        let stream_inner =
            self.build_stream_inner(conf, sample_format, alsa::Direction::Playback)?;
        let stream = Self::Stream::new_output(
            Arc::new(stream_inner),
            data_callback,
            error_callback,
            timeout,
        );
        Ok(stream)
    }
}

#[derive(Debug)]
struct TriggerSender(libc::c_int);

#[derive(Debug)]
struct TriggerReceiver(libc::c_int);

impl TriggerSender {
    fn wakeup(&self) {
        let buf = 1u64;
        loop {
            let ret = unsafe { libc::write(self.0, &buf as *const u64 as *const _, 8) };
            if ret == 8 {
                return;
            }
            // write() can be interrupted by a signal before writing any bytes; retry.
            assert_eq!(ret, -1, "wakeup: unexpected return value {ret}");
            if std::io::Error::last_os_error().kind() != std::io::ErrorKind::Interrupted {
                panic!("wakeup: {}", std::io::Error::last_os_error());
            }
        }
    }
}

impl TriggerReceiver {
    fn clear_pipe(&self) {
        let mut out = 0u64;
        loop {
            let ret = unsafe { libc::read(self.0, &mut out as *mut u64 as *mut _, 8) };
            if ret == 8 {
                return;
            }
            // read() can be interrupted by a signal before reading any bytes; retry.
            assert_eq!(ret, -1, "clear_pipe: unexpected return value {ret}");
            if std::io::Error::last_os_error().kind() != std::io::ErrorKind::Interrupted {
                panic!("clear_pipe: {}", std::io::Error::last_os_error());
            }
        }
    }
}

fn trigger() -> (TriggerSender, Arc<TriggerReceiver>) {
    let mut fds = [0, 0];
    match unsafe { libc::pipe(fds.as_mut_ptr()) } {
        0 => (TriggerSender(fds[1]), Arc::new(TriggerReceiver(fds[0]))),
        _ => panic!("Could not create pipe"),
    }
}

impl Drop for TriggerSender {
    fn drop(&mut self) {
        unsafe {
            libc::close(self.0);
        }
    }
}

impl Drop for TriggerReceiver {
    fn drop(&mut self) {
        unsafe {
            libc::close(self.0);
        }
    }
}

#[derive(Clone, Debug)]
pub struct Device {
    pcm_id: String,
    desc: Option<String>,
    direction: DeviceDirection,
    _context: Arc<AlsaContext>,
}

impl PartialEq for Device {
    fn eq(&self, other: &Self) -> bool {
        self.pcm_id == other.pcm_id
    }
}

impl Eq for Device {}

impl std::hash::Hash for Device {
    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
        self.pcm_id.hash(state);
    }
}

impl Device {
    fn build_stream_inner(
        &self,
        conf: StreamConfig,
        sample_format: SampleFormat,
        stream_type: alsa::Direction,
    ) -> Result<StreamInner, BuildStreamError> {
        // Validate buffer size if Fixed is specified. This is necessary because
        // `set_period_size_near()` with `ValueOr::Nearest` will accept ANY value and return the
        // "nearest" supported value, which could be wildly different (e.g., requesting 4096 frames
        // might return 512 frames if that's "nearest").
        if let BufferSize::Fixed(requested_size) = conf.buffer_size {
            // Note: We use `default_input_config`/`default_output_config` to get the buffer size
            // range. This queries the CURRENT device (`self.pcm_id`), not the default device. The
            // buffer size range is the same across all format configurations for a given device
            // (see `supported_configs()`).
            let supported_config = match stream_type {
                alsa::Direction::Capture => self.default_input_config(),
                alsa::Direction::Playback => self.default_output_config(),
            };
            if let Ok(config) = supported_config {
                if let SupportedBufferSize::Range { min, max } = config.buffer_size {
                    if !(min..=max).contains(&requested_size) {
                        return Err(BuildStreamError::StreamConfigNotSupported);
                    }
                }
            }
        }

        let open_result = {
            let _guard = ALSA_OPEN_MUTEX.lock().unwrap_or_else(|e| e.into_inner());
            alsa::pcm::PCM::new(&self.pcm_id, stream_type, true).map_err(|e| (e, e.errno()))
        };
        let handle = match open_result {
            Err((_, libc::ENOENT))
            | Err((_, libc::EPERM))
            | Err((_, libc::ENODEV))
            | Err((_, LIBC_ENOTSUPP)) => return Err(BuildStreamError::DeviceNotAvailable),
            Err((_, libc::EBUSY)) | Err((_, libc::EAGAIN)) => {
                return Err(BuildStreamError::DeviceBusy)
            }
            Err((_, libc::EINVAL)) => return Err(BuildStreamError::InvalidArgument),
            Err((e, _)) => return Err(e.into()),
            Ok(handle) => handle,
        };

        let can_pause = set_hw_params_from_format(&handle, conf, sample_format)?;
        let period_samples = set_sw_params_from_format(&handle, conf, stream_type)?;

        handle.prepare()?;

        let num_descriptors = handle.count();
        if num_descriptors == 0 {
            let description = "poll descriptor count for stream was 0".to_string();
            let err = BackendSpecificError { description };
            return Err(err.into());
        }

        // Check to see if we can retrieve valid timestamps from the device.
        // Related: https://bugs.freedesktop.org/show_bug.cgi?id=88503
        let ts = handle.status()?.get_htstamp();
        let creation_instant = std::time::Instant::now();
        let use_hw_timestamps = !(ts.tv_sec == 0 && ts.tv_nsec == 0);

        if let alsa::Direction::Capture = stream_type {
            handle.start()?;
        }

        // Pre-compute a period-sized buffer filled with silence values.
        let period_frames = period_samples / conf.channels as usize;
        let frame_size = sample_format.sample_size() * conf.channels as usize;
        let period_bytes = period_frames * frame_size;
        let mut silence_template = vec![0u8; period_bytes].into_boxed_slice();

        // Only fill buffer for unsigned formats that don't have a zero value for silence.
        if sample_format.is_uint() {
            fill_with_equilibrium(&mut silence_template, sample_format);
        }

        let stream_inner = StreamInner {
            dropping: AtomicBool::new(false),
            channel: handle,
            sample_format,
            num_descriptors,
            conf,
            period_samples,
            period_frames,
            frame_size,
            silence_template,
            can_pause,
            creation_instant,
            use_hw_timestamps,
            _context: self._context.clone(),
        };

        Ok(stream_inner)
    }

    fn name(&self) -> Result<String, DeviceNameError> {
        Ok(self.pcm_id.clone())
    }

    fn description(&self) -> Result<DeviceDescription, DeviceNameError> {
        let name = self
            .desc
            .as_ref()
            .and_then(|desc| desc.lines().next())
            .unwrap_or(&self.pcm_id)
            .to_string();

        let mut builder = DeviceDescriptionBuilder::new(name)
            .driver(self.pcm_id.clone())
            .direction(self.direction);

        if let Some(ref desc) = self.desc {
            let lines = desc
                .lines()
                .map(|line| line.trim().to_string())
                .filter(|line| !line.is_empty())
                .collect();
            builder = builder.extended(lines);
        }

        Ok(builder.build())
    }

    fn id(&self) -> Result<DeviceId, DeviceIdError> {
        Ok(DeviceId(crate::platform::HostId::Alsa, self.pcm_id.clone()))
    }

    fn supported_configs(
        &self,
        stream_t: alsa::Direction,
    ) -> Result<VecIntoIter<SupportedStreamConfigRange>, SupportedStreamConfigsError> {
        let open_result = {
            let _guard = ALSA_OPEN_MUTEX.lock().unwrap_or_else(|e| e.into_inner());
            alsa::pcm::PCM::new(&self.pcm_id, stream_t, true).map_err(|e| (e, e.errno()))
        };
        let pcm = match open_result {
            Err((_, libc::ENOENT))
            | Err((_, libc::EPERM))
            | Err((_, libc::ENODEV))
            | Err((_, LIBC_ENOTSUPP)) => {
                return Err(SupportedStreamConfigsError::DeviceNotAvailable)
            }
            Err((_, libc::EBUSY)) | Err((_, libc::EAGAIN)) => {
                return Err(SupportedStreamConfigsError::DeviceBusy)
            }
            Err((_, libc::EINVAL)) => return Err(SupportedStreamConfigsError::InvalidArgument),
            Err((e, _)) => return Err(e.into()),
            Ok(pcm) => pcm,
        };

        let hw_params = alsa::pcm::HwParams::any(&pcm)?;

        // Test both LE and BE formats to detect what the hardware actually supports.
        // LE is listed first as it's the common case for most audio hardware.
        // Hardware reports its supported formats regardless of CPU endianness.
        const FORMATS: [(SampleFormat, alsa::pcm::Format); 23] = [
            (SampleFormat::I8, alsa::pcm::Format::S8),
            (SampleFormat::U8, alsa::pcm::Format::U8),
            (SampleFormat::I16, alsa::pcm::Format::S16LE),
            (SampleFormat::I16, alsa::pcm::Format::S16BE),
            (SampleFormat::U16, alsa::pcm::Format::U16LE),
            (SampleFormat::U16, alsa::pcm::Format::U16BE),
            (SampleFormat::I24, alsa::pcm::Format::S24LE),
            (SampleFormat::I24, alsa::pcm::Format::S24BE),
            (SampleFormat::U24, alsa::pcm::Format::U24LE),
            (SampleFormat::U24, alsa::pcm::Format::U24BE),
            (SampleFormat::I32, alsa::pcm::Format::S32LE),
            (SampleFormat::I32, alsa::pcm::Format::S32BE),
            (SampleFormat::U32, alsa::pcm::Format::U32LE),
            (SampleFormat::U32, alsa::pcm::Format::U32BE),
            (SampleFormat::F32, alsa::pcm::Format::FloatLE),
            (SampleFormat::F32, alsa::pcm::Format::FloatBE),
            (SampleFormat::F64, alsa::pcm::Format::Float64LE),
            (SampleFormat::F64, alsa::pcm::Format::Float64BE),
            (SampleFormat::DsdU8, alsa::pcm::Format::DSDU8),
            (SampleFormat::DsdU16, alsa::pcm::Format::DSDU16LE),
            (SampleFormat::DsdU16, alsa::pcm::Format::DSDU16BE),
            (SampleFormat::DsdU32, alsa::pcm::Format::DSDU32LE),
            (SampleFormat::DsdU32, alsa::pcm::Format::DSDU32BE),
            //SND_PCM_FORMAT_IEC958_SUBFRAME_LE,
            //SND_PCM_FORMAT_IEC958_SUBFRAME_BE,
            //SND_PCM_FORMAT_MU_LAW,
            //SND_PCM_FORMAT_A_LAW,
            //SND_PCM_FORMAT_IMA_ADPCM,
            //SND_PCM_FORMAT_MPEG,
            //SND_PCM_FORMAT_GSM,
            //SND_PCM_FORMAT_SPECIAL,
            //SND_PCM_FORMAT_S24_3LE,
            //SND_PCM_FORMAT_S24_3BE,
            //SND_PCM_FORMAT_U24_3LE,
            //SND_PCM_FORMAT_U24_3BE,
            //SND_PCM_FORMAT_S20_3LE,
            //SND_PCM_FORMAT_S20_3BE,
            //SND_PCM_FORMAT_U20_3LE,
            //SND_PCM_FORMAT_U20_3BE,
            //SND_PCM_FORMAT_S18_3LE,
            //SND_PCM_FORMAT_S18_3BE,
            //SND_PCM_FORMAT_U18_3LE,
            //SND_PCM_FORMAT_U18_3BE,
        ];

        // Collect supported formats, deduplicating since we test both LE and BE variants.
        // If hardware supports both endiannesses (rare), we only report the format once.
        let mut supported_formats = Vec::new();
        for &(sample_format, alsa_format) in FORMATS.iter() {
            if hw_params.test_format(alsa_format).is_ok()
                && !supported_formats.contains(&sample_format)
            {
                supported_formats.push(sample_format);
            }
        }

        let min_rate = hw_params.get_rate_min()?;
        let max_rate = hw_params.get_rate_max()?;

        let sample_rates = if min_rate == max_rate || hw_params.test_rate(min_rate + 1).is_ok() {
            vec![(min_rate, max_rate)]
        } else {
            let mut rates = Vec::new();
            for &sample_rate in crate::COMMON_SAMPLE_RATES.iter() {
                if hw_params.test_rate(sample_rate).is_ok() {
                    rates.push((sample_rate, sample_rate));
                }
            }

            if rates.is_empty() {
                vec![(min_rate, max_rate)]
            } else {
                rates
            }
        };

        let min_channels = hw_params.get_channels_min()?;
        let max_channels = hw_params.get_channels_max()?;

        let max_channels = cmp::min(max_channels, 32); // TODO: limiting to 32 channels or too much stuff is returned
        let supported_channels = (min_channels..max_channels + 1)
            .filter_map(|num| {
                if hw_params.test_channels(num).is_ok() {
                    Some(num as ChannelCount)
                } else {
                    None
                }
            })
            .collect::<Vec<_>>();

        let (min_buffer_size, max_buffer_size) = hw_params_buffer_size_min_max(&hw_params);
        let buffer_size_range = SupportedBufferSize::Range {
            min: min_buffer_size,
            max: max_buffer_size,
        };

        let mut output = Vec::with_capacity(
            supported_formats.len() * supported_channels.len() * sample_rates.len(),
        );
        for &sample_format in supported_formats.iter() {
            for &channels in supported_channels.iter() {
                for &(min_rate, max_rate) in sample_rates.iter() {
                    output.push(SupportedStreamConfigRange {
                        channels,
                        min_sample_rate: min_rate,
                        max_sample_rate: max_rate,
                        buffer_size: buffer_size_range,
                        sample_format,
                    });
                }
            }
        }

        Ok(output.into_iter())
    }

    fn supported_input_configs(
        &self,
    ) -> Result<SupportedInputConfigs, SupportedStreamConfigsError> {
        self.supported_configs(alsa::Direction::Capture)
    }

    fn supported_output_configs(
        &self,
    ) -> Result<SupportedOutputConfigs, SupportedStreamConfigsError> {
        self.supported_configs(alsa::Direction::Playback)
    }

    // ALSA does not offer default stream formats, so instead we compare all supported formats by
    // the `SupportedStreamConfigRange::cmp_default_heuristics` order and select the greatest.
    fn default_config(
        &self,
        stream_t: alsa::Direction,
    ) -> Result<SupportedStreamConfig, DefaultStreamConfigError> {
        let mut formats: Vec<_> = {
            match self.supported_configs(stream_t) {
                Err(SupportedStreamConfigsError::DeviceNotAvailable) => {
                    return Err(DefaultStreamConfigError::DeviceNotAvailable);
                }
                Err(SupportedStreamConfigsError::DeviceBusy) => {
                    return Err(DefaultStreamConfigError::DeviceBusy);
                }
                Err(SupportedStreamConfigsError::InvalidArgument) => {
                    // this happens sometimes when querying for input and output capabilities, but
                    // the device supports only one
                    return Err(DefaultStreamConfigError::StreamTypeNotSupported);
                }
                Err(SupportedStreamConfigsError::BackendSpecific { err }) => {
                    return Err(err.into());
                }
                Ok(fmts) => fmts.collect(),
            }
        };

        formats.sort_by(|a, b| a.cmp_default_heuristics(b));

        match formats.into_iter().next_back() {
            Some(f) => {
                let min_r = f.min_sample_rate;
                let max_r = f.max_sample_rate;
                let mut format = f.with_max_sample_rate();
                const HZ_44100: SampleRate = 44_100;
                if min_r <= HZ_44100 && HZ_44100 <= max_r {
                    format.sample_rate = HZ_44100;
                }
                Ok(format)
            }
            None => Err(DefaultStreamConfigError::StreamTypeNotSupported),
        }
    }

    fn default_input_config(&self) -> Result<SupportedStreamConfig, DefaultStreamConfigError> {
        self.default_config(alsa::Direction::Capture)
    }

    fn default_output_config(&self) -> Result<SupportedStreamConfig, DefaultStreamConfigError> {
        self.default_config(alsa::Direction::Playback)
    }
}

impl Default for Device {
    fn default() -> Self {
        // "default" is a virtual ALSA device that redirects to the configured default. We cannot
        // determine its actual capabilities without opening it, so we return Unknown direction.
        Self {
            pcm_id: DEFAULT_DEVICE.to_owned(),
            desc: Some("Default Audio Device".to_string()),
            direction: DeviceDirection::Unknown,
            _context: Arc::new(
                AlsaContext::new().expect("Failed to initialize ALSA configuration"),
            ),
        }
    }
}

#[derive(Debug)]
struct StreamInner {
    // Flag used to check when to stop polling, regardless of the state of the stream
    // (e.g. broken due to a disconnected device).
    dropping: AtomicBool,

    // The ALSA channel.
    channel: alsa::pcm::PCM,

    // When converting between file descriptors and `snd_pcm_t`, this is the number of
    // file descriptors that this `snd_pcm_t` uses.
    num_descriptors: usize,

    // Format of the samples.
    sample_format: SampleFormat,

    // The configuration used to open this stream.
    conf: StreamConfig,

    // Cached values for performance in audio callback hot path
    period_samples: usize,
    period_frames: usize,
    frame_size: usize,
    silence_template: Box<[u8]>,

    #[allow(dead_code)]
    // Whether or not the hardware supports pausing the stream.
    // TODO: We need an API to expose this. See #197, #284.
    can_pause: bool,

    // Whether to attempt hardware timestamps via `get_htstamp` / `get_trigger_htstamp`.
    //
    // When `true`, hardware timestamps are tried first on every callback and we fall back silently
    // to `creation_instant` if they are transiently unavailable; e.g. the PulseAudio ALSA plugin
    // returns `(0, 0)` for the first several periods after the stream is triggered.
    use_hw_timestamps: bool,

    // Timestamp origin used by the fallback path. Faster without `Option`.
    creation_instant: std::time::Instant,

    // Keep ALSA context alive to prevent premature ALSA config cleanup
    _context: Arc<AlsaContext>,
}

// Assume that the ALSA library is built with thread safe option.
unsafe impl Sync for StreamInner {}

#[derive(Debug)]
pub struct Stream {
    /// The high-priority audio processing thread calling callbacks.
    /// Option used for moving out in destructor.
    thread: Option<JoinHandle<()>>,

    /// Handle to the underlying stream for playback controls.
    inner: Arc<StreamInner>,

    /// Used to signal to stop processing.
    trigger: TriggerSender,

    /// Keeps the read end of the self-pipe alive for the lifetime of the Stream, so that
    /// `trigger.wakeup()` never writes to a closed pipe, even if the worker exited early.
    _rx: Arc<TriggerReceiver>,
}

// Compile-time assertion that Stream is Send and Sync
crate::assert_stream_send!(Stream);
crate::assert_stream_sync!(Stream);

struct StreamWorkerContext {
    descriptors: Box<[libc::pollfd]>,
    transfer_buffer: Box<[u8]>,
    poll_timeout: i32,
}

impl StreamWorkerContext {
    fn new(poll_timeout: &Option<Duration>, stream: &StreamInner, rx: &TriggerReceiver) -> Self {
        let poll_timeout: i32 = if let Some(d) = poll_timeout {
            d.as_millis().try_into().unwrap()
        } else {
            -1 // Don't timeout, wait forever.
        };

        // Pre-allocate buffer to exactly one period size with proper equilibrium values.
        let transfer_buffer = stream.silence_template.clone();

        // Pre-allocate and initialize descriptors vector: 1 for self-pipe + stream.num_descriptors
        // for ALSA. The descriptor count is constant for the lifetime of stream parameters, and
        // poll() overwrites revents on each call, so we only need to set up fd and events once.
        let total_descriptors = 1 + stream.num_descriptors;
        let mut descriptors = vec![
            libc::pollfd {
                fd: 0,
                events: 0,
                revents: 0
            };
            total_descriptors
        ]
        .into_boxed_slice();

        // Set up self-pipe descriptor at index 0
        descriptors[0] = libc::pollfd {
            fd: rx.0,
            events: libc::POLLIN,
            revents: 0,
        };

        // Set up ALSA descriptors starting at index 1
        let filled = stream
            .channel
            .fill(&mut descriptors[1..])
            .expect("Failed to fill ALSA descriptors");
        debug_assert_eq!(filled, stream.num_descriptors);

        Self {
            descriptors,
            transfer_buffer,
            poll_timeout,
        }
    }
}

fn input_stream_worker(
    rx: Arc<TriggerReceiver>,
    stream: &StreamInner,
    data_callback: &mut (dyn FnMut(&Data, &InputCallbackInfo) + Send + 'static),
    error_callback: &mut (dyn FnMut(StreamError) + Send + 'static),
    timeout: Option<Duration>,
) {
    boost_current_thread_priority(stream.conf.buffer_size, stream.conf.sample_rate);

    let mut ctxt = StreamWorkerContext::new(&timeout, stream, &rx);
    loop {
        if stream.dropping.load(Ordering::Acquire) {
            return;
        }
        let result = match poll_for_period(&rx, stream, &mut ctxt) {
            Ok(Poll::Pending) => continue,
            Ok(Poll::Ready {
                status,
                delay_frames,
            }) => process_input(
                stream,
                &mut ctxt.transfer_buffer,
                status,
                delay_frames,
                data_callback,
            ),
            Err(err) => Err(err),
        };
        if let Err(err) = result {
            match err {
                StreamError::BufferUnderrun => {
                    error_callback(StreamError::BufferUnderrun);
                    if let Err(err) = stream.channel.prepare() {
                        error_callback(err.into());
                    } else if let Err(err) = stream.channel.start() {
                        error_callback(err.into());
                    }
                }
                StreamError::DeviceNotAvailable => {
                    error_callback(StreamError::DeviceNotAvailable);
                    return;
                }
                err => error_callback(err),
            }
        }
    }
}

fn output_stream_worker(
    rx: Arc<TriggerReceiver>,
    stream: &StreamInner,
    data_callback: &mut (dyn FnMut(&mut Data, &OutputCallbackInfo) + Send + 'static),
    error_callback: &mut (dyn FnMut(StreamError) + Send + 'static),
    timeout: Option<Duration>,
) {
    boost_current_thread_priority(stream.conf.buffer_size, stream.conf.sample_rate);

    let mut ctxt = StreamWorkerContext::new(&timeout, stream, &rx);

    loop {
        if stream.dropping.load(Ordering::Acquire) {
            return;
        }
        let result = match poll_for_period(&rx, stream, &mut ctxt) {
            Ok(Poll::Pending) => continue,
            Ok(Poll::Ready {
                status,
                delay_frames,
            }) => process_output(
                stream,
                &mut ctxt.transfer_buffer,
                status,
                delay_frames,
                data_callback,
            ),
            Err(err) => Err(err),
        };
        if let Err(err) = result {
            match err {
                StreamError::BufferUnderrun => {
                    error_callback(StreamError::BufferUnderrun);
                    if let Err(err) = stream.channel.prepare() {
                        error_callback(err.into());
                    }
                    // No need to call start() for output streams after prepare();
                    // ALSA automatically restarts them when the buffer is refilled
                    // and the stream is triggered again.
                }
                StreamError::DeviceNotAvailable => {
                    error_callback(StreamError::DeviceNotAvailable);
                    return;
                }
                err => error_callback(err),
            }
        }
    }
}

#[cfg(feature = "audio_thread_priority")]
fn boost_current_thread_priority(buffer_size: BufferSize, sample_rate: SampleRate) {
    use audio_thread_priority::promote_current_thread_to_real_time;

    let buffer_size = if let BufferSize::Fixed(buffer_size) = buffer_size {
        buffer_size
    } else {
        // if the buffer size isn't fixed, let audio_thread_priority choose a sensible default value
        0
    };

    if let Err(err) = promote_current_thread_to_real_time(buffer_size, sample_rate) {
        eprintln!("Failed to promote audio thread to real-time priority: {err}");
    }
}

#[cfg(not(feature = "audio_thread_priority"))]
fn boost_current_thread_priority(_: BufferSize, _: SampleRate) {}

/// Attempt hardware resume from a suspend event (`ESTRPIPE`).
fn try_resume(channel: &alsa::PCM) -> Result<Poll, StreamError> {
    match channel.resume() {
        // device resumed successfully and will continue running on its own
        Ok(()) => Ok(Poll::Pending),
        // device is still resuming; poll again until it is ready.
        Err(e) if e.errno() == libc::EAGAIN => Ok(Poll::Pending),
        // hardware does not support soft resume
        Err(e) if e.errno() == libc::ENOSYS => Err(StreamError::BufferUnderrun),
        Err(e) => Err(e.into()),
    }
}

enum Poll {
    Pending,
    Ready {
        status: alsa::pcm::Status,
        delay_frames: usize,
    },
}

// This block is shared between both input and output stream worker functions.
fn poll_for_period(
    rx: &TriggerReceiver,
    stream: &StreamInner,
    ctxt: &mut StreamWorkerContext,
) -> Result<Poll, StreamError> {
    let StreamWorkerContext {
        ref mut descriptors,
        ref poll_timeout,
        ..
    } = *ctxt;

    let res = alsa::poll::poll(descriptors, *poll_timeout)?;
    if res == 0 {
        // poll() returned 0: either a timeout or a spurious wakeup. Nothing to do.
        return Ok(Poll::Pending);
    }

    if descriptors[0].revents != 0 {
        // Self-pipe fired: the stream is being dropped. Clear the pipe and let the
        // worker loop detect the dropping flag on the next iteration.
        rx.clear_pipe();
        return Ok(Poll::Pending);
    }

    let revents = stream.channel.revents(&descriptors[1..])?;
    // No events: spurious wakeup, poll again.
    if revents.is_empty() {
        return Ok(Poll::Pending);
    }
    // POLLHUP/POLLNVAL: the device has been disconnected.
    if revents.intersects(alsa::poll::Flags::HUP | alsa::poll::Flags::NVAL) {
        return Err(StreamError::DeviceNotAvailable);
    }
    // POLLERR signals an xrun or suspend; avail() below returns EPIPE/ESTRPIPE accordingly.
    // POLLIN/POLLOUT: data is ready, fall through to process it.

    let status = stream.channel.status()?;
    let avail_frames = match stream.channel.avail() {
        // Xrun: recover via prepare() (+ start() for capture, handled by the worker).
        Err(err) if err.errno() == libc::EPIPE => return Err(StreamError::BufferUnderrun),
        // Suspend: try hardware resume first; fall back to prepare() if unsupported.
        Err(err) if err.errno() == libc::ESTRPIPE => return try_resume(&stream.channel),
        res => res,
    }? as usize;
    let delay_frames = match status.get_delay() {
        d if d < 0 => 0,
        d => d as usize,
    };
    let available_samples = avail_frames * stream.conf.channels as usize;

    // ALSA can have spurious wakeups where poll returns but avail < avail_min.
    // This is documented to occur with dmix (timer-driven) and other plugins.
    // Verify we have room for at least one full period before processing.
    // See: https://bugzilla.kernel.org/show_bug.cgi?id=202499
    if available_samples < stream.period_samples {
        return Ok(Poll::Pending);
    }

    Ok(Poll::Ready {
        status,
        delay_frames,
    })
}

// Read input data from ALSA and deliver it to the user.
fn process_input(
    stream: &StreamInner,
    buffer: &mut [u8],
    status: alsa::pcm::Status,
    delay_frames: usize,
    data_callback: &mut (dyn FnMut(&Data, &InputCallbackInfo) + Send + 'static),
) -> Result<(), StreamError> {
    let mut frames_read = 0;
    while frames_read < stream.period_frames {
        match stream
            .channel
            .io_bytes()
            .readi(&mut buffer[frames_read * stream.frame_size..])
        {
            Ok(n) => frames_read += n,
            Err(err) if err.errno() == libc::EPIPE || err.errno() == libc::ESTRPIPE => {
                // EPIPE = xrun, ESTRPIPE = hardware suspend. Both require prepare()+restart;
                // attempting resume mid-loop with a partial transfer in the ring buffer is unsafe.
                return Err(StreamError::BufferUnderrun);
            }
            Err(err) if err.errno() == libc::ENODEV => return Err(StreamError::DeviceNotAvailable),
            Err(err) => return Err(err.into()),
        }
    }
    let data = buffer.as_mut_ptr() as *mut ();
    let data = unsafe { Data::from_parts(data, stream.period_samples, stream.sample_format) };
    let callback = if stream.use_hw_timestamps {
        stream_timestamp_hardware(&status)
            .or_else(|_| stream_timestamp_fallback(stream.creation_instant))
    } else {
        stream_timestamp_fallback(stream.creation_instant)
    }?;
    let delay_duration = frames_to_duration(delay_frames, stream.conf.sample_rate);
    let capture = callback
        .sub(delay_duration)
        .ok_or_else(|| BackendSpecificError {
            description: "`capture` is earlier than representation supported by `StreamInstant`"
                .to_string(),
        })?;
    let timestamp = crate::InputStreamTimestamp { callback, capture };
    let info = crate::InputCallbackInfo { timestamp };
    data_callback(&data, &info);

    Ok(())
}

// Request data from the user's function and write it via ALSA.
fn process_output(
    stream: &StreamInner,
    buffer: &mut [u8],
    status: alsa::pcm::Status,
    delay_frames: usize,
    data_callback: &mut (dyn FnMut(&mut Data, &OutputCallbackInfo) + Send + 'static),
) -> Result<(), StreamError> {
    // Buffer is always pre-filled with equilibrium, user overwrites what they want
    buffer.copy_from_slice(&stream.silence_template);
    {
        let data = buffer.as_mut_ptr() as *mut ();
        let mut data =
            unsafe { Data::from_parts(data, stream.period_samples, stream.sample_format) };
        let callback = if stream.use_hw_timestamps {
            stream_timestamp_hardware(&status)
                .or_else(|_| stream_timestamp_fallback(stream.creation_instant))
        } else {
            stream_timestamp_fallback(stream.creation_instant)
        }?;
        let delay_duration = frames_to_duration(delay_frames, stream.conf.sample_rate);
        let playback = callback
            .add(delay_duration)
            .ok_or_else(|| BackendSpecificError {
                description: "`playback` occurs beyond representation supported by `StreamInstant`"
                    .to_string(),
            })?;
        let timestamp = crate::OutputStreamTimestamp { callback, playback };
        let info = crate::OutputCallbackInfo { timestamp };
        data_callback(&mut data, &info);
    }

    let mut frames_written = 0;
    while frames_written < stream.period_frames {
        match stream
            .channel
            .io_bytes()
            .writei(&buffer[frames_written * stream.frame_size..])
        {
            Ok(n) => frames_written += n,
            Err(err) if err.errno() == libc::EPIPE || err.errno() == libc::ESTRPIPE => {
                // EPIPE = xrun, ESTRPIPE = hardware suspend. Both require prepare()+restart;
                // attempting resume mid-loop with a partial transfer in the ring buffer is unsafe.
                return Err(StreamError::BufferUnderrun);
            }
            Err(err) if err.errno() == libc::ENODEV => return Err(StreamError::DeviceNotAvailable),
            Err(err) => return Err(err.into()),
        }
    }
    Ok(())
}

// Use hardware timestamps from ALSA.
//
// This ensures accurate timestamps based on actual hardware timing.
#[inline]
fn stream_timestamp_hardware(
    status: &alsa::pcm::Status,
) -> Result<crate::StreamInstant, BackendSpecificError> {
    let trigger_ts = status.get_trigger_htstamp();
    // trigger_htstamp records when the PCM stream started.
    // On the first few callbacks, it might not have been set yet,
    // which would yield a huge positive nanos nd cause non-monotonicity
    // once it is set. Bail out and let the caller use the fallback.
    // See https://github.com/RustAudio/cpal/issues/710
    if trigger_ts.tv_sec == 0 && trigger_ts.tv_nsec == 0 {
        return Err(BackendSpecificError {
            description: "trigger_htstamp not yet set".to_string(),
        });
    }
    let ts = status.get_htstamp();
    let nanos = timespec_diff_nanos(ts, trigger_ts);
    if nanos < 0 {
        let description = format!(
            "get_htstamp `{}.{}` was earlier than get_trigger_htstamp `{}.{}`",
            ts.tv_sec, ts.tv_nsec, trigger_ts.tv_sec, trigger_ts.tv_nsec
        );
        return Err(BackendSpecificError { description });
    }
    Ok(crate::StreamInstant::from_nanos(nanos))
}

// Use elapsed duration since stream creation as fallback when hardware timestamps are unavailable.
//
// This ensures positive values that are compatible with our `StreamInstant` representation.
#[inline]
fn stream_timestamp_fallback(
    creation: std::time::Instant,
) -> Result<crate::StreamInstant, BackendSpecificError> {
    let now = std::time::Instant::now();
    let duration = now.duration_since(creation);
    crate::StreamInstant::from_nanos_i128(duration.as_nanos() as i128).ok_or(BackendSpecificError {
        description: "stream duration has exceeded `StreamInstant` representation".to_string(),
    })
}

// Adapted from `timestamp2ns` here:
// https://fossies.org/linux/alsa-lib/test/audio_time.c
#[inline]
#[allow(clippy::unnecessary_cast)]
fn timespec_to_nanos(ts: libc::timespec) -> i64 {
    ts.tv_sec as i64 * 1_000_000_000 + ts.tv_nsec as i64
}

// Adapted from `timediff` here:
// https://fossies.org/linux/alsa-lib/test/audio_time.c
#[inline]
fn timespec_diff_nanos(a: libc::timespec, b: libc::timespec) -> i64 {
    timespec_to_nanos(a) - timespec_to_nanos(b)
}

// Convert the given duration in frames at the given sample rate to a `std::time::Duration`.
#[inline]
fn frames_to_duration(frames: usize, rate: crate::SampleRate) -> std::time::Duration {
    let secsf = frames as f64 / rate as f64;
    let secs = secsf as u64;
    let nanos = ((secsf - secs as f64) * 1_000_000_000.0) as u32;
    std::time::Duration::new(secs, nanos)
}

impl Stream {
    fn new_input<D, E>(
        inner: Arc<StreamInner>,
        mut data_callback: D,
        mut error_callback: E,
        timeout: Option<Duration>,
    ) -> Stream
    where
        D: FnMut(&Data, &InputCallbackInfo) + Send + 'static,
        E: FnMut(StreamError) + Send + 'static,
    {
        let (tx, rx) = trigger();
        let rx_thread = rx.clone();
        let stream = inner.clone();
        let thread = thread::Builder::new()
            .name("cpal_alsa_in".to_owned())
            .spawn(move || {
                input_stream_worker(
                    rx_thread,
                    &stream,
                    &mut data_callback,
                    &mut error_callback,
                    timeout,
                );
            })
            .unwrap();
        Self {
            thread: Some(thread),
            inner,
            trigger: tx,
            _rx: rx,
        }
    }

    fn new_output<D, E>(
        inner: Arc<StreamInner>,
        mut data_callback: D,
        mut error_callback: E,
        timeout: Option<Duration>,
    ) -> Stream
    where
        D: FnMut(&mut Data, &OutputCallbackInfo) + Send + 'static,
        E: FnMut(StreamError) + Send + 'static,
    {
        let (tx, rx) = trigger();
        let rx_thread = rx.clone();
        let stream = inner.clone();
        let thread = thread::Builder::new()
            .name("cpal_alsa_out".to_owned())
            .spawn(move || {
                output_stream_worker(
                    rx_thread,
                    &stream,
                    &mut data_callback,
                    &mut error_callback,
                    timeout,
                );
            })
            .unwrap();
        Self {
            thread: Some(thread),
            inner,
            trigger: tx,
            _rx: rx,
        }
    }
}

impl Drop for Stream {
    fn drop(&mut self) {
        self.inner.dropping.store(true, Ordering::Release);
        self.trigger.wakeup();
        if let Some(handle) = self.thread.take() {
            let _ = handle.join();
        }
    }
}

impl StreamTrait for Stream {
    fn play(&self) -> Result<(), PlayStreamError> {
        self.inner.channel.pause(false).ok();
        Ok(())
    }
    fn pause(&self) -> Result<(), PauseStreamError> {
        self.inner.channel.pause(true).ok();
        Ok(())
    }
    fn buffer_size(&self) -> Option<FrameCount> {
        Some(self.inner.period_frames as FrameCount)
    }
}

// Convert ALSA frames to FrameCount, clamping to valid range.
// ALSA Frames are i64 (64-bit) or i32 (32-bit).
fn clamp_frame_count(buffer_size: alsa::pcm::Frames) -> FrameCount {
    buffer_size.max(1).try_into().unwrap_or(FrameCount::MAX)
}

fn hw_params_buffer_size_min_max(hw_params: &alsa::pcm::HwParams) -> (FrameCount, FrameCount) {
    let min_buf = hw_params
        .get_buffer_size_min()
        .map(clamp_frame_count)
        .unwrap_or(1);
    let max_buf = hw_params
        .get_buffer_size_max()
        .map(clamp_frame_count)
        .unwrap_or(FrameCount::MAX);
    (min_buf, max_buf)
}

fn init_hw_params<'a>(
    pcm_handle: &'a alsa::pcm::PCM,
    config: StreamConfig,
    sample_format: SampleFormat,
) -> Result<alsa::pcm::HwParams<'a>, BackendSpecificError> {
    let hw_params = alsa::pcm::HwParams::any(pcm_handle)?;
    hw_params.set_access(alsa::pcm::Access::RWInterleaved)?;

    // Determine which endianness the hardware actually supports for this format.
    // We prefer native endian (no conversion needed) but fall back to the opposite
    // endian if that's all the hardware supports (e.g., LE USB DAC on BE system).
    let alsa_format = sample_format_to_alsa_format(&hw_params, sample_format)?;
    hw_params.set_format(alsa_format)?;

    hw_params.set_rate(config.sample_rate, alsa::ValueOr::Nearest)?;
    hw_params.set_channels(config.channels as u32)?;
    Ok(hw_params)
}

/// Convert SampleFormat to the appropriate alsa::pcm::Format based on what the hardware supports.
/// Prefers native endian, falls back to non-native if that's all the hardware supports.
fn sample_format_to_alsa_format(
    hw_params: &alsa::pcm::HwParams,
    sample_format: SampleFormat,
) -> Result<alsa::pcm::Format, BackendSpecificError> {
    use alsa::pcm::Format;

    // For each sample format, define (native_endian_format, opposite_endian_format) pairs
    let (native, opposite) = match sample_format {
        SampleFormat::I8 => return Ok(Format::S8), // No endianness
        SampleFormat::U8 => return Ok(Format::U8), // No endianness
        #[cfg(target_endian = "little")]
        SampleFormat::I16 => (Format::S16LE, Format::S16BE),
        #[cfg(target_endian = "big")]
        SampleFormat::I16 => (Format::S16BE, Format::S16LE),
        #[cfg(target_endian = "little")]
        SampleFormat::U16 => (Format::U16LE, Format::U16BE),
        #[cfg(target_endian = "big")]
        SampleFormat::U16 => (Format::U16BE, Format::U16LE),
        #[cfg(target_endian = "little")]
        SampleFormat::I24 => (Format::S24LE, Format::S24BE),
        #[cfg(target_endian = "big")]
        SampleFormat::I24 => (Format::S24BE, Format::S24LE),
        #[cfg(target_endian = "little")]
        SampleFormat::U24 => (Format::U24LE, Format::U24BE),
        #[cfg(target_endian = "big")]
        SampleFormat::U24 => (Format::U24BE, Format::U24LE),
        #[cfg(target_endian = "little")]
        SampleFormat::I32 => (Format::S32LE, Format::S32BE),
        #[cfg(target_endian = "big")]
        SampleFormat::I32 => (Format::S32BE, Format::S32LE),
        #[cfg(target_endian = "little")]
        SampleFormat::U32 => (Format::U32LE, Format::U32BE),
        #[cfg(target_endian = "big")]
        SampleFormat::U32 => (Format::U32BE, Format::U32LE),
        #[cfg(target_endian = "little")]
        SampleFormat::F32 => (Format::FloatLE, Format::FloatBE),
        #[cfg(target_endian = "big")]
        SampleFormat::F32 => (Format::FloatBE, Format::FloatLE),
        #[cfg(target_endian = "little")]
        SampleFormat::F64 => (Format::Float64LE, Format::Float64BE),
        #[cfg(target_endian = "big")]
        SampleFormat::F64 => (Format::Float64BE, Format::Float64LE),
        SampleFormat::DsdU8 => return Ok(Format::DSDU8),
        #[cfg(target_endian = "little")]
        SampleFormat::DsdU16 => (Format::DSDU16LE, Format::DSDU16BE),
        #[cfg(target_endian = "big")]
        SampleFormat::DsdU16 => (Format::DSDU16BE, Format::DSDU16LE),
        #[cfg(target_endian = "little")]
        SampleFormat::DsdU32 => (Format::DSDU32LE, Format::DSDU32BE),
        #[cfg(target_endian = "big")]
        SampleFormat::DsdU32 => (Format::DSDU32BE, Format::DSDU32LE),
        _ => {
            return Err(BackendSpecificError {
                description: format!("Sample format '{sample_format}' is not supported"),
            })
        }
    };

    // Try native endian first (optimal - no conversion needed)
    if hw_params.test_format(native).is_ok() {
        return Ok(native);
    }

    // Fall back to opposite endian if hardware only supports that
    if hw_params.test_format(opposite).is_ok() {
        return Ok(opposite);
    }

    Err(BackendSpecificError {
        description: format!(
            "Sample format '{sample_format}' is not supported by hardware in any endianness"
        ),
    })
}

fn set_hw_params_from_format(
    pcm_handle: &alsa::pcm::PCM,
    config: StreamConfig,
    sample_format: SampleFormat,
) -> Result<bool, BackendSpecificError> {
    let hw_params = init_hw_params(pcm_handle, config, sample_format)?;

    // When BufferSize::Fixed(x) is specified, we configure double-buffering with
    // buffer_size = 2x and period_size = x. This provides consistent low-latency
    // behavior across different ALSA implementations and hardware.
    if let BufferSize::Fixed(buffer_frames) = config.buffer_size {
        hw_params.set_buffer_size_near((2 * buffer_frames) as alsa::pcm::Frames)?;
        hw_params
            .set_period_size_near(buffer_frames as alsa::pcm::Frames, alsa::ValueOr::Nearest)?;
    }

    // Apply hardware parameters
    pcm_handle.hw_params(&hw_params)?;

    // For BufferSize::Default, constrain to device's configured period with 2-period buffering.
    // PipeWire-ALSA picks a good period size but pairs it with many periods (huge buffer).
    // We need to re-initialize hw_params and set BOTH period and buffer to constrain properly.
    if config.buffer_size == BufferSize::Default {
        if let Ok(period) = hw_params.get_period_size() {
            // Re-initialize hw_params to clear previous constraints
            let hw_params = init_hw_params(pcm_handle, config, sample_format)?;

            // Set both period (to device's chosen value) and buffer (to 2 periods)
            hw_params.set_period_size_near(period, alsa::ValueOr::Nearest)?;
            hw_params.set_buffer_size_near(2 * period)?;

            // Re-apply with new constraints
            pcm_handle.hw_params(&hw_params)?;
        }
    }

    Ok(hw_params.can_pause())
}

fn set_sw_params_from_format(
    pcm_handle: &alsa::pcm::PCM,
    config: StreamConfig,
    stream_type: alsa::Direction,
) -> Result<usize, BackendSpecificError> {
    let sw_params = pcm_handle.sw_params_current()?;

    let period_samples = {
        let (buffer, period) = pcm_handle.get_params()?;
        if buffer == 0 {
            return Err(BackendSpecificError {
                description: "initialization resulted in a null buffer".to_string(),
            });
        }
        let start_threshold = match stream_type {
            alsa::Direction::Playback => {
                // Start playback when 2 periods are filled. This ensures consistent low-latency
                // startup regardless of total buffer size (whether 2 or more periods).
                2 * period
            }
            alsa::Direction::Capture => 1,
        };
        sw_params.set_start_threshold(start_threshold as alsa::pcm::Frames)?;
        sw_params.set_avail_min(period as alsa::pcm::Frames)?;

        period as usize * config.channels as usize
    };

    sw_params.set_tstamp_mode(true)?;
    sw_params.set_tstamp_type(alsa::pcm::TstampType::MonotonicRaw)?;

    // tstamp_type param cannot be changed after the device is opened.
    // The default tstamp_type value on most Linux systems is "monotonic",
    // let's try to use it if setting the tstamp_type fails.
    if pcm_handle.sw_params(&sw_params).is_err() {
        sw_params.set_tstamp_type(alsa::pcm::TstampType::Monotonic)?;
        pcm_handle.sw_params(&sw_params)?;
    }

    Ok(period_samples)
}

fn canonical_pcm_id(pcm_id: &str) -> String {
    if let Some((prefix, rest)) = pcm_id.split_once(':') {
        let (card_str, device_str) = match rest.split_once(',') {
            Some((c, d)) => (c.trim(), d.trim()),
            None => (rest.trim(), "0"),
        };
        if !card_str.contains('=') {
            if let Ok(device) = device_str.parse::<u32>() {
                return format!("{prefix}:CARD={card_str},DEV={device}");
            }
        }
    }
    pcm_id.to_owned()
}

impl From<alsa::Error> for BackendSpecificError {
    fn from(err: alsa::Error) -> Self {
        Self {
            description: err.to_string(),
        }
    }
}

impl From<alsa::Error> for BuildStreamError {
    fn from(err: alsa::Error) -> Self {
        let err: BackendSpecificError = err.into();
        err.into()
    }
}

impl From<alsa::Error> for SupportedStreamConfigsError {
    fn from(err: alsa::Error) -> Self {
        let err: BackendSpecificError = err.into();
        err.into()
    }
}

impl From<alsa::Error> for PlayStreamError {
    fn from(err: alsa::Error) -> Self {
        let err: BackendSpecificError = err.into();
        err.into()
    }
}

impl From<alsa::Error> for PauseStreamError {
    fn from(err: alsa::Error) -> Self {
        let err: BackendSpecificError = err.into();
        err.into()
    }
}

impl From<alsa::Error> for StreamError {
    fn from(err: alsa::Error) -> Self {
        let err: BackendSpecificError = err.into();
        err.into()
    }
}