feat(leptos): Phase 4 Task 6 - PCM audio playback + mic transmit hooks

services/frontend-leptos/Cargo.lock

@@ -492,6 +492,7 @@ version = "0.1.0"
 dependencies = [
  "console_error_panic_hook",
  "gloo-net",
+ "gloo-timers",
  "js-sys",
  "leptos 0.7.8",
  "leptos-use",

services/frontend-leptos/frontend/Cargo.toml

@@ -23,10 +23,16 @@ web-sys = { version = "0.3", features = [
     "AudioContext",
     "AudioBuffer",
     "AudioBufferSourceNode",
+    "AudioDestinationNode",
+    "AudioNode",
+    "AudioProcessingEvent",
+    "MediaStreamAudioSourceNode",
+    "ScriptProcessorNode",
     "Window",
     "Document",
     "Element",
     "HtmlElement",
+    "HtmlSelectElement",
     "KeyboardEvent",
     "Storage",
     "IntersectionObserver",
@@ -42,10 +48,13 @@ web-sys = { version = "0.3", features = [
     "HtmlCanvasElement",
     "MediaDevices",
     "MediaStream",
+    "MediaStreamConstraints",
+    "MediaStreamTrack",
     "Navigator",
     "console",
 ] }
 gloo-net = "0.6"
+gloo-timers = "0.3"
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 serde-wasm-bindgen = "0.6"

services/frontend-leptos/frontend/src/features/live/audio/mod.rs

@@ -0,0 +1,2 @@
+pub mod ring_buffer;
+pub mod pcm_decoder;

services/frontend-leptos/frontend/src/features/live/audio/pcm_decoder.rs

@@ -0,0 +1,69 @@
+/// PCM Frame decoded from binary WebSocket data
+/// Format: [u32 userId (4 bytes)][i16 samples (N bytes)]
+pub struct PcmFrame {
+    pub user_id: u32,
+    pub samples: Vec<f32>, // Normalized to [-1.0, 1.0]
+}
+
+/// Decode a binary WebSocket message into PCM frames
+/// Returns None if data is too short or malformed
+pub fn decode_pcm_frame(data: &[u8]) -> Option<PcmFrame> {
+    if data.len() < 4 {
+        return None;
+    }
+
+    let user_id = u32::from_le_bytes([data[0], data[1], data[2], data[3]]);
+    let sample_bytes = &data[4..];
+    let sample_count = sample_bytes.len() / 2;
+
+    if sample_count == 0 {
+        return None;
+    }
+
+    let samples = decode_i16_samples(sample_bytes);
+    Some(PcmFrame { user_id, samples })
+}
+
+/// Decode raw i16 PCM bytes to normalized f32 samples [-1.0, 1.0]
+pub fn decode_i16_samples(data: &[u8]) -> Vec<f32> {
+    let count = data.len() / 2;
+    let mut out = Vec::with_capacity(count);
+
+    for i in 0..count {
+        let offset = i * 2;
+        if offset + 1 < data.len() {
+            let sample = i16::from_le_bytes([data[offset], data[offset + 1]]);
+            out.push((sample as f32) / 32768.0);
+        }
+    }
+
+    out
+}
+
+/// Encode f32 samples [-1.0, 1.0] to base64 for WebSocket transmission
+/// Uses JavaScript btoa for encoding
+pub fn encode_samples_to_base64(samples: &[f32]) -> String {
+    // Convert f32 samples to i16 bytes
+    let mut bytes = Vec::with_capacity(samples.len() * 2);
+    for &sample in samples {
+        let clamped = sample.max(-1.0).min(1.0);
+        let int_sample = (clamped * 32767.0) as i16;
+        bytes.extend_from_slice(&int_sample.to_le_bytes());
+    }
+    encode_bytes_base64(&bytes)
+}
+
+/// Encode raw bytes to base64 using JavaScript's btoa
+fn encode_bytes_base64(data: &[u8]) -> String {
+    // Build binary string for btoa
+    let binary: String = data.iter().map(|&b| b as char).collect();
+
+    // Call btoa from JavaScript via js_sys::eval
+    let js_code = format!("btoa('{}')", binary.replace('\'', "\\'"));
+    js_sys::eval(&js_code)
+        .ok()
+        .and_then(|r| r.as_string())
+        .unwrap_or_default()
+}
+
+use wasm_bindgen::prelude::*;

services/frontend-leptos/frontend/src/features/live/audio/ring_buffer.rs

@@ -0,0 +1,124 @@
+use std::sync::{Arc, Mutex};
+
+/// AudioRingBuffer — Fixed-size circular buffer for real-time PCM streaming
+/// Provides thread-safe write/read with automatic overwrite protection
+pub struct AudioRingBuffer {
+    buffer: Vec<f32>,
+    capacity: usize,
+    write_pos: usize,
+    read_pos: usize,
+    available: usize,
+}
+
+impl AudioRingBuffer {
+    /// Create a new ring buffer with given capacity (in samples)
+    pub fn new(capacity: usize) -> Self {
+        Self {
+            buffer: vec![0.0; capacity],
+            capacity,
+            write_pos: 0,
+            read_pos: 0,
+            available: 0,
+        }
+    }
+
+    /// Write samples to the ring buffer. Overwrites oldest data if full.
+    pub fn write(&mut self, samples: &[f32]) {
+        let mut written = 0;
+        while written < samples.len() {
+            let chunk = (samples.len() - written).min(self.capacity - self.write_pos);
+            let src = &samples[written..written + chunk];
+            let dest = &mut self.buffer[self.write_pos..self.write_pos + chunk];
+            dest.copy_from_slice(src);
+            written += chunk;
+            self.write_pos = (self.write_pos + chunk) % self.capacity;
+            self.available = (self.available + chunk).min(self.capacity);
+            // If we overwrote unread data, advance read_pos
+            if self.available == self.capacity {
+                self.read_pos = self.write_pos;
+            }
+        }
+    }
+
+    /// Read up to `max_samples` from the buffer. Returns the samples read.
+    pub fn read(&mut self, max_samples: usize) -> Vec<f32> {
+        let to_read = max_samples.min(self.available);
+        let mut out = Vec::with_capacity(to_read);
+        let mut remaining = to_read;
+
+        while remaining > 0 {
+            let chunk = remaining.min(self.capacity - self.read_pos);
+            out.extend_from_slice(&self.buffer[self.read_pos..self.read_pos + chunk]);
+            remaining -= chunk;
+            self.read_pos = (self.read_pos + chunk) % self.capacity;
+        }
+
+        self.available -= to_read;
+        out
+    }
+
+    /// Number of samples available to read
+    pub fn available_samples(&self) -> usize {
+        self.available
+    }
+
+    /// Clear all buffered data
+    pub fn clear(&mut self) {
+        self.write_pos = 0;
+        self.read_pos = 0;
+        self.available = 0;
+    }
+}
+
+/// Thread-safe wrapper around AudioRingBuffer
+pub struct SharedRingBuffer {
+    inner: Arc<Mutex<AudioRingBuffer>>,
+}
+
+impl SharedRingBuffer {
+    pub fn new(capacity: usize) -> Self {
+        Self {
+            inner: Arc::new(Mutex::new(AudioRingBuffer::new(capacity))),
+        }
+    }
+
+    pub fn write(&self, samples: &[f32]) {
+        if let Ok(mut guard) = self.inner.lock() {
+            guard.write(samples);
+        }
+    }
+
+    pub fn read(&self, max_samples: usize) -> Vec<f32> {
+        if let Ok(mut guard) = self.inner.lock() {
+            guard.read(max_samples)
+        } else {
+            Vec::new()
+        }
+    }
+
+    pub fn available_samples(&self) -> usize {
+        if let Ok(guard) = self.inner.lock() {
+            guard.available_samples()
+        } else {
+            0
+        }
+    }
+
+    pub fn clear(&self) {
+        if let Ok(mut guard) = self.inner.lock() {
+            guard.clear();
+        }
+    }
+
+    pub fn clone_inner(&self) -> Arc<Mutex<AudioRingBuffer>> {
+        self.inner.clone()
+    }
+}
+
+impl Clone for SharedRingBuffer {
+    fn clone(&self) -> Self {
+        Self {
+            inner: self.inner.clone(),
+        }
+    }
+}

services/frontend-leptos/frontend/src/features/live/hooks/mod.rs

@@ -1,2 +1,4 @@
 pub mod use_voice_control;
 pub mod use_media_control;
+pub mod use_audio_playback;
+pub mod use_audio_transmit;

services/frontend-leptos/frontend/src/features/live/hooks/use_audio_playback.rs

@@ -0,0 +1,107 @@
+use leptos::prelude::*;
+use std::sync::Arc;
+use crate::features::live::audio::pcm_decoder::decode_pcm_frame;
+use crate::features::live::audio::ring_buffer::SharedRingBuffer;
+
+/// AudioPlaybackState — Manages PCM audio playback from WebSocket binary frames
+pub struct AudioPlaybackState {
+    /// Ring buffer for incoming PCM data
+    pub buffer: SharedRingBuffer,
+    /// Whether playback is active
+    pub active: RwSignal<bool>,
+    /// Volume level (0.0-1.0)
+    pub volume: RwSignal<f64>,
+}
+
+/// Create and initialize audio playback state
+pub fn use_audio_playback() -> AudioPlaybackState {
+    let buffer = SharedRingBuffer::new(44100 * 5); // 5 seconds at 44.1kHz
+    let active = create_rw_signal::<bool>(false);
+    let volume = create_rw_signal::<f64>(0.5);
+
+    AudioPlaybackState {
+        buffer,
+        active,
+        volume,
+    }
+}
+
+/// Process incoming binary data from WebSocket (PCM audio frame)
+/// Format: [u32 userId (4 bytes)][i16 samples (N bytes)]
+pub fn process_pcm_data(state: &AudioPlaybackState, data: Vec<u8>) {
+    if let Some(frame) = decode_pcm_frame(&data) {
+        state.buffer.write(&frame.samples);
+    }
+}
+
+/// Start consuming the ring buffer and playing through AudioContext
+pub fn start_playback(state: &AudioPlaybackState) {
+    if state.active.get() {
+        return;
+    }
+    state.active.set(true);
+
+    let buffer = state.buffer.clone();
+    let active = state.active;
+
+    wasm_bindgen_futures::spawn_local(async move {
+        let ctx = match web_sys::AudioContext::new() {
+            Ok(ctx) => ctx,
+            Err(_) => {
+                active.set(false);
+                return;
+            }
+        };
+
+        let ctx_ref = &ctx;
+        let _ = ctx_ref.resume();
+
+        while active.get() {
+            let available = buffer.available_samples();
+            if available >= 4410 {
+                // ~100ms worth at 44.1kHz
+                let samples = buffer.read(4410);
+                if !samples.is_empty() {
+                    play_samples(&ctx, &samples);
+                }
+            }
+            let _ = gloo_timers::future::TimeoutFuture::new(50).await;
+        }
+
+        let _ = ctx.close();
+    });
+}
+
+/// Stop playback and clear buffer
+pub fn stop_playback(state: &AudioPlaybackState) {
+    state.active.set(false);
+    state.buffer.clear();
+}
+
+/// Play a chunk of PCM samples through AudioContext using AudioBufferSourceNode
+fn play_samples(ctx: &web_sys::AudioContext, samples: &[f32]) {
+    let frame_count = samples.len() as u32;
+    let Ok(audio_buffer) = ctx.create_buffer(1, frame_count, ctx.sample_rate()) else {
+        return;
+    };
+
+    // Write samples into the buffer channel
+    let Ok(channel_data) = audio_buffer.get_channel_data(0) else {
+        return;
+    };
+
+    let len = samples.len().min(channel_data.len() as usize);
+    if len == 0 {
+        return;
+    }
+
+    // Copy samples directly to audio buffer channel
+    let _ = audio_buffer.copy_to_channel(&samples[..len], 0);
+
+    // Create source and play
+    if let Ok(source) = ctx.create_buffer_source() {
+        source.set_buffer(Some(&audio_buffer));
+        source.set_loop(false);
+        let _ = source.start();
+    }
+}

services/frontend-leptos/frontend/src/features/live/hooks/use_audio_transmit.rs

@@ -0,0 +1,79 @@
+use leptos::prelude::*;
+use wasm_bindgen::prelude::*;
+use wasm_bindgen_futures::spawn_local;
+use web_sys::{MediaStream, MediaStreamConstraints, MediaStreamTrack};
+
+/// AudioTransmitState — Manages microphone capture state
+pub struct AudioTransmitState {
+    pub active: RwSignal<bool>,
+    pub stream: StoredValue<Option<MediaStream>>,
+}
+
+/// Create microphone transmit state
+pub fn use_audio_transmit() -> AudioTransmitState {
+    let active = create_rw_signal::<bool>(false);
+    let stream = StoredValue::new(None::<MediaStream>);
+    AudioTransmitState { active, stream }
+}
+
+/// Start microphone capture - requests getUserMedia and stores the stream
+pub fn start_transmit(state: &AudioTransmitState) {
+    if state.active.get() {
+        return;
+    }
+    state.active.set(true);
+
+    let constraints = MediaStreamConstraints::new();
+    let _ = js_sys::Reflect::set(
+        &constraints,
+        &JsValue::from_str("audio"),
+        &JsValue::from_bool(true),
+    );
+
+    let window = match web_sys::window() {
+        Some(w) => w,
+        None => return,
+    };
+    let media_devices = match window.navigator().media_devices() {
+        Ok(md) => md,
+        Err(_) => return,
+    };
+
+    let promise = match media_devices.get_user_media_with_constraints(&constraints) {
+        Ok(p) => p,
+        Err(_) => return,
+    };
+
+    // Clone signals before spawning async task to avoid reference escaping
+    let active_signal = state.active;
+    let stream_signal = state.stream;
+
+    spawn_local(async move {
+        match wasm_bindgen_futures::JsFuture::from(promise).await {
+            Ok(val) => {
+                if let Ok(s) = val.dyn_into::<MediaStream>() {
+                    stream_signal.set_value(Some(s));
+                }
+            }
+            Err(_) => {
+                active_signal.set(false);
+            }
+        }
+    });
+}
+
+/// Stop microphone transmission
+pub fn stop_transmit(state: &AudioTransmitState) {
+    state.active.set(false);
+    state.stream.update_value(|s| {
+        if let Some(stream) = s.take() {
+            let tracks = stream.get_tracks();
+            for i in 0..tracks.length() {
+                let track_val = tracks.get(i);
+                if let Ok(track) = track_val.dyn_into::<MediaStreamTrack>() {
+                    track.stop();
+                }
+            }
+        }
+    });
+}

services/frontend-leptos/frontend/src/features/live/mod.rs

@@ -1,5 +1,6 @@
 pub mod components;
 pub mod hooks;
+pub mod audio;
 
 use leptos::prelude::*;
 use crate::ui::card::Card;