From 25115ac447e5356ca865e625515eb7c1f55b39bc Mon Sep 17 00:00:00 2001 From: salmonik Date: Wed, 9 Sep 2026 13:42:38 +0200 Subject: first draft --- src/commands/record.rs | 102 ++++++++++++++++ src/recording/mod.rs | 308 +++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 410 insertions(+) create mode 100644 src/commands/record.rs create mode 100644 src/recording/mod.rs diff --git a/src/commands/record.rs b/src/commands/record.rs new file mode 100644 index 0000000..6f2cfcd --- /dev/null +++ b/src/commands/record.rs @@ -0,0 +1,102 @@ +use tracing::warn; + +use crate::error::Result; +use crate::lang; +use crate::recording; + +type Context<'a> = poise::Context<'a, crate::Data, crate::error::Error>; + +/// Dump the last (up to 15 min) of voice activity as an mp3 to Copyparty. +/// +/// The bot continuously records (into a rolling in-memory ring) while it is +/// connected to a voice channel, if `RECORDING_ENABLED=true`. This command is a +/// read-only snapshot — the buffer keeps recording afterwards. Only callable by +/// someone in the same voice channel as the bot. +#[poise::command(slash_command)] +pub async fn record(ctx: Context<'_>) -> Result<()> { + ctx.defer().await?; + let data = ctx.data(); + + if !data.voice_manager.recording_enabled() { + ctx.say(lang::L_RECORD_DISABLED).await?; + return Ok(()); + } + + let guild_id = ctx.guild_id().ok_or_else(|| lang::err_not_in_guild())?; + + // ── Same-VC gate (mirrors /join's voice-state check) ─────────────── + let caller_channel = ctx.serenity_context().cache.guild(guild_id).and_then(|g| { + g.voice_states + .get(&ctx.author().id) + .and_then(|vs| vs.channel_id) + }); + let caller_channel = match caller_channel { + Some(c) => c, + None => { + ctx.say(lang::L_RECORD_NOT_IN_VOICE).await?; + return Ok(()); + } + }; + + let bot_channel = match data.voice_manager.get_call(guild_id) { + Some(call) => call.lock().await.current_channel(), + None => None, + }; + + match bot_channel { + None => { + ctx.say(lang::L_RECORD_NOT_RECORDING).await?; + Ok(()) + } + Some(bc) if bc.0.get() != caller_channel.get() => { + ctx.say(lang::L_RECORD_NOT_SAME_VC).await?; + Ok(()) + } + Some(_) => record_impl(ctx).await, + } +} + +async fn record_impl(ctx: Context<'_>) -> Result<()> { + let data = ctx.data(); + + let Some(rec) = data.voice_manager.get_recorder().await else { + ctx.say(lang::L_RECORD_NOT_RECORDING).await?; + return Ok(()); + }; + + let pcm = rec.flush(); + let sample_rate = rec.sample_rate(); + if pcm.is_empty() { + ctx.say(lang::L_RECORD_NOTHING).await?; + return Ok(()); + } + + let mp3 = match recording::encode_mp3(&pcm, sample_rate).await { + Ok(b) => b, + Err(e) => { + warn!("recording mp3 encode failed: {e}"); + ctx.say(lang::L_RECORD_ENCODE_FAILED).await?; + return Ok(()); + } + }; + + let filename = format!( + "nagranie-{}.mp3", + chrono::Local::now().format("%Y-%m-%d_%H-%M-%S") + ); + + match data.copyparty.upload(&filename, mp3).await { + Ok(url) => { + ctx.say(lang::fmt_record_uploaded(&url)).await?; + } + Err(e) => { + warn!("recording upload failed: {e}"); + ctx.say(lang::L_RECORD_UPLOAD_FAILED).await?; + } + } + Ok(()) +} + +pub fn commands() -> Vec> { + vec![record()] +} diff --git a/src/recording/mod.rs b/src/recording/mod.rs new file mode 100644 index 0000000..6ee4b11 --- /dev/null +++ b/src/recording/mod.rs @@ -0,0 +1,308 @@ +//! 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; + } + } + } + } +} -- cgit v1.2.3