diff --git a/services/discord-gateway/src/modules/voice-pcm-ws/index.ts b/services/discord-gateway/src/modules/voice-pcm-ws/index.ts new file mode 100644 index 0000000..89ceaf4 --- /dev/null +++ b/services/discord-gateway/src/modules/voice-pcm-ws/index.ts @@ -0,0 +1 @@ +export { VoicePcmWsClient } from "./voicePcmWsClient.js"; diff --git a/services/discord-gateway/src/modules/voice-pcm-ws/voicePcmWsClient.ts b/services/discord-gateway/src/modules/voice-pcm-ws/voicePcmWsClient.ts new file mode 100644 index 0000000..c5d5a31 --- /dev/null +++ b/services/discord-gateway/src/modules/voice-pcm-ws/voicePcmWsClient.ts @@ -0,0 +1,151 @@ +/** + * Direct WebSocket client for streaming voice PCM to the backend. + * + * Bypasses Redis pub/sub entirely — gateway connects as a WS client + * to the backend's existing WebSocket server and sends pre-formatted + * binary frames (4-byte FNV-1a userId hash + raw Int16LE PCM). + * The backend forwards these unchanged to frontend clients. + */ + +import { createChildLogger } from "@bete/shared/logger"; +import WebSocket from "ws"; + +const logger = createChildLogger("voice-pcm-ws"); + +/** 32-bit FNV-1a hash — mirrors backend/ws/redis-bridge.ts and frontend useAudioPlayback.ts */ +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 class VoicePcmWsClient { + private ws: WebSocket | null = null; + private readonly token: string; + private readonly url: string; + private reconnectTimer: ReturnType | null = null; + private closed = false; + private reconnectMs = 1000; + + constructor(url: string, token: string) { + this.url = url; + this.token = token; + } + + /** Open connection and begin automatic reconnect loop. */ + connect(): void { + if (this.closed) return; + this.doConnect(); + } + + /** + * Send a PCM chunk to the backend. + * + * Binary frame format (identical to what frontend expects): + * Byte 0–3: FNV-1a 32-bit userId hash (UInt32LE) + * Byte 4+: PCM audio (24kHz mono Int16LE) + * + * Silently drops if not connected — real-time audio, stale + * chunks from a disconnect gap are useless. + */ + sendPcm(userId: string, pcmBuffer: Buffer): void { + if (!this.ws || this.ws.readyState !== WebSocket.OPEN) return; + + const hash = hashUserId(userId); + const frame = Buffer.alloc(4 + pcmBuffer.length); + frame.writeUInt32LE(hash, 0); + pcmBuffer.copy(frame, 4); + + try { + this.ws.send(frame); + } catch (err) { + logger.warn({ err }, "Failed to send PCM frame"); + } + } + + /** Graceful close — send close frame, clear reconnect timer. */ + async close(): Promise { + this.closed = true; + this.clearReconnectTimer(); + + if (!this.ws) return; + + // Remove listeners so reconnect doesn't fire during close + this.ws.removeAllListeners(); + + if (this.ws.readyState === WebSocket.OPEN) { + this.ws.close(1000, "shutdown"); + } + + this.ws = null; + } + + // ── internals ──────────────────────────────────────────────────────── + + private getUrlWithAuth(): string { + const sep = this.url.includes("?") ? "&" : "?"; + return `${this.url}${sep}token=${encodeURIComponent(this.token)}`; + } + + private doConnect(): void { + if (this.closed) return; + + const authUrl = this.getUrlWithAuth(); + logger.info({ url: this.url }, "Connecting to backend WebSocket"); + + try { + this.ws = new WebSocket(authUrl); + } catch (err) { + logger.error({ err }, "Failed to create WebSocket"); + this.scheduleReconnect(); + return; + } + + this.ws.on("open", () => { + logger.info("Voice PCM WS client connected"); + this.reconnectMs = 1000; // reset backoff on successful connect + }); + + this.ws.on("close", (code, reason) => { + logger.warn( + { code, reason: reason.toString() }, + "Voice PCM WS client disconnected", + ); + this.ws = null; + this.scheduleReconnect(); + }); + + this.ws.on("error", (err) => { + logger.error({ err: err.message }, "Voice PCM WS client error"); + // on("close") fires after on("error") — reconnect handled there + }); + } + + private scheduleReconnect(): void { + if (this.closed) return; + this.clearReconnectTimer(); + + // Exponential backoff with jitter: 1s → 2s → 4s → ... → 30s cap + const delay = this.reconnectMs + Math.random() * 1000; + logger.info({ delayMs: Math.round(delay) }, "Scheduling WS reconnect"); + + this.reconnectTimer = setTimeout(() => { + this.reconnectTimer = null; + if (!this.closed) { + this.doConnect(); + } + }, delay); + + this.reconnectMs = Math.min(this.reconnectMs * 2, 30_000); + } + + private clearReconnectTimer(): void { + if (this.reconnectTimer !== null) { + clearTimeout(this.reconnectTimer); + this.reconnectTimer = null; + } + } +}