use crate::audio::{AudioBackend, AudioError};
use std::sync::mpsc::{Sender, Receiver, RecvTimeoutError};
use std::sync::{Arc, Mutex};
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use std::thread::{self, JoinHandle};
use std::time::Duration;
use pipewire as pw;
use pw::{properties::properties, spa};
use spa::pod::Pod;
use ringbuf::{HeapRb, traits::{Consumer, Producer, Split}};

pub struct PipeWireBackend {
    capture_state: Mutex<Option<CaptureState>>,
    playback_state: Mutex<Option<PlaybackState>>,
}

struct CaptureState {
    cmd_tx: pw::channel::Sender<()>,
    thread: JoinHandle<()>,
}

struct PlaybackState {
    cmd_tx: pw::channel::Sender<()>,
    thread: JoinHandle<()>,
}

impl Default for PipeWireBackend {
    fn default() -> Self {
        Self::new()
    }
}

impl PipeWireBackend {
    pub fn new() -> Self {
        pw::init();
        Self {
            capture_state: Mutex::new(None),
            playback_state: Mutex::new(None),
        }
    }
}

impl AudioBackend for PipeWireBackend {
    fn start_capture(&self, tx: Sender<Vec<i16>>, target_node: Option<String>) -> Result<(), AudioError> {
        let mut capture_guard = self.capture_state.lock().unwrap();
        if capture_guard.is_some() {
            return Err(AudioError::Stream("Capture already started".to_string()));
        }

        let (cmd_tx, cmd_rx) = pw::channel::channel::<()>();
        let tx_clone = tx.clone();

        let thread = thread::Builder::new()
            .name("peerspeak-capture".to_string())
            .spawn(move || {
                if let Err(e) = run_capture(cmd_rx, tx_clone, target_node) {
                    eprintln!("Capture thread error: {:?}", e);
                }
            })
            .map_err(|e| AudioError::Init(e.to_string()))?;

        *capture_guard = Some(CaptureState { cmd_tx, thread });
        Ok(())
    }

    fn start_playback(
        &self,
        rx: Receiver<Vec<i16>>,
        target_node: Option<String>,
        ring_fill: Arc<AtomicUsize>,
    ) -> Result<(), AudioError> {
        let mut playback_guard = self.playback_state.lock().unwrap();
        if playback_guard.is_some() {
            return Err(AudioError::Stream("Playback already started".to_string()));
        }

        let (cmd_tx, cmd_rx) = pw::channel::channel::<()>();

        let thread = thread::Builder::new()
            .name("peerspeak-playback".to_string())
            .spawn(move || {
                if let Err(e) = run_playback(cmd_rx, rx, target_node, ring_fill) {
                    eprintln!("Playback thread error: {:?}", e);
                }
            })
            .map_err(|e| AudioError::Init(e.to_string()))?;

        *playback_guard = Some(PlaybackState { cmd_tx, thread });
        Ok(())
    }

    fn stop(&self) -> Result<(), AudioError> {
        // Stop capture
        let mut capture_guard = self.capture_state.lock().unwrap();
        if let Some(state) = capture_guard.take() {
            let _ = state.cmd_tx.send(());
            let _ = state.thread.join();
        }

        // Stop playback
        let mut playback_guard = self.playback_state.lock().unwrap();
        if let Some(state) = playback_guard.take() {
            let _ = state.cmd_tx.send(());
            let _ = state.thread.join();
        }

        Ok(())
    }
}

fn run_capture(cmd_rx: pw::channel::Receiver<()>, tx: Sender<Vec<i16>>, target_node: Option<String>) -> Result<(), AudioError> {
    let mainloop = pw::main_loop::MainLoopRc::new(None)
        .map_err(|e| AudioError::Init(e.to_string()))?;
    let context = pw::context::ContextRc::new(&mainloop, None)
        .map_err(|e| AudioError::Init(e.to_string()))?;
    let core = context.connect_rc(None)
        .map_err(|e| AudioError::Init(e.to_string()))?;

    // Ring buffer setup: 9600 samples (200ms capacity for mono 48kHz)
    let rb = HeapRb::<i16>::new(9600);
    let (producer, mut consumer) = rb.split();

    // Command receiver to quit main loop
    let mainloop_clone = mainloop.clone();
    let _cmd_recv = cmd_rx.attach(mainloop.loop_(), move |_| {
        mainloop_clone.quit();
    });

    let mut props = properties! {
        *pw::keys::MEDIA_TYPE => "Audio",
        *pw::keys::MEDIA_CATEGORY => "Capture",
        *pw::keys::MEDIA_ROLE => "Communication",
    };
    if let Some(target) = target_node {
        props.insert("node.target", target);
    }

    let stream = pw::stream::StreamBox::new(&core, "peerspeak-capture-stream", props)
        .map_err(|e| AudioError::Init(e.to_string()))?;

    struct CaptureUserData<P: ringbuf::traits::Producer<Item = i16>> {
        producer: P,
    }

    let _listener = stream
        .add_local_listener_with_user_data(CaptureUserData { producer })
        .process(|stream, user_data| {
            if let Some(mut buffer) = stream.dequeue_buffer() {
                let datas = buffer.datas_mut();
                if !datas.is_empty() {
                    let data = &mut datas[0];
                    let size = data.chunk().size() as usize;
                    if let Some(slice) = data.data() {
                        for_each_capture_sample(slice, size, |sample| {
                            let _ = user_data.producer.try_push(sample);
                        });
                    }
                }
            }
        })
        .register()
        .map_err(|e| AudioError::Init(e.to_string()))?;

    let mut audio_info = spa::param::audio::AudioInfoRaw::new();
    audio_info.set_format(spa::param::audio::AudioFormat::S16LE);
    audio_info.set_rate(48000);
    audio_info.set_channels(1); // Mono

    let obj = pw::spa::pod::Object {
        type_: pw::spa::utils::SpaTypes::ObjectParamFormat.as_raw(),
        id: pw::spa::param::ParamType::EnumFormat.as_raw(),
        properties: audio_info.into(),
    };
    let values: Vec<u8> = pw::spa::pod::serialize::PodSerializer::serialize(
        std::io::Cursor::new(Vec::new()),
        &pw::spa::pod::Value::Object(obj),
    )
    .unwrap()
    .0
    .into_inner();

    let mut params = [Pod::from_bytes(&values).unwrap()];

    stream.connect(
        spa::utils::Direction::Input,
        None,
        pw::stream::StreamFlags::AUTOCONNECT
            | pw::stream::StreamFlags::MAP_BUFFERS
            | pw::stream::StreamFlags::RT_PROCESS,
        &mut params,
    )
    .map_err(|e| AudioError::Stream(e.to_string()))?;

    // Spawn the worker thread to pop from consumer and send Vec<i16> frames
    let running = Arc::new(AtomicBool::new(true));
    let running_clone = running.clone();
    let worker_handle = thread::spawn(move || {
        let mut frame = Vec::with_capacity(960);
        while running_clone.load(Ordering::Relaxed) {
            let mut popped = false;
            while let Some(sample) = consumer.try_pop() {
                popped = true;
                frame.push(sample);
                if frame.len() == 960 {
                    if tx.send(frame).is_err() {
                        return;
                    }
                    frame = Vec::with_capacity(960);
                }
            }
            if !popped {
                thread::sleep(Duration::from_millis(5));
            }
        }
    });

    mainloop.run();

    running.store(false, Ordering::Relaxed);
    let _ = worker_handle.join();

    Ok(())
}

/// Visit the complete S16LE samples in the portion PipeWire reports as filled.
/// Clamp the reported byte count to the mapped slice before indexing: a bad
/// chunk size must not panic from the realtime capture callback.
fn for_each_capture_sample(slice: &[u8], size: usize, mut visit: impl FnMut(i16)) {
    let size = size.min(slice.len());
    for chunk in slice[..size].chunks_exact(2) {
        visit(i16::from_le_bytes([chunk[0], chunk[1]]));
    }
}

/// Frames the playback RT callback should produce this cycle.
///
/// `requested` is the graph's per-cycle quantum from `Buffer::requested()` (0 if
/// the graph didn't set it — first buffers / non-driver cycles). `mapped_frames`
/// is how many frames the mapped buffer slice can physically hold.
///
/// Honoring the requested quantum is what makes playback correct on any graph
/// quantum, instead of the old hard-pinned 1024 (which crackled when a machine's
/// `clock.quantum` wasn't 1024). Rules:
/// - never write past the mapped slice (clamp to `mapped_frames`);
/// - when `requested == 0`, fall back to a safe cap (≤ one 1024-frame quantum) —
///   never the whole slice, because over-pulling past the ring depth is exactly
///   the sustained crackle this replaces. Under-filling a cycle is a harmless
///   brief glitch (PipeWire pads).
// How often the playback worker wakes to re-check its `running` flag while idle,
// bounding how long `stop` can block joining the worker (bug A7). A plain
// blocking receive would never re-check the flag, waking only on a new frame or
// the sender dropping, so shutdown would hang if the sender outlives `stop`.
const WORKER_POLL: Duration = Duration::from_millis(100);

/// Pump frames from `rx` to `on_frame` until `running` goes false or the sender
/// disconnects. Uses a timed receive so the loop re-checks `running` at least
/// every `WORKER_POLL` even when no frames arrive — this is what lets `stop()`
/// join the worker promptly instead of hanging on a parked blocking `recv()`
/// (bug A7). Pure w.r.t. its inputs (no PipeWire), so it's unit-testable.
fn drain_loop(
    rx: &Receiver<Vec<i16>>,
    running: &AtomicBool,
    mut on_frame: impl FnMut(Vec<i16>),
) {
    while running.load(Ordering::Relaxed) {
        match rx.recv_timeout(WORKER_POLL) {
            Ok(frame) => on_frame(frame),
            Err(RecvTimeoutError::Timeout) => continue, // re-check `running`
            Err(RecvTimeoutError::Disconnected) => return, // sender gone for good
        }
    }
}

/// Reserve exact occupancy before making a frame visible to the consumer.
/// `after_reserve` is empty in production and lets the regression test force a
/// consumer interleaving at the critical ordering boundary.
fn publish_frame<P: Producer<Item = i16>>(
    fill: &AtomicUsize,
    dropped: &AtomicU64,
    producer: &mut P,
    frame: &[i16],
    after_reserve: impl FnOnce(),
) {
    fill.fetch_add(frame.len(), Ordering::Relaxed);
    after_reserve();
    let pushed = producer.push_slice(frame);
    if pushed != frame.len() {
        fill.fetch_sub(frame.len() - pushed, Ordering::Relaxed);
        dropped.fetch_add(1, Ordering::Relaxed);
    }
}

fn frames_to_produce(requested: usize, mapped_frames: usize) -> usize {
    /// Safe per-cycle fallback when the graph doesn't report a quantum.
    const FALLBACK_FRAMES: usize = 1024;
    let want = if requested > 0 { requested } else { FALLBACK_FRAMES };
    want.min(mapped_frames)
}

fn run_playback(
    cmd_rx: pw::channel::Receiver<()>,
    rx: Receiver<Vec<i16>>,
    target_node: Option<String>,
    fill_gauge: Arc<AtomicUsize>,
) -> Result<(), AudioError> {
    let mainloop = pw::main_loop::MainLoopRc::new(None)
        .map_err(|e| AudioError::Init(e.to_string()))?;
    let context = pw::context::ContextRc::new(&mainloop, None)
        .map_err(|e| AudioError::Init(e.to_string()))?;
    let core = context.connect_rc(None)
        .map_err(|e| AudioError::Init(e.to_string()))?;

    // Ring buffer setup: 19200 interleaved samples (200ms capacity for stereo
    // 48kHz).
    const RING_CAPACITY: usize = 9600 * crate::audio::PLAYBACK_CHANNELS;
    let rb = HeapRb::<i16>::new(RING_CAPACITY);
    let (mut producer, consumer) = rb.split();

    // Prefill to the target depth so playout starts at its steady-state level:
    // the mixer keeps the ring near `PLAYBACK_TARGET_SAMPLES` by reading
    // `fill_gauge`, so production tracks the PipeWire hardware clock instead of
    // a fixed timer.
    //
    // `fill_gauge` is an EXACT occupancy counter, maintained by deltas: the
    // worker `fetch_add`s every pushed sample, the RT callback `fetch_sub`s
    // every popped one. We can't use ringbuf's `occupied_len()` for this — the
    // split producer/consumer keep *cached* head/tail indices, so their length
    // observers are approximate and stale, which would feed the mixer a fill
    // reading that lies high and starve the ring. (The try_push/try_pop data
    // path itself is exact; only the length observers are cached.)
    for _ in 0..crate::audio::PLAYBACK_TARGET_SAMPLES {
        let _ = producer.try_push(0);
    }
    fill_gauge.store(crate::audio::PLAYBACK_TARGET_SAMPLES, Ordering::Relaxed);

    // Playout-health instrumentation (diagnostic). `underrun_samples` counts
    // samples the RT callback had to substitute with silence because the ring
    // was empty (clicks); `dropped_frames` counts whole frames the worker
    // discarded because the ring was full (overrun). With clock-paced
    // production both should stay at zero; a steady non-zero trend means the
    // pacing isn't keeping up, bursts mean scheduling jitter. All RT-safe: the
    // callback does a single wait-free fetch_add/store per quantum.
    let underrun_samples = Arc::new(AtomicU64::new(0));
    let dropped_frames = Arc::new(AtomicU64::new(0));
    // Diagnostic: the per-cycle frame count the RT callback is asked to fill
    // (`slice.len()/stride`) and how many callbacks fire per second. If the
    // count is ~1024 (the graph quantum) at ~47/s the consumer is normal; if
    // it's large (the buffer maxsize) we're over-pulling the whole slice
    // instead of the requested quantum — the leading suspect for the ~46%
    // underrun. See handoff.md 2026-05-31 PM entry.
    let last_quantum = Arc::new(AtomicUsize::new(0));
    let callback_count = Arc::new(AtomicU64::new(0));

    // Command receiver to quit main loop
    let mainloop_clone = mainloop.clone();
    let _cmd_recv = cmd_rx.attach(mainloop.loop_(), move |_| {
        mainloop_clone.quit();
    });

    let mut props = properties! {
        *pw::keys::MEDIA_TYPE => "Audio",
        *pw::keys::MEDIA_CATEGORY => "Playback",
        *pw::keys::MEDIA_ROLE => "Communication",
        // Low-latency hint (~21ms @ 48kHz). On its own this does NOT shrink the
        // buffer — the real fix is the explicit Buffers param below — but it
        // expresses the intended quantum for any node that honours it.
        *pw::keys::NODE_LATENCY => "1024/48000",
    };
    if let Some(target) = target_node {
        props.insert("node.target", target);
    }

    let stream = pw::stream::StreamBox::new(&core, "peerspeak-playback-stream", props)
        .map_err(|e| AudioError::Init(e.to_string()))?;

    struct PlaybackUserData<C: ringbuf::traits::Consumer<Item = i16>> {
        consumer: C,
        underrun_samples: Arc<AtomicU64>,
        fill_gauge: Arc<AtomicUsize>,
        last_quantum: Arc<AtomicUsize>,
        callback_count: Arc<AtomicU64>,
    }

    let _listener = stream
        .add_local_listener_with_user_data(PlaybackUserData {
            consumer,
            underrun_samples: underrun_samples.clone(),
            fill_gauge: fill_gauge.clone(),
            last_quantum: last_quantum.clone(),
            callback_count: callback_count.clone(),
        })
        .process(|stream, user_data| {
            if let Some(mut buffer) = stream.dequeue_buffer() {
                // Read the graph's requested quantum BEFORE the mutable
                // `datas_mut()` borrow below (`requested()` borrows `&buffer`).
                let requested = buffer.requested() as usize;
                let datas = buffer.datas_mut();
                if !datas.is_empty() {
                    let data = &mut datas[0];
                    let mut total_size = 0;
                    if let Some(slice) = data.data() {
                        let stride = 2 * crate::audio::PLAYBACK_CHANNELS; // S16LE stereo
                        // Fill exactly what the graph asked for this cycle (with
                        // a safe fallback), never the whole mapped slice — that
                        // over-pull past the ring depth was the original crackle.
                        let mapped_frames = slice.len() / stride;
                        let n_frames = frames_to_produce(requested, mapped_frames);
                        // Diagnostic: record the quantum the device asked for and
                        // count the callback. Wait-free, RT-safe.
                        user_data.last_quantum.store(n_frames, Ordering::Relaxed);
                        user_data.callback_count.fetch_add(1, Ordering::Relaxed);
                        let mut starved = 0u64;
                        for i in 0..n_frames {
                            let start = i * stride;
                            for ch in 0..crate::audio::PLAYBACK_CHANNELS {
                                let val = match user_data.consumer.try_pop() {
                                    Some(v) => v,
                                    None => {
                                        starved += 1;
                                        0
                                    }
                                };
                                let bytes = val.to_le_bytes();
                                let offset = start + ch * 2;
                                slice[offset] = bytes[0];
                                slice[offset + 1] = bytes[1];
                            }
                        }
                        if starved > 0 {
                            // One wait-free atomic add per quantum — RT-safe.
                            user_data.underrun_samples.fetch_add(starved, Ordering::Relaxed);
                        }
                        // Decrement the exact occupancy counter by the samples we
                        // actually pulled (excluding underruns, which removed
                        // nothing) so the mixer paces against true ring depth.
                        // Wait-free fetch_sub, RT-safe.
                        let requested_samples = n_frames * crate::audio::PLAYBACK_CHANNELS;
                        let popped = requested_samples - starved as usize;
                        if popped > 0 {
                            user_data.fill_gauge.fetch_sub(popped, Ordering::Relaxed);
                        }
                        total_size = n_frames * stride;
                    }
                    let chunk = data.chunk_mut();
                    *chunk.offset_mut() = 0;
                    *chunk.stride_mut() = (2 * crate::audio::PLAYBACK_CHANNELS) as _;
                    *chunk.size_mut() = total_size as _;
                }
            }
        })
        .register()
        .map_err(|e| AudioError::Init(e.to_string()))?;

    let mut audio_info = spa::param::audio::AudioInfoRaw::new();
    audio_info.set_format(spa::param::audio::AudioFormat::S16LE);
    audio_info.set_rate(48000);
    audio_info.set_channels(crate::audio::PLAYBACK_CHANNELS as u32); // Stereo playback

    let obj = pw::spa::pod::Object {
        type_: pw::spa::utils::SpaTypes::ObjectParamFormat.as_raw(),
        id: pw::spa::param::ParamType::EnumFormat.as_raw(),
        properties: audio_info.into(),
    };
    let values: Vec<u8> = pw::spa::pod::serialize::PodSerializer::serialize(
        std::io::Cursor::new(Vec::new()),
        &pw::spa::pod::Value::Object(obj),
    )
    .unwrap()
    .0
    .into_inner();

    // Explicit Buffers param. History: without ANY Buffers param PipeWire handed
    // this stream a ~256ms (12288-frame) maxsize buffer and called `process`
    // only ~4x/sec; the callback then over-pulled the whole 12288-frame slice
    // (far more than the 200ms ring holds → ~half silence → the crackle). The
    // first fix pinned the size to exactly one 1024-frame quantum, which only
    // works when the graph quantum IS 1024 — on other hardware (quantum 512 or
    // 2048) the pinned slice mismatches the device's per-cycle demand and the
    // crackle returns. The correct fix: give the buffer enough room for any
    // plausible quantum (the desktop's quantum-limit is 8192) and let the RT
    // callback fill exactly `Buffer::requested()` frames per cycle (see
    // `frames_to_produce`). `requested()`, not the buffer size, now governs
    // per-cycle output, so this is a generous max rather than a hard pin.
    const MAX_QUANTUM_FRAMES: i32 = 8192;
    const STRIDE: i32 = 2 * crate::audio::PLAYBACK_CHANNELS as i32; // S16LE stereo
    let buffers_obj = pw::spa::pod::Object {
        type_: pw::spa::utils::SpaTypes::ObjectParamBuffers.as_raw(),
        id: pw::spa::param::ParamType::Buffers.as_raw(),
        properties: vec![
            // Let PipeWire pick the buffer count (>=2 for double-buffering).
            pw::spa::pod::Property::new(
                pw::spa::sys::SPA_PARAM_BUFFERS_buffers,
                pw::spa::pod::Value::Choice(pw::spa::pod::ChoiceValue::Int(
                    pw::spa::utils::Choice(
                        pw::spa::utils::ChoiceFlags::empty(),
                        pw::spa::utils::ChoiceEnum::Range { default: 8, min: 2, max: 64 },
                    ),
                )),
            ),
            pw::spa::pod::Property::new(
                pw::spa::sys::SPA_PARAM_BUFFERS_blocks,
                pw::spa::pod::Value::Int(1),
            ),
            pw::spa::pod::Property::new(
                pw::spa::sys::SPA_PARAM_BUFFERS_size,
                pw::spa::pod::Value::Int(MAX_QUANTUM_FRAMES * STRIDE),
            ),
            pw::spa::pod::Property::new(
                pw::spa::sys::SPA_PARAM_BUFFERS_stride,
                pw::spa::pod::Value::Int(STRIDE),
            ),
        ],
    };
    let buffers_values: Vec<u8> = pw::spa::pod::serialize::PodSerializer::serialize(
        std::io::Cursor::new(Vec::new()),
        &pw::spa::pod::Value::Object(buffers_obj),
    )
    .unwrap()
    .0
    .into_inner();

    let mut params = [
        Pod::from_bytes(&values).unwrap(),
        Pod::from_bytes(&buffers_values).unwrap(),
    ];

    stream.connect(
        spa::utils::Direction::Output,
        None,
        pw::stream::StreamFlags::AUTOCONNECT
            | pw::stream::StreamFlags::MAP_BUFFERS
            | pw::stream::StreamFlags::RT_PROCESS,
        &mut params,
    )
    .map_err(|e| AudioError::Stream(e.to_string()))?;

    // Spawn a worker thread to read from rx and push to producer
    let running = Arc::new(AtomicBool::new(true));
    let running_clone = running.clone();
    let worker_dropped = dropped_frames.clone();
    let worker_fill = fill_gauge.clone();
    let worker_handle = thread::spawn(move || {
        drain_loop(&rx, &running_clone, |frame| {
            // Drop the whole frame (rather than tearing it) only if it truly
            // won't fit — checked against the exact occupancy counter, not
            // the ring's stale length observer. With clock-paced production
            // this should never fire.
            if worker_fill.load(Ordering::Relaxed) + frame.len() > RING_CAPACITY {
                worker_dropped.fetch_add(1, Ordering::Relaxed);
                return;
            }
            // Reserve occupancy BEFORE publishing samples. Otherwise the RT
            // consumer can pop a newly-visible sample before it is counted and
            // wrap the exact fill gauge to usize::MAX, wedging mixer pacing.
            // `push_slice` also publishes the frame as one operation rather than
            // exposing a half-written stereo pair.
            publish_frame(&worker_fill, &worker_dropped, &mut producer, &frame, || {});
        });
    });

    // Diagnostic logger: once per second, report the ring fill and the delta in
    // underrun samples / dropped frames since the last report. Quiet line (all
    // zeros) means the local playout path is healthy; a steady non-zero trend is
    // the clock-drift signature, bursts are scheduling jitter.
    let log_running = running.clone();
    let log_underrun = underrun_samples.clone();
    let log_dropped = dropped_frames.clone();
    let log_fill = fill_gauge.clone();
    let log_quantum = last_quantum.clone();
    let log_callbacks = callback_count.clone();
    // In normal operation this stays quiet — it only logs a second where the
    // playout actually glitched (underrun or dropped > 0). Set
    // PEERSPEAK_AUDIO_VERBOSE=1 (the `audio_probe` tool does) to get the full
    // once-per-second heartbeat for diagnostics.
    let verbose = std::env::var_os("PEERSPEAK_AUDIO_VERBOSE").is_some();
    let logger_handle = thread::spawn(move || {
        let (mut last_u, mut last_d, mut last_c) = (0u64, 0u64, 0u64);
        while log_running.load(Ordering::Relaxed) {
            thread::sleep(Duration::from_secs(1));
            let u = log_underrun.load(Ordering::Relaxed);
            let d = log_dropped.load(Ordering::Relaxed);
            let fill = log_fill.load(Ordering::Relaxed);
            let q = log_quantum.load(Ordering::Relaxed);
            let c = log_callbacks.load(Ordering::Relaxed);
            let (du, dd, dc) = (u - last_u, d - last_d, c - last_c);
            last_u = u;
            last_d = d;
            last_c = c;
            if verbose || du > 0 || dd > 0 {
                crate::log_msg(&format!(
                    "playout-health: fill={fill} samples (~{}ms) | underrun +{du} samples/s (total {u}) | dropped +{dd} frames/s (total {d}) | quantum={q} frames, {dc} callbacks/s",
                    fill / (48 * crate::audio::PLAYBACK_CHANNELS),
                ));
            }
        }
    });

    mainloop.run();

    running.store(false, Ordering::Relaxed);
    let _ = worker_handle.join();
    let _ = logger_handle.join();

    Ok(())
}

#[cfg(test)]
mod tests {
    use super::{drain_loop, for_each_capture_sample, frames_to_produce, publish_frame};
    use ringbuf::{HeapRb, traits::{Consumer, Producer, Split}};
    use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
    use std::sync::{Arc, Mutex};
    use std::time::Duration;
    use std::{sync::mpsc, thread};

    #[test]
    fn requested_in_range_is_honored() {
        // The graph's requested quantum is produced verbatim when it fits.
        assert_eq!(frames_to_produce(512, 8192), 512);
        assert_eq!(frames_to_produce(1024, 8192), 1024);
        assert_eq!(frames_to_produce(2048, 8192), 2048);
    }

    #[test]
    fn requested_over_mapped_is_clamped() {
        // Never write past the mapped slice, even if the graph asks for more.
        assert_eq!(frames_to_produce(8192, 1024), 1024);
        assert_eq!(frames_to_produce(2048, 2048), 2048);
    }

    #[test]
    fn zero_requested_uses_safe_fallback() {
        // requested()==0 (first buffers / non-driver cycles) → capped fallback,
        // never the whole mapped slice (that's the over-pull crackle).
        assert_eq!(frames_to_produce(0, 8192), 1024);
        // ...still clamped to a small mapped slice.
        assert_eq!(frames_to_produce(0, 256), 256);
    }

    #[test]
    fn degenerate_zero_mapped_is_zero() {
        // No buffer to write into → produce nothing, no panic / underflow.
        assert_eq!(frames_to_produce(0, 0), 0);
        assert_eq!(frames_to_produce(1024, 0), 0);
    }

    #[test]
    fn capture_size_larger_than_mapping_is_clamped() {
        let mut samples = Vec::new();
        for_each_capture_sample(&[1, 0, 2, 0, 3], usize::MAX, |sample| {
            samples.push(sample)
        });
        assert_eq!(samples, vec![1, 2]);
    }

    #[test]
    fn occupancy_is_reserved_before_frame_is_published() {
        let rb = HeapRb::<i16>::new(8);
        let (mut producer, mut consumer) = rb.split();
        assert!(producer.try_push(7).is_ok());

        let fill = AtomicUsize::new(1);
        let dropped = AtomicU64::new(0);
        publish_frame(&fill, &dropped, &mut producer, &[10, 11], || {
            // Force the consumer to drain the old sample after the new frame's
            // occupancy is reserved but before that frame is published.
            assert_eq!(consumer.try_pop(), Some(7));
            assert_eq!(fill.fetch_sub(1, Ordering::Relaxed), 3);
        });

        assert_eq!(fill.load(Ordering::Relaxed), 2);
        assert_eq!(consumer.try_pop(), Some(10));
        assert_eq!(fill.fetch_sub(1, Ordering::Relaxed), 2);
        assert_eq!(consumer.try_pop(), Some(11));
        assert_eq!(fill.fetch_sub(1, Ordering::Relaxed), 1);
        assert_eq!(fill.load(Ordering::Relaxed), 0);
        assert_eq!(dropped.load(Ordering::Relaxed), 0);
    }

    // --- drain_loop (A7: worker must not hang shutdown) ---

    #[test]
    fn drain_loop_exits_when_running_flips_even_with_sender_alive() {
        // The exact A7 hang scenario: the frame sender is STILL alive (never
        // dropped) when `running` goes false. A blocking `recv()` would park
        // forever; `drain_loop` must wake within a poll interval and return.
        let (tx, rx) = mpsc::channel::<Vec<i16>>();
        let running = Arc::new(AtomicBool::new(true));
        let r2 = running.clone();
        let h = thread::spawn(move || drain_loop(&rx, &r2, |_| {}));
        // Let it park in recv_timeout, then signal stop.
        thread::sleep(Duration::from_millis(50));
        running.store(false, Ordering::Relaxed);
        // Wait comfortably longer than one poll interval; it must have exited.
        thread::sleep(super::WORKER_POLL + Duration::from_millis(150));
        assert!(
            h.is_finished(),
            "drain_loop must exit after running=false even while the sender is alive"
        );
        drop(tx); // keep tx alive until here so the test really covers the case
        h.join().unwrap();
    }

    #[test]
    fn drain_loop_returns_on_disconnect() {
        // Sender dropped before we start → recv errors Disconnected → return.
        // If this hangs, the test runner times out (i.e. it would fail loudly).
        let (tx, rx) = mpsc::channel::<Vec<i16>>();
        let running = Arc::new(AtomicBool::new(true));
        drop(tx);
        drain_loop(&rx, &running, |_| panic!("no frame should arrive"));
    }

    #[test]
    fn drain_loop_delivers_frames_to_callback() {
        let (tx, rx) = mpsc::channel::<Vec<i16>>();
        let running = Arc::new(AtomicBool::new(true));
        let r2 = running.clone();
        let got = Arc::new(Mutex::new(Vec::new()));
        let g2 = got.clone();
        let h = thread::spawn(move || drain_loop(&rx, &r2, |f| g2.lock().unwrap().push(f)));
        tx.send(vec![1, 2, 3]).unwrap();
        tx.send(vec![4, 5]).unwrap();
        thread::sleep(Duration::from_millis(50));
        running.store(false, Ordering::Relaxed);
        drop(tx);
        h.join().unwrap();
        let got = got.lock().unwrap();
        assert_eq!(*got, vec![vec![1, 2, 3], vec![4, 5]]);
    }
}