path: root/src/recording/mod.rs

//! 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;
                }
            }
        }
    }
}