djdeck/src/audio.rs

/// Audio engine
///
/// Responsibilities:
///   - Decode audio files (symphonia) into interleaved stereo f32 PCM
///   - Drive one PipeWire playback stream per deck
///   - Run the per-deck DSP graph (`crate::dsp::DeckDsp`) inside the realtime
///     process callback: pitch + tempo + cutoff + reverb + fader, all in-process
///   - Stream connects with AUTOCONNECT to the default sink (no filter chains,
///     no `node.target`)
///   - Effect parameters flow lock-free via `Arc<DeckParams>` atomics shared
///     with the app
use std::{
    path::PathBuf,
    sync::{Arc, Mutex, atomic::Ordering::Relaxed},
    thread,
};

use anyhow::{Result, anyhow};
use crossbeam_channel::{Receiver, Sender, bounded};
use symphonia::core::{
    audio::GenericAudioBufferRef,
    codecs::{CodecParameters, audio::AudioDecoderOptions},
    formats::{FormatOptions, FormatReader, TrackType, probe::Hint},
    io::MediaSourceStream,
    meta::{MetadataOptions, StandardTag},
};

use crate::{
    deck::{DeckId, next_bar_sample},
    dsp::{DeckDsp, DeckParams, TEMPO_MAX, TEMPO_MIN},
};

// ─── Commands ────────────────────────────────────────────────────────────────

#[derive(Debug)]
pub enum AudioCmd {
    Load { deck: DeckId, path: PathBuf },
    Play(DeckId),
    Pause(DeckId),
    Stop(DeckId),
    Seek { deck: DeckId, position: f32 },
    SetCue(DeckId),
    GotoCue(DeckId),
    /// Append a decoded PCM chunk (interleaved stereo f32) to a deck's buffer.
    /// `load_gen` must match the deck's current generation to guard against
    /// stale decodes from a previous Load.
    AppendPcm { deck: DeckId, load_gen: u64, frames: Vec<f32> },
    /// Signal that decoding is complete for this generation.
    MarkDecodingDone { deck: DeckId, load_gen: u64 },
    /// Forwarded by the app after receiving `AudioStatus::BeatGrid`. Stores
    /// the grid on the engine-side `DeckState` so SyncPlay can do bar-math.
    SetBeatGrid { deck: DeckId, beats: Vec<u64>, bpm: f32 },
    /// Beat-aligned auto-mix start: the slave deck waits until the master's
    /// next bar boundary, then begins on its own first detected downbeat
    /// with tempo matched to master's effective BPM.
    SyncPlay {
        deck:           DeckId,
        master:         DeckId,
        bars_to_wait:   u32,
        bass_kill_bars: u32,
    },
}

// ─── Status updates ──────────────────────────────────────────────────────────

#[derive(Debug, Clone)]
pub enum AudioStatus {
    /// Sent immediately from file headers — before full decode.
    Metadata {
        deck:             DeckId,
        name:             String,
        artist:           Option<String>,
        album:            Option<String>,
        /// Estimated from codec_params; exact after full decode.
        duration_samples: u64,
        sample_rate:      u32,
    },
    /// Waveform peak data sent in chunks as the file decodes (~4096-sample chunks).
    /// Each entry is `[bass_peak, mid_peak, high_peak]` — peak amplitudes after
    /// splitting the mono mix through a Linkwitz-Riley-ish crossover (LP 250 Hz,
    /// BP 250–4000 Hz, HP 4000 Hz). The UI blends these into per-column color.
    WaveformChunk {
        deck:        DeckId,
        bands:       Vec<[f32; 3]>,
        /// Sample-frame offset of the first peak in this chunk.
        frame_offset: u64,
    },
    /// Beat grid + BPM estimate from `timestretch::detect_beat_grid` over the
    /// first ~60 s of decoded audio. Emitted once per Load, after `Ready`.
    /// `beats` contains sample-frame positions; the first entry is treated as
    /// the downbeat of bar 1 (the detector does not label downbeats).
    BeatGrid { deck: DeckId, beats: Vec<u64>, bpm: f32 },
    /// A sync-play has been armed: the slave deck is queued and will start
    /// at the master's next bar boundary. UI uses this to flip the deck
    /// into `PlayState::WaitingForSync`.
    SyncQueued { deck: DeckId, master: DeckId },
    /// The wait fired — slave is now playing in phase with master.
    SyncFired { deck: DeckId },
    /// Decoded PCM chunk — routed back through the app into AudioCmd::AppendPcm.
    PcmChunk {
        deck:        DeckId,
        load_gen:    u64,
        frames:      Vec<f32>,
        sample_rate: u32,
    },
    /// Full PCM is in memory; deck is ready to play.
    Ready {
        deck:             DeckId,
        load_gen:         u64,
        duration_samples: u64,
        sample_rate:      u32,
    },
    Position { deck: DeckId, position: u64 },
    Vu       { deck: DeckId, l: f32, r: f32 },
    TrackEnded(DeckId),
    Error { deck: DeckId, msg: String },
}

// ─── Per-deck internal state ──────────────────────────────────────────────────

/// Pending sync-play state. The slave callback fires when master's published
/// `position` atomic crosses `master_target_frame`.
#[derive(Debug, Clone)]
pub(crate) struct SyncTarget {
    pub master_idx:           usize,
    pub master_target_frame:  u64,
    pub slave_start_frame:    u64,
    pub bass_kill_out_frames: u64,
}

struct DeckState {
    id:      DeckId,
    playing: bool,
    /// Play was requested before PCM arrived; start as soon as first chunk lands.
    pending_play: bool,
    /// True once the decode thread has sent every frame (safe to stop at EOF).
    decoding_done: bool,
    /// Incremented on each Load; guards AppendPcm from a stale decode thread.
    load_gen: u64,
    pcm:     Vec<f32>,
    sample_rate: u32,
    /// Frame index into `pcm` (no fractional resampling — the DSP stretcher
    /// handles tempo). Advanced by the number of frames consumed each block.
    read_pos: usize,
    cue_pos:  usize,
    /// Detected BPM, or 0.0 if unknown.
    bpm:      f32,
    /// Sample-frame positions of detected beats (engine-side copy of the
    /// app-visible `Deck::beat_grid`). Empty until decode finishes.
    beat_grid: Vec<u64>,
    /// Armed by `AudioCmd::SyncPlay`; consumed by the realtime callback when
    /// master's position crosses the target.
    sync_pending: Option<SyncTarget>,
    dsp:      DeckDsp,
    params:   Arc<DeckParams>,
    last_vu_update: std::time::Instant,
}

impl DeckState {
    fn new(id: DeckId, sample_rate: u32, params: Arc<DeckParams>) -> Self {
        Self {
            id,
            playing:       false,
            pending_play:  false,
            decoding_done: true,
            load_gen:      0,
            pcm:           Vec::new(),
            sample_rate,
            read_pos:      0,
            cue_pos:       0,
            bpm:           0.0,
            beat_grid:     Vec::new(),
            sync_pending:  None,
            dsp:           DeckDsp::new(sample_rate),
            params,
            last_vu_update: std::time::Instant::now(),
        }
    }

    fn total_frames(&self) -> usize { self.pcm.len() / 2 }
}

// ─── Public handle ────────────────────────────────────────────────────────────

pub struct AudioEngine {
    pub cmd_tx:    Sender<AudioCmd>,
    pub status_rx: Receiver<AudioStatus>,
    /// One `Arc<DeckParams>` per deck — the app writes effect parameters into
    /// these atomics directly. The realtime callback snapshots them per block.
    pub params:    Vec<Arc<DeckParams>>,
}

impl AudioEngine {
    pub fn new() -> Result<Self> {
        let (cmd_tx, cmd_rx)       = bounded::<AudioCmd>(64);
        let (status_tx, status_rx) = bounded::<AudioStatus>(1024);

        let params: Vec<Arc<DeckParams>> = (0..4).map(|_| DeckParams::new()).collect();
        let params_for_engine = params.clone();

        thread::Builder::new()
            .name("audio-engine".into())
            .spawn(move || {
                if let Err(e) = engine_thread(cmd_rx, status_tx, params_for_engine) {
                    tracing::error!("Audio engine crashed: {e}");
                }
            })?;

        Ok(Self { cmd_tx, status_rx, params })
    }
}

// ─── Engine thread ────────────────────────────────────────────────────────────

const STREAM_SR: u32 = 44_100;

fn engine_thread(
    cmd_rx: Receiver<AudioCmd>,
    status_tx: Sender<AudioStatus>,
    params: Vec<Arc<DeckParams>>,
) -> Result<()> {
    use pipewire::{context::ContextRc, main_loop::MainLoopRc, stream::StreamRc};

    let main_loop = MainLoopRc::new(None)?;
    let context   = ContextRc::new(&main_loop, None)?;
    let core      = context.connect_rc(None)?;

    let deck_states: Vec<Arc<Mutex<DeckState>>> = (0..4)
        .map(|i| {
            let id = DeckId::from_index(i).unwrap();
            Arc::new(Mutex::new(DeckState::new(id, STREAM_SR, params[i].clone())))
        })
        .collect();

    let mut _streams:   Vec<StreamRc> = Vec::new();
    let mut _listeners              = Vec::new();

    for (i, ds) in deck_states.iter().enumerate() {
        let deck_id      = DeckId::from_index(i).unwrap();
        let ds_cb        = Arc::clone(ds);
        let status_tx_cb = status_tx.clone();
        // Per-callback view of every deck's `Arc<DeckParams>` — needed so the
        // slave's callback can lock-free peek the master's `position` atomic
        // to decide when its sync wait should fire.
        let params_all   = params.clone();

        let props = pipewire::properties::properties! {
            *pipewire::keys::MEDIA_TYPE     => "Audio",
            *pipewire::keys::MEDIA_CATEGORY => "Playback",
            *pipewire::keys::MEDIA_ROLE     => "Music",
            *pipewire::keys::APP_NAME       => "djdeck",
            *pipewire::keys::NODE_NAME      => format!("djdeck.deck.{}", deck_id.node_suffix()),
        };

        let stream = StreamRc::new(
            core.clone(),
            &format!("djdeck-deck-{}", deck_id.label()),
            props,
        )?;

        let listener = stream
            .add_local_listener_with_user_data(())
            .process(move |stream: &pipewire::stream::Stream, _| {
                let mut ds_guard = ds_cb.lock();
                let ds = ds_guard.as_mut().unwrap();

                if let Some(mut buf) = stream.dequeue_buffer() {
                    let datas = buf.datas_mut();
                    if datas.is_empty() { return; }
                    let data = &mut datas[0];
                    let out_bytes = data.data().unwrap();
                    let n_frames  = out_bytes.len() / 8;

                    let out = unsafe {
                        std::slice::from_raw_parts_mut(
                            out_bytes.as_ptr() as *mut f32,
                            n_frames * 2,
                        )
                    };

                    // ── Sync-play wait: fire if master crossed the target ───
                    if let Some(t) = ds.sync_pending.clone() {
                        let master_pos = params_all[t.master_idx].position.load(Relaxed);
                        if master_pos >= t.master_target_frame {
                            ds.read_pos      = (t.slave_start_frame as usize)
                                                .min(ds.total_frames());
                            ds.playing       = true;
                            ds.pending_play  = false;
                            ds.sync_pending  = None;
                            ds.params.bass_kill_remaining_out_frames
                                .store(t.bass_kill_out_frames, Relaxed);
                            ds.params.bass_kill_initial_out_frames
                                .store(t.bass_kill_out_frames, Relaxed);
                            let _ = status_tx_cb.try_send(
                                AudioStatus::SyncFired { deck: ds.id });
                        }
                    }

                    if ds.playing && !ds.pcm.is_empty() {
                        let total = ds.total_frames();
                        if ds.read_pos >= total {
                            if ds.decoding_done {
                                ds.playing = false;
                                let _ = status_tx_cb.try_send(AudioStatus::TrackEnded(ds.id));
                            }
                            out.fill(0.0);
                        } else {
                            // Split disjoint field borrows so the DSP can read
                            // `pcm` while mutating its internal state.
                            let start  = ds.read_pos;
                            let DeckState { dsp, pcm, params, .. } = &mut **ds;
                            let consumed = dsp.process(params, pcm, start, out);
                            ds.read_pos = (start + consumed).min(total);

                            // VU from the post-DSP output.
                            let mut vu_l = 0.0f32;
                            let mut vu_r = 0.0f32;
                            for i in 0..n_frames {
                                vu_l = vu_l.max(out[i * 2].abs());
                                vu_r = vu_r.max(out[i * 2 + 1].abs());
                            }

                            let now = std::time::Instant::now();
                            if now.duration_since(ds.last_vu_update) >= std::time::Duration::from_millis(16) {
                                let _ = status_tx_cb.try_send(AudioStatus::Vu { deck: ds.id, l: vu_l, r: vu_r });
                                ds.last_vu_update = now;
                            }
                        }
                    } else {
                        out.fill(0.0);
                    }

                    {
                        let chunk = data.chunk_mut();
                        *chunk.offset_mut() = 0;
                        *chunk.stride_mut() = 8;
                        *chunk.size_mut()   = (n_frames * 8) as u32;
                    }

                    // Publish source position for cross-deck sync visibility,
                    // and tick the bass-kill timer down in wall-clock frames.
                    ds.params.position.store(ds.read_pos as u64, Relaxed);
                    if ds.playing {
                        let bk = ds.params.bass_kill_remaining_out_frames.load(Relaxed);
                        ds.params.bass_kill_remaining_out_frames
                            .store(bk.saturating_sub(n_frames as u64), Relaxed);
                    }

                    let _ = status_tx_cb.try_send(AudioStatus::Position {
                        deck: ds.id,
                        position: ds.read_pos as u64,
                    });
                }
            })
            .register()?;

        let pod_bytes = make_f32_stereo_pod()?;
        let mut params = [pipewire::spa::pod::Pod::from_bytes(&pod_bytes).unwrap()];
        stream.connect(
            pipewire::spa::utils::Direction::Output,
            None,
            pipewire::stream::StreamFlags::AUTOCONNECT
                | pipewire::stream::StreamFlags::MAP_BUFFERS
                | pipewire::stream::StreamFlags::RT_PROCESS,
            &mut params,
        )?;

        _listeners.push(listener);
        _streams.push(stream);
    }

    // Command processing thread
    let ds_cmd     = deck_states.clone();
    let st_cmd     = status_tx.clone();
    let params_cmd = params.clone();
    thread::spawn(move || {
        for cmd in cmd_rx.iter() {
            handle_cmd(cmd, &ds_cmd, &params_cmd, &st_cmd);
        }
    });

    main_loop.run();
    Ok(())
}

// ─── Command handler ──────────────────────────────────────────────────────────

fn handle_cmd(
    cmd: AudioCmd,
    states: &[Arc<Mutex<DeckState>>],
    params: &[Arc<DeckParams>],
    status_tx: &Sender<AudioStatus>,
) {
    match cmd {
        AudioCmd::Load { deck, path } => {
            let load_gen = {
                let mut ds = states[deck as usize].lock().unwrap();
                ds.pcm.clear();
                ds.read_pos    = 0;
                ds.cue_pos     = 0;
                ds.playing     = false;
                ds.pending_play = false;
                ds.decoding_done = false;
                ds.load_gen    = ds.load_gen.wrapping_add(1);
                ds.load_gen
            };
            let st = status_tx.clone();
            thread::spawn(move || {
                if let Err(e) = decode_file_streaming(&path, deck, load_gen, &st) {
                    let _ = st.send(AudioStatus::Error { deck, msg: e.to_string() });
                }
            });
        }
        AudioCmd::Play(deck) => {
            let mut ds = states[deck as usize].lock().unwrap();
            if !ds.pcm.is_empty() {
                ds.playing = true;
            } else {
                ds.pending_play = true;
            }
        }
        AudioCmd::Pause(deck) => {
            let mut ds = states[deck as usize].lock().unwrap();
            ds.playing      = false;
            ds.pending_play = false;
        }
        AudioCmd::Stop(deck) => {
            let mut ds = states[deck as usize].lock().unwrap();
            ds.playing      = false;
            ds.pending_play = false;
            ds.read_pos     = 0;
        }
        AudioCmd::Seek { deck, position } => {
            let mut ds = states[deck as usize].lock().unwrap();
            let total = ds.total_frames();
            ds.read_pos = ((total as f32 * position.clamp(0.0, 1.0)) as usize).min(total);
        }
        AudioCmd::SetCue(deck) => {
            let mut ds = states[deck as usize].lock().unwrap();
            ds.cue_pos = ds.read_pos;
        }
        AudioCmd::GotoCue(deck) => {
            let mut ds = states[deck as usize].lock().unwrap();
            ds.read_pos = ds.cue_pos;
        }
        AudioCmd::AppendPcm { deck, load_gen, frames } => {
            let mut ds = states[deck as usize].lock().unwrap();
            if ds.load_gen != load_gen { return; } // stale decode
            let was_empty = ds.pcm.is_empty();
            ds.pcm.extend_from_slice(&frames);
            if was_empty && ds.pending_play {
                ds.playing      = true;
                ds.pending_play = false;
            }
        }
        AudioCmd::MarkDecodingDone { deck, load_gen } => {
            let mut ds = states[deck as usize].lock().unwrap();
            if ds.load_gen == load_gen {
                ds.decoding_done = true;
            }
        }
        AudioCmd::SetBeatGrid { deck, beats, bpm } => {
            let mut ds = states[deck as usize].lock().unwrap();
            ds.bpm       = bpm;
            ds.beat_grid = beats;
        }
        AudioCmd::SyncPlay { deck, master, bars_to_wait, bass_kill_bars } => {
            if deck == master {
                tracing::warn!("SyncPlay refused: deck and master are both {}", deck.label());
                return;
            }
            let slave_idx  = deck   as usize;
            let master_idx = master as usize;

            // Snapshot master state — published position is fresh enough
            // (last realtime block, < ~25 ms ago).
            let master_pos   = params[master_idx].position.load(Relaxed);
            let master_tempo = params[master_idx].tempo.load(Relaxed);
            let (master_bpm, target_frame) = {
                let m = states[master_idx].lock().unwrap();
                let t = next_bar_sample(&m.beat_grid, master_pos, bars_to_wait);
                (m.bpm, t)
            };

            let Some(target_frame) = target_frame else {
                tracing::warn!(
                    "SyncPlay deck {}: master {} has no beat grid (or no more bars). Falling back to immediate Play.",
                    deck.label(), master.label()
                );
                let mut ds = states[slave_idx].lock().unwrap();
                if !ds.pcm.is_empty() { ds.playing = true; } else { ds.pending_play = true; }
                return;
            };
            if master_bpm <= 0.0 {
                tracing::warn!("SyncPlay deck {}: master has no BPM; immediate Play.", deck.label());
                let mut ds = states[slave_idx].lock().unwrap();
                if !ds.pcm.is_empty() { ds.playing = true; } else { ds.pending_play = true; }
                return;
            }

            // Slave-side beat info.
            let (slave_bpm, slave_first_beat) = {
                let s = states[slave_idx].lock().unwrap();
                (s.bpm, s.beat_grid.first().copied())
            };
            let Some(slave_first_beat) = slave_first_beat else {
                tracing::warn!(
                    "SyncPlay deck {}: slave has no beat grid yet; immediate Play.",
                    deck.label()
                );
                let mut ds = states[slave_idx].lock().unwrap();
                if !ds.pcm.is_empty() { ds.playing = true; } else { ds.pending_play = true; }
                return;
            };
            if slave_bpm <= 0.0 {
                let mut ds = states[slave_idx].lock().unwrap();
                if !ds.pcm.is_empty() { ds.playing = true; } else { ds.pending_play = true; }
                return;
            }

            // Beat-match.
            let master_eff_bpm = master_bpm * master_tempo;
            let slave_tempo    = (master_eff_bpm / slave_bpm).clamp(TEMPO_MIN, TEMPO_MAX);
            params[slave_idx].tempo.store(slave_tempo, Relaxed);

            let bass_kill_out_frames = if bass_kill_bars > 0 {
                ((bass_kill_bars as f32 * 4.0 * 60.0 / master_eff_bpm)
                    * STREAM_SR as f32) as u64
            } else { 0 };

            {
                let mut ds = states[slave_idx].lock().unwrap();
                ds.sync_pending = Some(SyncTarget {
                    master_idx,
                    master_target_frame: target_frame,
                    slave_start_frame:   slave_first_beat,
                    bass_kill_out_frames,
                });
                ds.playing      = false;
                ds.pending_play = false;
            }

            tracing::info!(
                "SyncPlay armed: deck {} → master {} at frame {} (tempo {:.3}, bass-kill {} frames)",
                deck.label(), master.label(), target_frame, slave_tempo, bass_kill_out_frames
            );
            let _ = status_tx.send(AudioStatus::SyncQueued { deck, master });
        }
    }
}

// ─── Symphonia streaming decode ───────────────────────────────────────────────

/// Number of PCM stereo frames per waveform peak bucket (~0.09 s at 44.1 kHz).
const PEAK_BUCKET: usize = 4096;

/// Open file, emit Metadata immediately, then decode to PCM while streaming
/// both waveform peaks and raw PCM chunks every 64 peak-buckets (~6 s).
fn decode_file_streaming(
    path: &PathBuf,
    deck: DeckId,
    load_gen: u64,
    status_tx: &Sender<AudioStatus>,
) -> Result<()> {
    let name = path
        .file_name()
        .and_then(|n| n.to_str())
        .unwrap_or("unknown")
        .to_string();

    let file = std::fs::File::open(path)?;
    let mss  = MediaSourceStream::new(Box::new(file), Default::default());

    let mut hint = Hint::new();
    if let Some(ext) = path.extension().and_then(|e| e.to_str()) {
        hint.with_extension(ext);
    }

    let mut format = symphonia::default::get_probe().probe(
        &hint,
        mss,
        FormatOptions::default(),
        MetadataOptions::default(),
    )?;

    let track = format
        .default_track(TrackType::Audio)
        .ok_or_else(|| anyhow!("No audio track found"))?
        .clone();

    let audio_params = match &track.codec_params {
        Some(CodecParameters::Audio(p)) => p.clone(),
        _ => return Err(anyhow!("track has no audio codec parameters")),
    };

    let sr = audio_params.sample_rate.unwrap_or(44100);

    let est_duration = track.num_frames.unwrap_or(0);

    let (artist, album) = extract_tags(&mut format);

    let _ = status_tx.send(AudioStatus::Metadata {
        deck,
        name,
        artist,
        album,
        duration_samples: est_duration,
        sample_rate: sr,
    });

    let mut decoder = symphonia::default::get_codecs()
        .make_audio_decoder(&audio_params, &AudioDecoderOptions::default())?;

    let track_id = track.id;

    let mut pcm_batch:    Vec<f32> = Vec::with_capacity(64 * PEAK_BUCKET * 2);
    let mut total_frames: u64 = 0;
    // Mono accumulator used for BPM detection. Capped at 60 s.
    const BPM_MAX_SAMPLES_SECS: u64 = 60;
    let bpm_cap = (sr as u64 * BPM_MAX_SAMPLES_SECS) as usize;
    let mut mono_for_bpm: Vec<f32> = Vec::with_capacity(bpm_cap.min(2_700_000));
    let mut wf = WaveformAccum::new(sr);

    loop {
        let packet = match format.next_packet() {
            Ok(Some(p)) => p,
            Ok(None) => break, // end of stream
            Err(symphonia::core::errors::Error::IoError(_))      => break,
            Err(symphonia::core::errors::Error::ResetRequired)   => break,
            Err(e) => return Err(e.into()),
        };
        if packet.track_id != track_id { continue; }

        let decoded = decoder.decode(&packet)?;
        push_samples_and_peaks(
            &mut pcm_batch,
            &mut total_frames,
            decoded,
            &mut wf,
            &mut mono_for_bpm,
            bpm_cap,
            deck,
            load_gen,
            sr,
            status_tx,
        );
    }

    wf.flush(&mut |_| {});
    if !wf.bands_batch.is_empty() {
        let _ = status_tx.send(AudioStatus::WaveformChunk {
            deck,
            bands: std::mem::take(&mut wf.bands_batch),
            frame_offset: wf.bucket_start_frame,
        });
    }
    if !pcm_batch.is_empty() {
        let _ = status_tx.send(AudioStatus::PcmChunk {
            deck, load_gen,
            frames: pcm_batch,
            sample_rate: sr,
        });
    }

    let _ = status_tx.send(AudioStatus::Ready {
        deck,
        load_gen,
        duration_samples: total_frames,
        sample_rate: sr,
    });

    // BPM detection on the mono accumulator. Runs synchronously in the decode
    // thread, so it doesn't compete with the realtime callback. ~hundreds of
    // ms for a 60 s clip; the Ready status above means the track is already
    // playable before this fires.
    if !mono_for_bpm.is_empty() {
        let grid = timestretch::detect_beat_grid(&mono_for_bpm, sr);
        if grid.bpm.is_finite() && grid.bpm > 0.0 && !grid.beats.is_empty() {
            // `beats` from timestretch are `usize` sample indices into the
            // 60 s analysis window — same sample-frame coordinate space the
            // engine uses for `read_pos`.
            let beats: Vec<u64> = grid.beats.iter().map(|&b| b as u64).collect();
            let _ = status_tx.send(AudioStatus::BeatGrid {
                deck,
                beats,
                bpm: grid.bpm as f32,
            });
        }
    }

    Ok(())
}

// ─── Per-decode waveform accumulator ──────────────────────────────────────────
//
// Splits the mono mix into bass / mid / high via biquads, then takes the peak
// per band over PEAK_BUCKET frames. Each `bands_batch` entry is one bucket
// `[bass, mid, high]`. Sent off every 64 buckets (~6 s) as a `WaveformChunk`.

struct WaveformAccum {
    bass_lp: biquad::DirectForm1<f32>,
    mid_bp:  biquad::DirectForm1<f32>,
    high_hp: biquad::DirectForm1<f32>,
    bucket_bass:  f32,
    bucket_mid:   f32,
    bucket_high:  f32,
    bucket_count: usize,
    bands_batch:  Vec<[f32; 3]>,
    frame_offset: u64,
    bucket_start_frame: u64,
}

impl WaveformAccum {
    fn new(sample_rate: u32) -> Self {
        use biquad::{Coefficients, DirectForm1, Hertz, Q_BUTTERWORTH_F32, Type};
        let fs = Hertz::<f32>::from_hz(sample_rate as f32).unwrap();
        let lp_co = Coefficients::<f32>::from_params(
            Type::LowPass, fs, Hertz::<f32>::from_hz(250.0).unwrap(), Q_BUTTERWORTH_F32
        ).unwrap();
        let bp_co = Coefficients::<f32>::from_params(
            Type::BandPass, fs, Hertz::<f32>::from_hz(1500.0).unwrap(), 0.7
        ).unwrap();
        let hp_co = Coefficients::<f32>::from_params(
            Type::HighPass, fs, Hertz::<f32>::from_hz(4000.0).unwrap(), Q_BUTTERWORTH_F32
        ).unwrap();
        Self {
            bass_lp: DirectForm1::<f32>::new(lp_co),
            mid_bp:  DirectForm1::<f32>::new(bp_co),
            high_hp: DirectForm1::<f32>::new(hp_co),
            bucket_bass:  0.0,
            bucket_mid:   0.0,
            bucket_high:  0.0,
            bucket_count: 0,
            bands_batch:  Vec::with_capacity(64),
            frame_offset: 0,
            bucket_start_frame: 0,
        }
    }

    #[inline]
    fn push_sample(&mut self, mono: f32) {
        use biquad::Biquad as _;
        let b = self.bass_lp.run(mono).abs();
        let m = self.mid_bp .run(mono).abs();
        let h = self.high_hp.run(mono).abs();
        if b > self.bucket_bass { self.bucket_bass = b; }
        if m > self.bucket_mid  { self.bucket_mid  = m; }
        if h > self.bucket_high { self.bucket_high = h; }
        self.bucket_count += 1;
        self.frame_offset += 1;

        if self.bucket_count >= PEAK_BUCKET {
            // Compress: sqrt expands quiet detail.
            self.bands_batch.push([
                self.bucket_bass.sqrt(),
                self.bucket_mid.sqrt(),
                self.bucket_high.sqrt(),
            ]);
            self.bucket_bass  = 0.0;
            self.bucket_mid   = 0.0;
            self.bucket_high  = 0.0;
            self.bucket_count = 0;
        }
    }

    /// Flush any in-progress bucket if non-empty.
    fn flush(&mut self, _emit: &mut dyn FnMut(&[[f32; 3]])) {
        if self.bucket_count > 0 {
            self.bands_batch.push([
                self.bucket_bass.sqrt(),
                self.bucket_mid.sqrt(),
                self.bucket_high.sqrt(),
            ]);
            self.bucket_bass = 0.0;
            self.bucket_mid  = 0.0;
            self.bucket_high = 0.0;
            self.bucket_count = 0;
        }
    }
}

fn extract_tags(format: &mut Box<dyn FormatReader>) -> (Option<String>, Option<String>) {
    let mut artist = None;
    let mut album  = None;

    if let Some(rev) = format.metadata().current() {
        for tag in &rev.media.tags {
            match &tag.std {
                Some(StandardTag::Artist(v))      => artist = Some(v.to_string()),
                Some(StandardTag::AlbumArtist(v)) if artist.is_none()
                                                  => artist = Some(v.to_string()),
                Some(StandardTag::Album(v))       => album  = Some(v.to_string()),
                _ => {}
            }
        }
    }

    (artist, album)
}

#[allow(clippy::too_many_arguments)]
fn push_samples_and_peaks(
    pcm_batch:          &mut Vec<f32>,
    total_frames:       &mut u64,
    buf:                GenericAudioBufferRef,
    wf:                 &mut WaveformAccum,
    mono_for_bpm:       &mut Vec<f32>,
    bpm_cap:            usize,
    deck:               DeckId,
    load_gen:           u64,
    sample_rate:        u32,
    status_tx:          &Sender<AudioStatus>,
) {
    // symphonia 0.6 converts any sample format to interleaved f32 for us.
    let ch = buf.num_planes().max(1);
    let n = buf.samples_interleaved();
    let mut inter = vec![0f32; n];
    buf.copy_to_slice_interleaved::<f32, _>(inter.as_mut_slice());

    let frames = n / ch;
    for f in 0..frames {
        let lf = inter[f * ch];
        let rf = if ch >= 2 { inter[f * ch + 1] } else { lf };
        pcm_batch.push(lf);
        pcm_batch.push(rf);
        *total_frames += 1;

        let mono = (lf + rf) * 0.5;
        wf.push_sample(mono);

        if mono_for_bpm.len() < bpm_cap {
            mono_for_bpm.push(mono);
        }

        if wf.bands_batch.len() >= 64 {
            let _ = status_tx.send(AudioStatus::WaveformChunk {
                deck,
                bands: std::mem::take(&mut wf.bands_batch),
                frame_offset: wf.bucket_start_frame,
            });
            let _ = status_tx.send(AudioStatus::PcmChunk {
                deck,
                load_gen,
                frames: std::mem::take(pcm_batch),
                sample_rate,
            });
            wf.bucket_start_frame = wf.frame_offset;
        }
    }
}

// ─── PipeWire format pod ──────────────────────────────────────────────────────

fn make_f32_stereo_pod() -> Result<Vec<u8>> {
    use pipewire::spa::param::ParamType;
    use pipewire::spa::param::audio::AudioFormat;
    use pipewire::spa::param::format::{FormatProperties, MediaSubtype, MediaType};
    use pipewire::spa::pod::serialize::PodSerializer;
    use pipewire::spa::pod::{Value, object, property};
    use pipewire::spa::utils::SpaTypes;

    let bytes = PodSerializer::serialize(
        std::io::Cursor::new(Vec::new()),
        &Value::Object(object!(
            SpaTypes::ObjectParamFormat,
            ParamType::EnumFormat,
            property!(FormatProperties::MediaType,    Id, MediaType::Audio),
            property!(FormatProperties::MediaSubtype, Id, MediaSubtype::Raw),
            property!(FormatProperties::AudioFormat,  Id, AudioFormat::F32LE),
            property!(FormatProperties::AudioRate,    Int, STREAM_SR as i32),
            property!(FormatProperties::AudioChannels, Int, 2i32),
        )),
    )?.0.into_inner();

    Ok(bytes)
}