//! Voice channel recording: a rolling in-memory PCM ring buffer holding the //! last `window` seconds of mixed audio, flushable to an mp3 on demand. //! //! ## How audio arrives //! songbird delivers a `CoreEvent::VoiceTick` every 20 ms carrying //! already-decoded, jitter-buffered mono i16 PCM per speaking SSRC — but only //! when the driver is configured with `DecodeMode::Decode(Channels::Mono, //! SampleRate::Hz48000)` (set in `main.rs` via `register_songbird_from_config`). //! That tick is *both* the audio source and the wall-clock driver: there is no //! separate tokio timer, no `CancellationToken`, no per-SSRC Opus decoder, no //! f32→s16 conversion. songbird pays the decode cost; we only mix. //! //! ## Lifecycle / memory //! The `EventHandler` registered on the `Call` holds a `Weak`. The //! sole strong `Arc` lives in `VoiceManager.recorder`. Tearing that //! slot down (`None`) drops the ring immediately — ~86 MB at 15 min / 48 kHz — //! even while the songbird `Call` itself lingers. While the slot is `None`, //! `Weak::upgrade` in the handler returns `None` and the tick is a no-op. //! //! ## Ring semantics //! `head` points at the first sample of the 20 ms window currently being //! written. Each tick: zero `[head, head+960)` (clearing audio aged out by one //! window-length, which is how silence is placed when nobody speaks), saturating //! add every speaker's 960 mono i16 samples into that window, then advance //! `head += 960` (mod `window_samples`). The newest live sample is therefore //! `head - 1`; the oldest is `head` (exactly one window old once the ring has //! wrapped). `flush` reads `min(elapsed, window, written)` samples oldest-→newest. use std::process::Stdio; use std::sync::{Arc, Mutex, Weak}; use std::time::Instant; use poise::serenity_prelude as serenity; use songbird::{Event, EventContext, EventHandler}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tracing::warn; use crate::error::{Error, Result}; /// The songbird mixer clock fires one `VoiceTick` per 20 ms ⇒ 50 ticks/sec. const TICKS_PER_SEC: usize = 50; pub struct Recorder { inner: Mutex, sample_rate: u32, samples_per_frame: usize, window_samples: usize, } struct Inner { ring: Vec, /// First sample index of the 20 ms window currently being written. head: usize, /// Monotonic count of 20 ms frames written since recording started. frames_written: u64, started: Instant, } impl Recorder { pub fn new(sample_rate: u32, window_seconds: u64) -> Self { let samples_per_frame = (sample_rate as usize) / TICKS_PER_SEC; let window_samples = (sample_rate as usize) * (window_seconds as usize); // `head` advances by `samples_per_frame` and wraps mod `window_samples`, // so the window length must be a whole number of frames. debug_assert!( window_samples % samples_per_frame == 0, "window must be a whole number of 20 ms frames" ); Self { inner: Mutex::new(Inner { ring: vec![0i16; window_samples], head: 0, frames_written: 0, started: Instant::now(), }), sample_rate, samples_per_frame, window_samples, } } pub fn sample_rate(&self) -> u32 { self.sample_rate } /// Snapshot `min(elapsed, window, written)` seconds of audio, oldest→newest, /// as little-endian mono i16 bytes (the `s16le` format ffmpeg consumes). /// /// Read-only: the ring is left intact (flush-and-keep, per the `/record` /// design — successive calls may overlap). pub fn flush(&self) -> Vec { let inner = match self.inner.lock() { Ok(g) => g, Err(e) => { warn!("recorder mutex poisoned: {e}"); return Vec::new(); } }; let len = self.window_samples; let spf = self.samples_per_frame; let written_samples = inner .frames_written .saturating_mul(spf as u64) .min(len as u64) as usize; let elapsed_samples = (inner.started.elapsed().as_secs_f64() * self.sample_rate as f64) as usize; let d = written_samples.min(elapsed_samples).min(len); if d == 0 { return Vec::new(); } // Oldest live sample sits at `head - d` (mod len); read `d` forward. let start = (inner.head + len - d) % len; let mut samples: Vec = Vec::with_capacity(d); if start + d <= len { samples.extend_from_slice(&inner.ring[start..start + d]); } else { let first = len - start; samples.extend_from_slice(&inner.ring[start..]); samples.extend_from_slice(&inner.ring[..d - first]); } let mut bytes = Vec::with_capacity(samples.len() * 2); for s in &samples { bytes.extend_from_slice(&s.to_le_bytes()); } bytes } } /// `songbird::EventHandler` registered on each `Call` while recording is /// enabled. Holds only a `Weak`, so dropping the strong `Arc` in /// `VoiceManager.recorder` immediately frees the ring; subsequent ticks no-op. pub struct RecorderEventHandler { weak: Weak, } impl RecorderEventHandler { pub fn new(recorder: &Arc) -> Self { Self { weak: Arc::downgrade(recorder), } } } #[async_trait::async_trait] impl EventHandler for RecorderEventHandler { async fn act(&self, ctx: &EventContext<'_>) -> Option { // Only registered for `CoreEvent::VoiceTick`, but match defensively. let tick = match ctx { EventContext::VoiceTick(t) => t, _ => return None, }; let Some(rec) = self.weak.upgrade() else { // Recorder torn down — stop doing work, stay registered (cheap no-op) // until songbird drops the Call and us along with it. return None; }; let spf = rec.samples_per_frame; let len = rec.window_samples; let mut inner = match rec.inner.lock() { Ok(g) => g, Err(_) => return None, }; // Zero the window about to be written. This is the silence-placement // step: it clears samples that aged out one window-length ago, so a // tick with no speakers writes pure silence rather than stale audio. for i in 0..spf { let pos = (inner.head + i) % len; inner.ring[pos] = 0; } // Mix every speaking SSRC's decoded mono i16 into the window. // `saturating_add` prevents wraparound when several speakers overlap // (the only real failure mode on the hot path); multi-speaker overlaps // may clip slightly, the correct trade for a 50 Hz × N-packet mix. for vd in tick.speaking.values() { if let Some(audio) = vd.decoded_voice.as_ref() { let n = audio.len().min(spf); for i in 0..n { let pos = (inner.head + i) % len; inner.ring[pos] = inner.ring[pos].saturating_add(audio[i]); } } } inner.head = (inner.head + spf) % len; inner.frames_written += 1; let _ = rec; // drop the strong ref obtained via upgrade() None // keep listening } } /// Encode little-endian mono s16 PCM to mp3 via an ffmpeg subprocess. /// /// Mirrors the bot's other ffmpeg-subprocess patterns (e.g. `ffmpeg_stream_input`) /// but for a *finite* buffer: write all PCM to stdin (EOF on drop), drain stderr /// so ffmpeg can't block on a full pipe, collect mp3 from stdout. pub async fn encode_mp3(pcm_s16le: &[u8], sample_rate: u32) -> Result> { let mut child = tokio::process::Command::new("ffmpeg") .args([ "-y", "-f", "s16le", "-ar", &sample_rate.to_string(), "-ac", "1", "-i", "pipe:0", "-c:a", "libmp3lame", "-b:a", "64k", "-f", "mp3", "pipe:1", ]) .stdin(Stdio::piped()) .stdout(Stdio::piped()) .stderr(Stdio::piped()) .kill_on_drop(true) .spawn() .map_err(|e| Error::Custom(format!("ffmpeg spawn: {e}")))?; let stdin = child.stdin.take(); let stdout = child .stdout .take() .ok_or_else(|| Error::Custom("ffmpeg stdout not captured".into()))?; let stderr = child.stderr.take(); let pcm_owned = pcm_s16le.to_vec(); let write_task = tokio::spawn(async move { if let Some(mut stdin) = stdin { let _ = stdin.write_all(&pcm_owned).await; // `stdin` drops here ⇒ ffmpeg sees EOF on its input. } }); // Drain stderr in the background so a full pipe can't deadlock ffmpeg, // and surface the trace in the logs if encoding fails. if let Some(stderr) = stderr { tokio::spawn(async move { let mut buf = Vec::with_capacity(4096); let mut reader = tokio::io::BufReader::new(stderr); let _ = reader.read_to_end(&mut buf).await; if !buf.is_empty() { warn!( "ffmpeg (mp3 encode) stderr: {}", String::from_utf8_lossy(&buf).trim_end() ); } }); } let mut mp3 = Vec::new(); let mut reader = tokio::io::BufReader::new(stdout); reader .read_to_end(&mut mp3) .await .map_err(|e| Error::Custom(format!("ffmpeg read: {e}")))?; let _ = write_task.await; let status = child .wait() .await .map_err(|e| Error::Custom(format!("ffmpeg wait: {e}")))?; if !status.success() { return Err(Error::Custom( "ffmpeg mp3 encode failed (non-zero exit)".into(), )); } Ok(mp3) } /// Reconcile the recorder with the bot's own voice state. /// /// Called from `events::handle_voice_state_update` for the bot itself (before /// the existing bot-skip). The explicit-join path (`VoiceManager::connect`) /// starts the recorder itself; this handles the cases `connect` can't see: /// admin moves (re-setup if the recorder was dropped) and all leaves — /// `/stop`, kicks, gateway drops — which tear it down. pub async fn on_bot_voice_state( voice_manager: &crate::voice::VoiceManager, guild_id: Option, old_channel: Option, new_channel: Option, ) { if !voice_manager.recording_enabled() { return; } // Already torn down by `VoiceManager::disconnect` in the `/stop` path; this // is the catch-all for leaves `disconnect` didn't initiate (kicks, drops). if new_channel.is_none() { voice_manager.stop_recording().await; return; } // Joined or moved into a channel. If the recorder is missing (e.g. a move // broke the prior Call), try to (re)register on the current Call. if new_channel != old_channel { let existing = voice_manager.get_recorder().await.is_some(); if !existing { if let Some(gid) = guild_id { if let Some(call) = voice_manager.get_call(gid) { voice_manager.start_recording(&call).await; } } } } }