diff --git a/services/backend/src/ws/redis-bridge.ts b/services/backend/src/ws/redis-bridge.ts index 37fa00f..cc1a9e2 100644 --- a/services/backend/src/ws/redis-bridge.ts +++ b/services/backend/src/ws/redis-bridge.ts @@ -25,7 +25,7 @@ import { import { createChildLogger } from "@bete/shared/logger"; import Redis from "ioredis"; import { config } from "../shared/config/index.js"; -import { broadcastEvent } from "./broadcast.js"; +import { broadcastBinary, broadcastEvent } from "./broadcast.js"; const logger = createChildLogger("ws.redis-bridge"); @@ -96,6 +96,25 @@ function handleSubscriptionMessage(channel: string, message: string): void { // We only want , not the full envelope. const data = envelope.data !== undefined ? envelope.data : envelope; + // Voice PCM: decode base64 → binary broadcast instead of JSON + if (mapping.eventType === "voice_pcm_data") { + const pcmPayload = data as { userId?: string; pcm?: string }; + if (pcmPayload?.pcm && pcmPayload?.userId) { + try { + const pcmBuffer = Buffer.from(pcmPayload.pcm, "base64"); + // Prepend userId as 4-byte FNV-1a hash + const userIdHash = hashUserId(pcmPayload.userId); + const binary = Buffer.alloc(4 + pcmBuffer.length); + binary.writeUInt32LE(userIdHash, 0); + pcmBuffer.copy(binary, 4); + broadcastBinary(binary); + return; + } catch { + // fallback to JSON broadcast on error + } + } + } + logger.debug( { channel, eventType: mapping.eventType }, "Broadcasting Redis event", @@ -103,6 +122,16 @@ function handleSubscriptionMessage(channel: string, message: string): void { broadcastEvent(mapping.eventType, data); } +/** Simple 32-bit FNV-1a hash for userId → 4-byte identifier */ +function hashUserId(userId: string): number { + let hash = 0x811c9dc5; + for (let i = 0; i < userId.length; i++) { + hash ^= userId.charCodeAt(i); + hash = Math.imul(hash, 0x01000193); + } + return hash >>> 0; +} + export async function startRedisBridge(): Promise { if (!config.REDIS_URL) { logger.info("Redis not configured, skipping Redis bridge"); diff --git a/services/backend/src/ws/server.ts b/services/backend/src/ws/server.ts index e9df76f..f589e15 100644 --- a/services/backend/src/ws/server.ts +++ b/services/backend/src/ws/server.ts @@ -78,6 +78,40 @@ export function createWebSocketServer(server: Server): WebSocketServer { ); ws.on("message", (data: Buffer) => { + // Handle binary PCM from browser (FE→Discord transmit) + // Format: 4-byte magic "PCM\0" + raw PCM Int16 LE + if ( + Buffer.isBuffer(data) && + data.length > 4 && + data[0] === 0x50 && // 'P' + data[1] === 0x43 && // 'C' + data[2] === 0x4d && // 'M' + data[3] === 0x00 // '\0' + ) { + const pcmBuffer = data.subarray(4); + const base64 = pcmBuffer.toString("base64"); + import("../shared/redis/index.js").then( + ({ getCommandPublisher }) => { + const publisher = getCommandPublisher(); + publisher + .publish( + BACKEND_VOICE_TRANSMIT, + JSON.stringify({ + type: "pcm", + buffer: base64, + }), + ) + .catch((err: Error) => { + logger.error( + { err }, + "Failed to publish voice transmit to Redis", + ); + }); + }, + ); + return; + } + // Handle JSON messages from browser if ( typeof data === "string" || @@ -87,7 +121,7 @@ export function createWebSocketServer(server: Server): WebSocketServer { const message = JSON.parse(data.toString()); if (message.type === "voice_transmit" && message.buffer) { - // Forward PCM data to Redis for discord-gateway + // Legacy: Forward PCM data to Redis for discord-gateway import("../shared/redis/index.js").then( ({ getCommandPublisher }) => { const publisher = getCommandPublisher(); diff --git a/services/frontend/src/App.tsx b/services/frontend/src/App.tsx index b0426e7..f654e08 100644 --- a/services/frontend/src/App.tsx +++ b/services/frontend/src/App.tsx @@ -72,8 +72,7 @@ export default function App() { }; const socket = useDashboardSocket({ - onVoicePcmData: (d) => - audio.handleIncomingPcm(d as { userId: string; pcm: string }), + onBinary: (d) => audio.handleIncomingBinary(d), onUserState: (users) => setActiveSpeakers( (users as (ActiveSpeaker & { heardAt?: number })[]).map((u) => ({ @@ -81,17 +80,20 @@ export default function App() { heardAt: Date.now(), })), ), - onVoiceActiveUser: (data) => + onVoiceActiveUser: (data) => { + const d = data as { userId?: string; id?: string; username: string; avatar: string; speaking: boolean }; + if (d.userId) audio.registerUserId(d.userId); setActiveSpeakers((prev) => updateSpeakerList( prev, - data as Partial & { + d as Partial & { userId?: string; id?: string; speaking: boolean; }, ), - ), + ); + }, onVoiceRecordingStarted: () => window.dispatchEvent(new CustomEvent("voice_recording_uploaded")), onVoiceRecordingStopped: () => diff --git a/services/frontend/src/shared/hooks/useAudioPlayback.ts b/services/frontend/src/shared/hooks/useAudioPlayback.ts index 95ad8a5..933398d 100644 --- a/services/frontend/src/shared/hooks/useAudioPlayback.ts +++ b/services/frontend/src/shared/hooks/useAudioPlayback.ts @@ -14,6 +14,9 @@ const LEVEL_SHAPE = Array.from( (_, i) => 0.3 + (Math.sin(i * 0.6) * 0.35 + 0.65) * 0.7, ); +/** Reverse lookup: userIdHash → userId, populated by handleIncomingBinary */ +const userIdHashToId = new Map(); + export function useAudioPlayback() { const [isListening, setIsListening] = useState(false); const [levels, setLevels] = useState( @@ -42,11 +45,84 @@ export function useAudioPlayback() { } }, []); + /** + * Handle incoming binary PCM from WS. + * Format per chunk: 4-byte userId hash (UInt32LE) + raw PCM (Int16). + * userId hash → userId mapping is populated by voice_active_user events. + */ + const handleIncomingBinary = useCallback( + (buffer: ArrayBuffer) => { + const view = new DataView(buffer); + if (buffer.byteLength < 5) return; // Need at least 4-byte hash + 1 PCM byte + const userIdHash = view.getUint32(0, true); + const userId = userIdHashToId.get(userIdHash) ?? `user:${userIdHash}`; + const pcmBytes = buffer.byteLength - 4; + if (pcmBytes === 0) return; + + const int16Array = new Int16Array( + buffer, + 4, + pcmBytes / 2, + ); + if (int16Array.length === 0) return; + + // RMS + level computation (same as before) + let sumSquares = 0; + const float32Array = new Float32Array(int16Array.length); + for (let i = 0; i < int16Array.length; i++) { + const normalized = int16Array[i] / 32768; + float32Array[i] = normalized; + sumSquares += normalized * normalized; + } + const rms = Math.sqrt(sumSquares / int16Array.length); + const dbLevel = Math.min(1, Math.max(0.04, rms * 8)); + + setLevels((prev) => + prev.map((_, index) => + Math.max(0.04, dbLevel * LEVEL_SHAPE[index] * 5), + ), + ); + + const audioContext = audioContextRef.current; + if (!isListening || !audioContext) return; + + const audioBuffer = audioContext.createBuffer( + CHANNELS, + float32Array.length, + SAMPLE_RATE, + ); + audioBuffer.getChannelData(0).set(float32Array); + + const source = audioContext.createBufferSource(); + source.buffer = audioBuffer; + source.connect(audioContext.destination); + + const currentTime = audioContext.currentTime; + let nextStart = userTimelinesRef.current.get(userId) || 0; + if (nextStart < currentTime) nextStart = currentTime + 0.05; + source.start(nextStart); + userTimelinesRef.current.set( + userId, + nextStart + audioBuffer.duration, + ); + pruneTimelines(); + }, + [isListening, pruneTimelines], + ); + + /** + * Register a userId → hash mapping from voice_active_user events. + */ + const registerUserId = useCallback((userId: string) => { + const hash = fnv1a32(userId); + userIdHashToId.set(hash, userId); + }, []); + + // Legacy JSON handler kept for backward compat const handleIncomingPcm = useCallback( (data: { userId: string; pcm: string }) => { // Decode base64 PCM data try { - // 5a: Replace manual charCodeAt loop with Uint8Array.from const bytes = Uint8Array.from(atob(data.pcm), (c) => c.charCodeAt(0)); if (bytes.length === 0) return; const int16Array = new Int16Array( @@ -133,7 +209,20 @@ export function useAudioPlayback() { isListening, levels, handleIncomingPcm, + handleIncomingBinary, + registerUserId, toggleListening, audioContextRef, }; } + +/** 32-bit FNV-1a hash for userId → consistent 4-byte identifier */ +function fnv1a32(str: string): number { + let hash = 0x811c9dc5; + for (let i = 0; i < str.length; i++) { + hash ^= str.charCodeAt(i); + hash = Math.imul(hash, 0x01000193); + } + return hash >>> 0; +} + diff --git a/services/frontend/src/shared/hooks/useAudioTransmit.ts b/services/frontend/src/shared/hooks/useAudioTransmit.ts index 34cd663..100eab0 100644 --- a/services/frontend/src/shared/hooks/useAudioTransmit.ts +++ b/services/frontend/src/shared/hooks/useAudioTransmit.ts @@ -145,20 +145,13 @@ export function useAudioTransmit(socketRef: { for (let i = 0; i < inputData.length; i++) pcmData[i] = Math.max(-1, Math.min(1, inputData[i])) * 32767; - // 6b: Safe loop instead of spread operator to avoid call-stack overflow - const bytes = new Uint8Array(pcmData.buffer); - let str = ''; - for (let i = 0; i < bytes.length; i++) { - str += String.fromCharCode(bytes[i]); - } - const base64 = btoa(str); - - socketRef.current.send( - JSON.stringify({ - type: "voice_transmit", - buffer: base64, - }), - ); + // Send as binary: 4-byte magic "PCM\0" + raw PCM Int16 + const magic = new Uint8Array([0x50, 0x43, 0x4d, 0x00]); // "PCM\0" + const pcmBytes = new Uint8Array(pcmData.buffer); + const buf = new Uint8Array(magic.length + pcmBytes.length); + buf.set(magic, 0); + buf.set(pcmBytes, magic.length); + socketRef.current.send(buf.buffer); }; }, [socketRef]);