first draft

1 files changed, 308 insertions, 0 deletions

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<Recorder>`. The
+//! sole strong `Arc<Recorder>` 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<Inner>,
+    sample_rate: u32,
+    samples_per_frame: usize,
+    window_samples: usize,
+}
+
+struct Inner {
+    ring: Vec<i16>,
+    /// 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<u8> {
+        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<i16> = 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<Recorder>`, so dropping the strong `Arc` in
+/// `VoiceManager.recorder` immediately frees the ring; subsequent ticks no-op.
+pub struct RecorderEventHandler {
+    weak: Weak<Recorder>,
+}
+
+impl RecorderEventHandler {
+    pub fn new(recorder: &Arc<Recorder>) -> Self {
+        Self {
+            weak: Arc::downgrade(recorder),
+        }
+    }
+}
+
+#[async_trait::async_trait]
+impl EventHandler for RecorderEventHandler {
+    async fn act(&self, ctx: &EventContext<'_>) -> Option<Event> {
+        // 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<Vec<u8>> {
+    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<serenity::GuildId>,
+    old_channel: Option<serenity::ChannelId>,
+    new_channel: Option<serenity::ChannelId>,
+) {
+    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;
+                }
+            }
+        }
+    }
+}