ringo_core/baresip/

ausrc.rs

//! Custom `ringo` baresip audio-source module.
//!
//! Why this exists: baresip's public `audio_set_source()` is *transient* — a
//! media re-negotiation (re-INVITE: hold/resume, transfer, line switch) rebuilds
//! the audio stream and resets the source to the account default. And the only
//! persistent knob (`account.ausrc_mod` / `config.audio.src_mod`) is either a
//! private struct field with no public setter, or process-global (all UAs in
//! this single process would collide — the old backend got away with a global
//! because each UA was its own process).
//!
//! So instead of fighting baresip's source management, ringo registers its OWN
//! source module via `ausrc_register()` (baresip's official extension point) and
//! points each UA's account at `audio_source=ringo,<key>`. baresip then re-allocs
//! *our* source on every stream rebuild, and we render whatever the per-key
//! [`REGISTRY`] currently says — race-free, persistent across re-INVITEs, and
//! per-UA isolated. Changing a UA's audio (tone/file/silence) is just a registry
//! update; the running render thread picks it up on the next frame.

use std::collections::{HashMap, VecDeque};
use std::os::raw::{c_char, c_void};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex, OnceLock};
use std::time::{Duration, Instant};

use super::bindings::*;

const AMPLITUDE: f64 = 0.25;

/// What a UA's source should emit. Cheap to clone (file PCM is shared via `Arc`).
#[derive(Clone)]
enum GenSpec {
    Silence,
    /// Sine wave at the given frequency in Hz.
    Tone(u32),
    /// Mono S16 samples at the given sample rate, looped.
    File(Arc<Vec<i16>>, u32),
    /// Live-streamed mono S16 at the given sample rate: each frame is pulled from
    /// the UA's [`STREAM_IN`] queue (fed via [`push_audio`]); underrun → silence.
    Stream(u32),
}

/// A registry entry: the desired spec plus a version that bumps on every change,
/// so a render thread can cheaply detect "my spec changed, reset my phase".
struct Entry {
    version: u64,
    spec: GenSpec,
}

/// Per-UA desired source, keyed by the device string (the `<key>` in
/// `audio_source=ringo,<key>`, which ringo sets to the account username).
static REGISTRY: OnceLock<Mutex<HashMap<String, Entry>>> = OnceLock::new();

fn registry() -> &'static Mutex<HashMap<String, Entry>> {
    REGISTRY.get_or_init(|| Mutex::new(HashMap::new()))
}

/// Registered `struct ausrc *` (kept alive for the process lifetime).
static AUSRC: OnceLock<usize> = OnceLock::new();

/// A mono PCM (s16) audio frame plus its sample rate — the unit fed into a call
/// via [`push_audio`] and handed out of one via [`subscribe_received_audio`].
#[derive(Debug, Clone)]
pub struct AudioFrame {
    /// Mono 16-bit PCM samples.
    pub samples: Vec<i16>,
    /// Sample rate in Hz.
    pub rate: u32,
}

/// A single UA's streamed-in PCM queue. Each UA gets its OWN lock so the
/// real-time render thread only ever contends with that UA's own [`push_audio`]
/// — never with other UAs' render/feed (which a single global lock would force).
type StreamQueue = Arc<Mutex<VecDeque<i16>>>;

/// Per-UA stream-in queues, keyed by username. The outer lock is held only for
/// the O(1) lookup/clone of the per-UA [`StreamQueue`]; all rendering/feeding
/// happens under the inner per-UA lock, off the global path.
static STREAM_IN: OnceLock<Mutex<HashMap<String, StreamQueue>>> = OnceLock::new();

fn stream_in() -> &'static Mutex<HashMap<String, StreamQueue>> {
    STREAM_IN.get_or_init(|| Mutex::new(HashMap::new()))
}

/// The per-UA stream-in queue for `key`, if streaming was started for it.
fn stream_queue(key: &str) -> Option<StreamQueue> {
    stream_in()
        .lock()
        .unwrap_or_else(|e| e.into_inner())
        .get(key)
        .map(Arc::clone)
}

/// Per-UA live subscriber for received (decoded) audio frames. One per key; a
/// new subscription replaces the old.
static RX_TAPS: OnceLock<Mutex<HashMap<String, std::sync::mpsc::Sender<AudioFrame>>>> =
    OnceLock::new();

fn rx_taps() -> &'static Mutex<HashMap<String, std::sync::mpsc::Sender<AudioFrame>>> {
    RX_TAPS.get_or_init(|| Mutex::new(HashMap::new()))
}

/// Cap on buffered streamed-in audio (samples ≈ 1 s at 48 kHz). Past this the
/// oldest samples are dropped, bounding added latency / memory under overrun.
const STREAM_IN_CAP_SAMPLES: usize = 48_000;

/// Translate a baresip-style source spec into a [`GenSpec`] and store it for
/// `key`, bumping the version so the render thread reloads it. Accepts the same
/// specs the engine already produces: `ausine,<freq>`, `aufile,<path>`, and
/// anything else (e.g. `aubridge,...`) → silence.
pub(super) fn set_generator(key: &str, spec: &str) {
    let parsed = parse_spec(spec);
    let mut map = registry().lock().unwrap_or_else(|e| e.into_inner());
    let version = map.get(key).map(|e| e.version + 1).unwrap_or(0);
    map.insert(
        key.to_string(),
        Entry {
            version,
            spec: parsed,
        },
    );
}

/// Switch the UA's audio source to live-streamed mono s16 PCM at `rate` Hz, fed
/// via [`push_audio`]. Until samples arrive (or on underrun) the source renders
/// silence; queued audio is emitted in real time by the render thread.
pub(super) fn start_audio_stream(key: &str, rate: u32) {
    stream_in()
        .lock()
        .unwrap_or_else(|e| e.into_inner())
        .entry(key.to_string())
        .or_default();
    let mut map = registry().lock().unwrap_or_else(|e| e.into_inner());
    let version = map.get(key).map(|e| e.version + 1).unwrap_or(0);
    map.insert(
        key.to_string(),
        Entry {
            version,
            spec: GenSpec::Stream(rate.max(1)),
        },
    );
}

/// Append mono s16 `samples` (at the rate given to [`start_audio_stream`]) to the
/// UA's stream-in queue. No-op if the queue was never started.
pub(super) fn push_audio(key: &str, samples: &[i16]) {
    let Some(q) = stream_queue(key) else {
        return;
    };
    let mut q = q.lock().unwrap_or_else(|e| e.into_inner());
    q.extend(samples.iter().copied());
    while q.len() > STREAM_IN_CAP_SAMPLES {
        q.pop_front();
    }
}

/// Subscribe to the UA's received (decoded) audio: returns a receiver of mono
/// s16 [`AudioFrame`]s as they arrive. Replaces any previous subscription.
pub(super) fn subscribe_received_audio(key: &str) -> std::sync::mpsc::Receiver<AudioFrame> {
    let (tx, rx) = std::sync::mpsc::channel();
    rx_taps()
        .lock()
        .unwrap_or_else(|e| e.into_inner())
        .insert(key.to_string(), tx);
    rx
}

/// Pre-seed a UA's source with silence (called at session setup) so the source
/// has a defined default before the first `set_generator`.
pub(super) fn init_generator(key: &str) {
    let mut map = registry().lock().unwrap_or_else(|e| e.into_inner());
    map.entry(key.to_string()).or_insert(Entry {
        version: 0,
        spec: GenSpec::Silence,
    });
}

/// Drop a UA's audio state (source generator + received-audio buffer), called
/// when the session is dropped.
pub(super) fn remove_generator(key: &str) {
    registry()
        .lock()
        .unwrap_or_else(|e| e.into_inner())
        .remove(key);
    rx_buffers()
        .lock()
        .unwrap_or_else(|e| e.into_inner())
        .remove(key);
    tx_buffers()
        .lock()
        .unwrap_or_else(|e| e.into_inner())
        .remove(key);
    stream_in()
        .lock()
        .unwrap_or_else(|e| e.into_inner())
        .remove(key);
    rx_taps()
        .lock()
        .unwrap_or_else(|e| e.into_inner())
        .remove(key);
}

fn spec_name(spec: &GenSpec) -> String {
    match spec {
        GenSpec::Silence => "silence".into(),
        GenSpec::Tone(f) => format!("tone({f}Hz)"),
        GenSpec::File(s, sr) => format!("file({} samples @{sr}Hz)", s.len()),
        GenSpec::Stream(sr) => format!("stream(@{sr}Hz)"),
    }
}

fn parse_spec(spec: &str) -> GenSpec {
    let (driver, device) = spec.split_once(',').unwrap_or((spec, ""));
    match driver {
        "ausine" => match device.split(',').next().unwrap_or("").parse::<u32>() {
            Ok(f) if (10..=20000).contains(&f) => GenSpec::Tone(f),
            _ => {
                crate::rlog!(
                    Warn,
                    "ringo ausrc: invalid tone freq in '{spec}', using silence"
                );
                GenSpec::Silence
            }
        },
        "aufile" => match load_wav_mono(device) {
            Some((samples, srate)) => GenSpec::File(Arc::new(samples), srate),
            None => {
                crate::rlog!(
                    Warn,
                    "ringo ausrc: failed to load '{device}', using silence"
                );
                GenSpec::Silence
            }
        },
        _ => GenSpec::Silence,
    }
}

/// Load a WAV file as mono S16 samples + its sample rate. Stereo is downmixed.
fn load_wav_mono(path: &str) -> Option<(Vec<i16>, u32)> {
    let data = std::fs::read(path).ok()?;
    let pcm = super::sounds::parse_wav(&data)?;
    // `samples` is S16LE bytes. Convert to i16, downmixing stereo to mono.
    let i16s: Vec<i16> = pcm
        .samples
        .chunks_exact(2)
        .map(|b| i16::from_le_bytes([b[0], b[1]]))
        .collect();
    let mono = if pcm.channels >= 2 {
        i16s.chunks_exact(pcm.channels as usize)
            .map(|frame| {
                let sum: i32 = frame.iter().take(2).map(|&s| s as i32).sum();
                (sum / 2) as i16
            })
            .collect()
    } else {
        i16s
    };
    if mono.is_empty() {
        return None;
    }
    Some((mono, pcm.srate))
}

// ─── baresip ausrc module ──────────────────────────────────────────────────

/// Rust-owned state for one active source, referenced from the mem_zalloc'd
/// `ausrc_st` cell. Its `Drop` stops and joins the render thread.
struct SrcState {
    key: String,
    run: Arc<AtomicBool>,
    thread: Option<std::thread::JoinHandle<()>>,
}

impl Drop for SrcState {
    fn drop(&mut self) {
        crate::rlog!(Info, "ringo ausrc: FREE key={}", self.key);
        self.run.store(false, Ordering::Release);
        if let Some(t) = self.thread.take() {
            let _ = t.join();
        }
    }
}

/// Pointers handed to the render thread. `rh`/`arg` come from baresip and are
/// only ever used to call back into baresip from this one thread (the same
/// pattern baresip's own `ausine` uses).
struct ReadCb {
    rh: ausrc_read_h,
    arg: usize,
}
// SAFETY: `arg` is a baresip pointer used solely to call `rh` from the render
// thread; baresip's audio path is built to have the ausrc read handler invoked
// from the source's own thread. No use-after-free: baresip frees the source via
// `stop_tx()` (src/audio.c), which mem_derefs `tx->ausrc` (→ our destructor →
// thread join) BEFORE freeing `tx->aubuf`/`tx->mtx` (the `arg`) — "audio source
// must be stopped first". So `rh` can never run against a freed `arg`.
unsafe impl Send for ReadCb {}

/// Called by baresip to free the `ausrc_st` (on stream teardown / re-INVITE).
unsafe extern "C" fn destructor(arg: *mut c_void) {
    // `arg` points at the mem_zalloc'd cell holding our `*mut SrcState`.
    let cell = arg as *mut *mut SrcState;
    let state = unsafe { *cell };
    if !state.is_null() {
        // Reclaim and drop → SrcState::Drop stops + joins the render thread.
        drop(unsafe { Box::from_raw(state) });
        unsafe { *cell = std::ptr::null_mut() };
    }
}

unsafe extern "C" fn alloc_handler(
    stp: *mut *mut ausrc_st,
    _as: *const ausrc,
    prm: *mut ausrc_prm,
    dev: *const c_char,
    rh: ausrc_read_h,
    _errh: ausrc_error_h,
    arg: *mut c_void,
) -> i32 {
    const EINVAL: i32 = 22;
    const ENOMEM: i32 = 12;
    const ENOTSUP: i32 = 95;

    if stp.is_null() || prm.is_null() || dev.is_null() || rh.is_none() {
        return EINVAL;
    }

    let prm = unsafe { &*prm };
    let key = unsafe { std::ffi::CStr::from_ptr(dev) }
        .to_string_lossy()
        .into_owned();

    let srate = prm.srate;
    let ptime = if prm.ptime == 0 { 20 } else { prm.ptime };
    let fmt = prm.fmt;

    // Reject degenerate params rather than masking them (e.g. `ch.max(1)`): a
    // channel/rate that doesn't match what the TX path negotiated would
    // mis-interleave or divide-by-zero.
    if srate == 0 || prm.ch == 0 {
        crate::rlog!(Warn, "ringo ausrc: invalid prm srate={srate} ch={}", prm.ch);
        return EINVAL;
    }
    let ch = prm.ch;

    // The audio TX drops any frame whose fmt != tx->src_fmt, so we must render
    // in exactly the requested format. Only S16LE and FLOAT are implemented
    // (matching ausine); telephony configs use S16LE.
    if fmt != aufmt::AUFMT_S16LE as i32 && fmt != aufmt::AUFMT_FLOAT as i32 {
        crate::rlog!(Warn, "ringo ausrc: unsupported sample format {fmt}");
        return ENOTSUP;
    }

    // Allocate the cell baresip will mem_deref (running our destructor).
    let cell = unsafe { mem_zalloc(std::mem::size_of::<*mut SrcState>(), Some(destructor)) };
    if cell.is_null() {
        return ENOMEM;
    }

    crate::rlog!(
        Info,
        "ringo ausrc: ALLOC key={key} srate={srate} ch={ch} ptime={ptime} fmt={fmt}"
    );

    let run = Arc::new(AtomicBool::new(true));
    let run_thread = run.clone();
    let cb = ReadCb {
        rh,
        arg: arg as usize,
    };

    // Arm the sent-audio buffer for this stream (keeps samples when full
    // capture is on — see alloc_buffer).
    alloc_buffer(tx_buffers(), &key, srate);

    let key_thread = key.clone();
    let thread = match std::thread::Builder::new()
        .name("ringo-ausrc".into())
        .spawn(move || render_loop(key_thread, srate, ch, ptime, fmt, run_thread, cb))
    {
        Ok(t) => t,
        Err(e) => {
            // Don't hand baresip a source that never produces frames; free the
            // cell (destructor no-ops on the still-null state) and fail.
            crate::rlog!(Error, "ringo ausrc: spawn render thread failed: {e}");
            unsafe { mem_deref(cell) };
            return ENOMEM;
        }
    };

    let state = Box::new(SrcState {
        key,
        run,
        thread: Some(thread),
    });
    unsafe {
        *(cell as *mut *mut SrcState) = Box::into_raw(state);
        *stp = cell as *mut ausrc_st;
    }
    0
}

/// Fill `mono` from the front of the stream-in queue `q`, nearest-sample
/// resampling input→output by `step` (= in_rate / out_rate). The input read is
/// consumed (dropped from the queue front); `pos` carries the fractional read
/// position across calls. Underrun (empty/short queue) renders silence, and an
/// emptied queue resets `pos` so newly-fed audio starts cleanly.
fn render_stream(q: Option<&mut VecDeque<i16>>, pos: &mut f64, step: f64, mono: &mut [i16]) {
    let Some(q) = q else {
        mono.iter_mut().for_each(|s| *s = 0);
        return;
    };
    for s in mono.iter_mut() {
        *s = q.get(*pos as usize).copied().unwrap_or(0);
        *pos += step;
    }
    // Consume the read input by popping from the front (O(consumed), no memmove
    // of the tail — this runs under the per-UA lock on the real-time thread).
    let consumed = (*pos as usize).min(q.len());
    for _ in 0..consumed {
        q.pop_front();
    }
    *pos -= consumed as f64;
    if q.is_empty() {
        *pos = 0.0; // underran: restart cleanly when audio resumes
    }
}

/// The render thread: every `ptime` ms, generate one frame from the current
/// registry spec for `key` and hand it to baresip via `rh`.
fn render_loop(
    key: String,
    srate: u32,
    ch: u8,
    ptime: u32,
    fmt: i32,
    run: Arc<AtomicBool>,
    cb: ReadCb,
) {
    let is_float = fmt == aufmt::AUFMT_FLOAT as i32;
    let sample_size = if is_float { 4 } else { 2 };
    let frames = (srate as usize * ptime as usize / 1000).max(1);
    let sampc = frames * ch as usize;

    let mut sampv = vec![0u8; sampc * sample_size];
    let mut mono = vec![0i16; frames];

    // Render state (reset when the spec version changes).
    let mut cur_version = u64::MAX;
    let mut cur_spec = GenSpec::Silence;
    let mut phase = 0.0f64; // tone phase accumulator
    let mut file_pos = 0.0f64; // fractional read position into the file
    let mut stream_pos = 0.0f64; // fractional read position into the stream-in queue

    let mut start = Instant::now();
    let mut frame_idx: u64 = 0;

    while run.load(Ordering::Acquire) {
        // Pace to real time: wake at start + frame_idx * ptime.
        let target = start + Duration::from_millis(frame_idx * ptime as u64);
        let now = Instant::now();
        if target > now {
            std::thread::sleep(target - now);
        } else if now - target > Duration::from_millis(ptime as u64 * 4) {
            // Fell far behind (suspend / scheduling stall): rebase the clock so
            // we don't fire a long burst of catch-up frames (which the TX aubuf
            // would just drop as overruns anyway).
            start = now;
            frame_idx = 0;
        }
        if !run.load(Ordering::Acquire) {
            break;
        }

        // Reload spec if it changed.
        let present = {
            let map = registry().lock().unwrap_or_else(|e| e.into_inner());
            map.get(&key).map(|e| (e.version, e.spec.clone()))
        };
        match &present {
            Some((version, spec)) => {
                if *version != cur_version {
                    cur_version = *version;
                    cur_spec = spec.clone();
                    phase = 0.0;
                    file_pos = 0.0;
                    stream_pos = 0.0;
                    crate::rlog!(
                        Info,
                        "ringo ausrc: key={key} spec={} (v{version})",
                        spec_name(spec)
                    );
                }
            }
            None => {
                // Registry entry gone (shouldn't happen while a source is live):
                // render silence rather than a stale spec.
                if cur_version != u64::MAX {
                    crate::rlog!(Warn, "ringo ausrc: key={key} no registry entry, silence");
                    cur_version = u64::MAX;
                    cur_spec = GenSpec::Silence;
                }
            }
        }

        // Render `frames` mono samples.
        match &cur_spec {
            GenSpec::Silence => mono.iter_mut().for_each(|s| *s = 0),
            GenSpec::Tone(freq) => {
                let step = std::f64::consts::TAU * (*freq as f64) / (srate as f64);
                for s in mono.iter_mut() {
                    *s = (phase.sin() * AMPLITUDE * 32767.0) as i16;
                    phase += step;
                    if phase >= std::f64::consts::TAU {
                        phase -= std::f64::consts::TAU;
                    }
                }
            }
            GenSpec::File(samples, file_srate) => {
                let step = *file_srate as f64 / srate as f64;
                let len = samples.len() as f64;
                for s in mono.iter_mut() {
                    // `while`, not `if`: a single step may overshoot `len` for a
                    // tiny file (len < step), so wrap fully to keep looping.
                    while file_pos >= len {
                        file_pos -= len;
                    }
                    let i = file_pos as usize;
                    *s = samples.get(i).copied().unwrap_or(0);
                    file_pos += step;
                }
            }
            GenSpec::Stream(stream_srate) => {
                let step = (*stream_srate as f64 / srate as f64).max(f64::MIN_POSITIVE);
                // Resolve the per-UA queue (brief global lock for the O(1) clone),
                // then render under that UA's own lock — no cross-UA contention.
                match stream_queue(&key) {
                    Some(q) => {
                        let mut q = q.lock().unwrap_or_else(|e| e.into_inner());
                        render_stream(Some(&mut q), &mut stream_pos, step, &mut mono);
                    }
                    None => render_stream(None, &mut stream_pos, step, &mut mono),
                }
            }
        }

        // Capture the sent audio for --save-audio (full capture only).
        if FULL_CAPTURE.load(Ordering::Acquire) {
            capture_mono(tx_buffers(), &key, &mono);
        }

        // Pack mono → interleaved sampv in the requested format.
        if is_float {
            let out =
                unsafe { std::slice::from_raw_parts_mut(sampv.as_mut_ptr() as *mut f32, sampc) };
            for (f, &m) in out.chunks_mut(ch as usize).zip(mono.iter()) {
                let v = m as f32 / 32768.0;
                f.iter_mut().for_each(|x| *x = v);
            }
        } else {
            let out =
                unsafe { std::slice::from_raw_parts_mut(sampv.as_mut_ptr() as *mut i16, sampc) };
            for (f, &m) in out.chunks_mut(ch as usize).zip(mono.iter()) {
                f.iter_mut().for_each(|x| *x = m);
            }
        }

        // Build the auframe and push it to baresip.
        let mut af: auframe = unsafe { std::mem::zeroed() };
        unsafe {
            auframe_init(
                &mut af,
                if is_float {
                    aufmt::AUFMT_FLOAT
                } else {
                    aufmt::AUFMT_S16LE
                },
                sampv.as_mut_ptr() as *mut c_void,
                sampc,
                srate,
                ch,
            );
        }
        af.timestamp = frame_idx * ptime as u64 * 1000; // AUDIO_TIMEBASE = microseconds

        if let Some(rh) = cb.rh {
            unsafe { rh(&mut af, cb.arg as *mut c_void) };
        }

        frame_idx += 1;
    }
}

// ─── ringo auplay (self-clocked RX sink) ───────────────────────────────────
//
// Why we also need a player: the headless RX path (decode → aubuf → player) is
// *pull-driven* — something must call the player's write handler at real-time
// pace, or the decoder never advances and the received audio is silent.
// aubridge's player is clocked by its device thread, which only runs when BOTH
// an aubridge source AND player share a device (modules/aubridge/device.c) — but
// ringo replaced the source, so that thread would never start. So ringo provides
// its own self-clocked player: a timer thread that pulls `wh()` every ptime,
// which both clocks the decode pipeline and yields the received frames we
// capture for verify-audio / --save-audio.

/// Registered `struct auplay *` (kept alive for the process lifetime).
static AUPLAY: OnceLock<usize> = OnceLock::new();

struct PlayState {
    run: Arc<AtomicBool>,
    thread: Option<std::thread::JoinHandle<()>>,
}

impl Drop for PlayState {
    fn drop(&mut self) {
        self.run.store(false, Ordering::Release);
        if let Some(t) = self.thread.take() {
            let _ = t.join();
        }
    }
}

struct WriteCb {
    wh: auplay_write_h,
    arg: usize,
}
// SAFETY: same contract as ReadCb — `arg`/`wh` are baresip pointers used only to
// call back into baresip from this player's own thread. baresip frees the player
// via `aurecv_stop()` (src/aureceiver.c), which mem_derefs `ar->auplay` (→ our
// destructor → thread join) before freeing the receiver `arg`, so `wh` never
// runs against a freed `arg`.
unsafe impl Send for WriteCb {}

/// Per-UA mono audio buffer (one for received, one for sent). Lets ringo-flow
/// verify a received tone in-process (Goertzel on these samples) and save
/// recordings — instead of baresip's sndfile WAV dumps. No shared-dir race, no
/// disk round-trip, per-UA isolated by construction.
struct AudioBuf {
    srate: u32,
    samples: VecDeque<i16>,
    /// Holds a finished recording (capture was disarmed): no more samples are
    /// appended, so the 3 s verify cap can't trim it before it is saved.
    /// Cleared by the next arm or stream (re)alloc.
    frozen: bool,
}

/// Seconds of audio retained per UA for verification (the verify window tail).
const VERIFY_RETAIN_SECS: usize = 3;
/// Cap on a full-call recording: only the newest [`RECORDING_MAX_SECS`] seconds
/// are retained per UA (older audio is dropped; a save flags it truncated).
/// Public via `ringo_core::RECORDING_MAX_SECS` so callers can flag truncation
/// against the same cap.
pub const RECORDING_MAX_SECS: u64 = 600;
/// Seconds retained when full capture is on (`--save-audio`); bounds a runaway
/// call. Test calls are far shorter.
const FULL_RETAIN_SECS: usize = RECORDING_MAX_SECS as usize;

/// Whether to retain the whole call (for `--save-audio`) vs. just the verify
/// window. Set once per process from `BackendOptions.record_audio`.
static FULL_CAPTURE: AtomicBool = AtomicBool::new(false);

static RX_BUFFERS: OnceLock<Mutex<HashMap<String, AudioBuf>>> = OnceLock::new();
static TX_BUFFERS: OnceLock<Mutex<HashMap<String, AudioBuf>>> = OnceLock::new();

fn rx_buffers() -> &'static Mutex<HashMap<String, AudioBuf>> {
    RX_BUFFERS.get_or_init(|| Mutex::new(HashMap::new()))
}
fn tx_buffers() -> &'static Mutex<HashMap<String, AudioBuf>> {
    TX_BUFFERS.get_or_init(|| Mutex::new(HashMap::new()))
}

/// Enable/disable full-call capture (for `--save-audio`). When off, only the
/// rolling verify window is retained and the sent buffer isn't captured.
pub(super) fn set_full_capture(on: bool) {
    FULL_CAPTURE.store(on, Ordering::Release);
}

/// Arm or disarm full-call capture for `key`. Arming starts a fresh recording
/// (the UA's buffered samples are cleared, the srate stays); disarming returns
/// to the 3 s verify window (sent side no longer captured) but KEEPS the
/// buffers until the next arm/alloc, so a save after hangup still works.
pub(super) fn arm_capture(key: &str, full: bool) {
    let was_full = FULL_CAPTURE.swap(full, Ordering::AcqRel);
    for buffers in [tx_buffers(), rx_buffers()] {
        if let Some(buf) = buffers
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .get_mut(key)
        {
            if full {
                buf.samples.clear();
                buf.frozen = false;
            } else if was_full {
                buf.frozen = true;
            }
        }
    }
}

fn retain_secs() -> usize {
    if FULL_CAPTURE.load(Ordering::Acquire) {
        FULL_RETAIN_SECS
    } else {
        VERIFY_RETAIN_SECS
    }
}

/// Reset a UA's buffer in `buffers` to empty at `srate` (called when a stream
/// (re)alloc must start fresh — see [`alloc_buffer`]).
fn reset_buffer(buffers: &Mutex<HashMap<String, AudioBuf>>, key: &str, srate: u32) {
    buffers.lock().unwrap_or_else(|e| e.into_inner()).insert(
        key.to_string(),
        AudioBuf {
            srate,
            samples: VecDeque::new(),
            frozen: false,
        },
    );
}

/// Prepare a UA's buffer for a (re)allocated stream: create it at the negotiated
/// `srate` if absent, reset it if the rate changed (mixing rates would corrupt
/// the WAV), otherwise keep the samples.
fn prepare_buffer(buffers: &Mutex<HashMap<String, AudioBuf>>, key: &str, srate: u32) {
    let mut map = buffers.lock().unwrap_or_else(|e| e.into_inner());
    match map.get_mut(key) {
        None => {
            map.insert(
                key.to_string(),
                AudioBuf {
                    srate,
                    samples: VecDeque::new(),
                    frozen: false,
                },
            );
        }
        Some(buf) if buf.srate != srate => {
            crate::rlog!(
                Info,
                "ringo ausrc: key={key} srate {} -> {srate}, resetting buffer",
                buf.srate
            );
            buf.srate = srate;
            buf.samples.clear();
            buf.frozen = false;
        }
        _ => {}
    }
}

/// Buffer policy on stream (re)alloc: with full capture on, samples survive
/// (a re-INVITE hold/resume must not wipe a running recording); with it off,
/// reset — the verify window reflects only the current stream, and a disarmed
/// recording is bounded by the next call.
fn alloc_buffer(buffers: &Mutex<HashMap<String, AudioBuf>>, key: &str, srate: u32) {
    if FULL_CAPTURE.load(Ordering::Acquire) {
        prepare_buffer(buffers, key, srate);
    } else {
        reset_buffer(buffers, key, srate);
    }
}

/// The retained received mono samples for `key` and their sample rate.
pub(super) fn received_window(key: &str) -> Option<(Vec<i16>, u32)> {
    buffer_window(rx_buffers(), key)
}

/// The retained sent mono samples for `key` and their sample rate (full capture
/// only; empty otherwise).
pub(super) fn sent_window(key: &str) -> Option<(Vec<i16>, u32)> {
    buffer_window(tx_buffers(), key)
}

fn buffer_window(buffers: &Mutex<HashMap<String, AudioBuf>>, key: &str) -> Option<(Vec<i16>, u32)> {
    let map = buffers.lock().unwrap_or_else(|e| e.into_inner());
    let buf = map.get(key)?;
    Some((buf.samples.iter().copied().collect(), buf.srate))
}

/// Append already-mono `samples` to the UA's buffer in `buffers`, capped.
fn capture_mono(buffers: &Mutex<HashMap<String, AudioBuf>>, key: &str, samples: &[i16]) {
    let mut map = buffers.lock().unwrap_or_else(|e| e.into_inner());
    let Some(buf) = map.get_mut(key) else {
        return;
    };
    if buf.frozen {
        return;
    }
    buf.samples.extend(samples.iter().copied());
    let cap = (buf.srate as usize * retain_secs()).max(1);
    while buf.samples.len() > cap {
        buf.samples.pop_front();
    }
}

/// Append the channel-0 mono samples from an interleaved player frame to the
/// UA's received buffer. `wh` always fills the full `sampc` (aurecv_read pads an
/// underrun with silence), so the whole `sampv` is valid decoded audio.
fn capture_rx(key: &str, sampv: &[u8], is_float: bool, ch: usize, srate: u32) {
    let ch = ch.max(1);
    let mono: Vec<i16> = if is_float {
        let f =
            unsafe { std::slice::from_raw_parts(sampv.as_ptr() as *const f32, sampv.len() / 4) };
        f.chunks_exact(ch)
            .map(|fr| (fr[0] * 32767.0) as i16)
            .collect()
    } else {
        let s =
            unsafe { std::slice::from_raw_parts(sampv.as_ptr() as *const i16, sampv.len() / 2) };
        s.chunks_exact(ch).map(|fr| fr[0]).collect()
    };
    capture_mono(rx_buffers(), key, &mono);
    // Live tap: hand the frame to a subscriber (e.g. streaming RX out for STT).
    // Drop the subscription if the receiver is gone.
    let mut taps = rx_taps().lock().unwrap_or_else(|e| e.into_inner());
    if let Some(tx) = taps.get(key) {
        let frame = AudioFrame {
            samples: mono,
            rate: srate,
        };
        if tx.send(frame).is_err() {
            taps.remove(key);
        }
    }
}

unsafe extern "C" fn play_destructor(arg: *mut c_void) {
    let cell = arg as *mut *mut PlayState;
    let state = unsafe { *cell };
    if !state.is_null() {
        drop(unsafe { Box::from_raw(state) });
        unsafe { *cell = std::ptr::null_mut() };
    }
}

unsafe extern "C" fn play_alloc_handler(
    stp: *mut *mut auplay_st,
    _ap: *const auplay,
    prm: *mut auplay_prm,
    dev: *const c_char,
    wh: auplay_write_h,
    arg: *mut c_void,
) -> i32 {
    const EINVAL: i32 = 22;
    const ENOMEM: i32 = 12;

    if stp.is_null() || prm.is_null() || wh.is_none() {
        return EINVAL;
    }

    let prm = unsafe { &*prm };
    let srate = prm.srate;
    let ptime = if prm.ptime == 0 { 20 } else { prm.ptime };
    let fmt = prm.fmt;

    if srate == 0 || prm.ch == 0 {
        crate::rlog!(
            Warn,
            "ringo auplay: invalid prm srate={srate} ch={}",
            prm.ch
        );
        return EINVAL;
    }
    let ch = prm.ch;

    // The device key (= account username) identifies this UA's RX buffer.
    // Arm it for this stream (reset while full capture is off, keep samples
    // while it is on — see alloc_buffer).
    let key = if dev.is_null() {
        String::new()
    } else {
        unsafe { std::ffi::CStr::from_ptr(dev) }
            .to_string_lossy()
            .into_owned()
    };
    if !key.is_empty() {
        alloc_buffer(rx_buffers(), &key, srate);
    }

    let cell = unsafe { mem_zalloc(std::mem::size_of::<*mut PlayState>(), Some(play_destructor)) };
    if cell.is_null() {
        return ENOMEM;
    }

    let run = Arc::new(AtomicBool::new(true));
    let run_thread = run.clone();
    let cb = WriteCb {
        wh,
        arg: arg as usize,
    };

    let thread = match std::thread::Builder::new()
        .name("ringo-auplay".into())
        .spawn(move || play_loop(key, srate, ch, ptime, fmt, run_thread, cb))
    {
        Ok(t) => t,
        Err(e) => {
            // A player that never pulls would stall the decode pipeline; free the
            // cell (destructor no-ops on the null state) and fail.
            crate::rlog!(Error, "ringo auplay: spawn play thread failed: {e}");
            unsafe { mem_deref(cell) };
            return ENOMEM;
        }
    };

    let state = Box::new(PlayState {
        run,
        thread: Some(thread),
    });
    unsafe {
        *(cell as *mut *mut PlayState) = Box::into_raw(state);
        *stp = cell as *mut auplay_st;
    }
    0
}

/// Self-clocked player: every `ptime` ms, pull one frame from baresip via `wh`
/// (which drives the decode pipeline) and capture it into the UA's RX buffer for
/// in-process tone verification.
fn play_loop(
    key: String,
    srate: u32,
    ch: u8,
    ptime: u32,
    fmt: i32,
    run: Arc<AtomicBool>,
    cb: WriteCb,
) {
    let is_float = fmt == aufmt::AUFMT_FLOAT as i32;
    let sample_size = if is_float { 4 } else { 2 };
    let frames = (srate as usize * ptime as usize / 1000).max(1);
    let sampc = frames * ch as usize;
    let mut sampv = vec![0u8; sampc * sample_size];

    let af_fmt = if is_float {
        aufmt::AUFMT_FLOAT
    } else {
        aufmt::AUFMT_S16LE
    };

    let mut start = Instant::now();
    let mut frame_idx: u64 = 0;

    while run.load(Ordering::Acquire) {
        let target = start + Duration::from_millis(frame_idx * ptime as u64);
        let now = Instant::now();
        if target > now {
            std::thread::sleep(target - now);
        } else if now - target > Duration::from_millis(ptime as u64 * 4) {
            start = now;
            frame_idx = 0;
        }
        if !run.load(Ordering::Acquire) {
            break;
        }

        let mut af: auframe = unsafe { std::mem::zeroed() };
        unsafe {
            auframe_init(
                &mut af,
                af_fmt,
                sampv.as_mut_ptr() as *mut c_void,
                sampc,
                srate,
                ch,
            );
        }
        af.timestamp = frame_idx * ptime as u64 * 1000;

        if let Some(wh) = cb.wh {
            // wh fills sampv with the decoded received audio (aurecv_read →
            // aubuf_read_auframe), then we capture it for in-process verify.
            unsafe { wh(&mut af, cb.arg as *mut c_void) };
            if !key.is_empty() {
                capture_rx(&key, &sampv, is_float, ch as usize, srate);
            }
        }

        frame_idx += 1;
    }
}

/// Register the `ringo` audio source + player modules with baresip. Must run
/// once after `baresip_init`/`ua_init` (so `baresip_ausrcl()`/`baresip_auplayl()`
/// exist) on the RE thread. Returns `Err` if registration fails — the caller
/// must abort, since the agents would otherwise start with no working audio.
/// The registered `ausrc`/`auplay` are intentionally never `mem_deref`'d (kept
/// for the process lifetime); we stash the pointers for documentation.
pub(super) fn register_module() -> Result<(), String> {
    let mut asp: *mut ausrc = std::ptr::null_mut();
    let rc = unsafe {
        ausrc_register(
            &mut asp,
            baresip_ausrcl(),
            c"ringo".as_ptr(),
            Some(alloc_handler),
        )
    };
    if rc != 0 {
        return Err(format!("ausrc_register(ringo) failed (rc={rc})"));
    }
    let _ = AUSRC.set(asp as usize);

    let mut pp: *mut auplay = std::ptr::null_mut();
    let rc = unsafe {
        auplay_register(
            &mut pp,
            baresip_auplayl(),
            c"ringo".as_ptr(),
            Some(play_alloc_handler),
        )
    };
    if rc != 0 {
        return Err(format!("auplay_register(ringo) failed (rc={rc})"));
    }
    let _ = AUPLAY.set(pp as usize);
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn render_stream_same_rate_drains_in_order() {
        let mut q: VecDeque<i16> = (1..=6).collect();
        let mut pos = 0.0;
        let mut out = [0i16; 4];
        render_stream(Some(&mut q), &mut pos, 1.0, &mut out);
        assert_eq!(out, [1, 2, 3, 4]);
        assert_eq!(q.iter().copied().collect::<Vec<_>>(), vec![5, 6]); // 4 consumed
        // Next frame continues where we left off.
        let mut out2 = [0i16; 2];
        render_stream(Some(&mut q), &mut pos, 1.0, &mut out2);
        assert_eq!(out2, [5, 6]);
        assert!(q.is_empty());
    }

    #[test]
    fn render_stream_underrun_is_silence() {
        let mut q: VecDeque<i16> = VecDeque::from(vec![7, 8]);
        let mut pos = 0.0;
        let mut out = [0i16; 4];
        render_stream(Some(&mut q), &mut pos, 1.0, &mut out);
        assert_eq!(out, [7, 8, 0, 0]); // ran out → silence
        assert!(q.is_empty());
        assert_eq!(pos, 0.0); // reset on underrun
        // No subscription / no queue → all silence.
        let mut out2 = [9i16; 3];
        render_stream(None, &mut pos, 1.0, &mut out2);
        assert_eq!(out2, [0, 0, 0]);
    }

    #[test]
    fn render_stream_downsamples_2to1() {
        // step = in/out = 2.0 → take every other input sample.
        let mut q: VecDeque<i16> = (0..8).collect();
        let mut pos = 0.0;
        let mut out = [0i16; 4];
        render_stream(Some(&mut q), &mut pos, 2.0, &mut out);
        assert_eq!(out, [0, 2, 4, 6]);
        assert!(q.is_empty()); // consumed all 8
    }

    #[test]
    fn push_audio_is_noop_before_start_then_caps() {
        let key = "ringo-test-stream-cap"; // unique key, isolated from other tests
        // Before start_audio_stream: no queue exists, push is a no-op.
        push_audio(key, &[1, 2, 3]);
        assert!(stream_queue(key).is_none());

        // After start: samples are buffered, bounded by STREAM_IN_CAP_SAMPLES.
        start_audio_stream(key, 8000);
        push_audio(key, &vec![7i16; STREAM_IN_CAP_SAMPLES + 100]);
        let q = stream_queue(key).expect("queue exists after start");
        assert_eq!(
            q.lock().unwrap_or_else(|e| e.into_inner()).len(),
            STREAM_IN_CAP_SAMPLES
        );

        // Teardown drops the per-UA queue.
        remove_generator(key);
        assert!(stream_queue(key).is_none());
    }

    #[test]
    fn prepare_buffer_creates_keeps_and_resets_on_rate_change() {
        let key = "ringo-test-prepare";
        // Missing → created empty at the negotiated rate.
        prepare_buffer(rx_buffers(), key, 8000);
        assert_eq!(buffer_window(rx_buffers(), key), Some((vec![], 8000)));
        capture_mono(rx_buffers(), key, &[1, 2, 3]);
        // Same rate → samples survive (a hold/resume re-alloc must not wipe).
        prepare_buffer(rx_buffers(), key, 8000);
        assert_eq!(
            buffer_window(rx_buffers(), key),
            Some((vec![1, 2, 3], 8000))
        );
        // Different rate → reset (mixing rates would corrupt the WAV).
        prepare_buffer(rx_buffers(), key, 16000);
        assert_eq!(buffer_window(rx_buffers(), key), Some((vec![], 16000)));
        remove_generator(key);
    }

    // FULL_CAPTURE is a process-global static shared by parallel tests, so
    // every test that toggles it lives in this ONE test fn.
    #[test]
    fn arm_capture_scopes_retention_and_allocs() {
        let key = "ringo-test-arm";
        let srate = 8000u32;
        let sec = srate as usize;

        // Armed: a re-arm clears the previous recording (fresh start), the
        // buffers keep their srate, and retention is capped at 600 s.
        arm_capture(key, true);
        prepare_buffer(tx_buffers(), key, srate);
        prepare_buffer(rx_buffers(), key, srate);
        capture_mono(tx_buffers(), key, &vec![7i16; sec * 601]);
        assert_eq!(
            buffer_window(tx_buffers(), key).unwrap().0.len(),
            sec * FULL_RETAIN_SECS
        );
        arm_capture(key, true);
        assert_eq!(
            (
                buffer_window(tx_buffers(), key),
                buffer_window(rx_buffers(), key)
            ),
            (Some((vec![], srate)), Some((vec![], srate)))
        );

        // With capture on, a (re)alloc keeps samples (hold/resume survives).
        capture_mono(rx_buffers(), key, &[3]);
        alloc_buffer(rx_buffers(), key, srate);
        assert_eq!(buffer_window(rx_buffers(), key), Some((vec![3], srate)));

        // Disarmed mid-call: the recording is kept as is — audio still
        // arriving must neither append to it nor trim it to the 3 s window…
        let long = vec![3i16; sec * 10];
        capture_mono(rx_buffers(), key, &long[1..]);
        arm_capture(key, false);
        capture_mono(rx_buffers(), key, &vec![5i16; sec * 4]);
        assert_eq!(
            buffer_window(rx_buffers(), key),
            Some((long.clone(), srate))
        );
        // …and a second disarm doesn't change that.
        arm_capture(key, false);
        assert_eq!(
            buffer_window(rx_buffers(), key).unwrap().0.len(),
            long.len()
        );
        // The next stream (re)alloc starts fresh with the 3 s verify window.
        alloc_buffer(rx_buffers(), key, srate);
        assert_eq!(buffer_window(rx_buffers(), key), Some((vec![], srate)));
        capture_mono(rx_buffers(), key, &vec![5i16; sec * 4]);
        assert_eq!(
            buffer_window(rx_buffers(), key).unwrap().0.len(),
            sec * VERIFY_RETAIN_SECS
        );
        // A new arm unfreezes and clears.
        arm_capture(key, false);
        arm_capture(key, true);
        assert_eq!(buffer_window(rx_buffers(), key), Some((vec![], srate)));

        // Leave the process-global flag off for the other tests.
        arm_capture(key, false);
        remove_generator(key);
    }
}