diff --git a/services/discord-gateway/src/modules/command-handler/commandHandler.ts b/services/discord-gateway/src/modules/command-handler/commandHandler.ts index a1af7b0..bd64330 100644 --- a/services/discord-gateway/src/modules/command-handler/commandHandler.ts +++ b/services/discord-gateway/src/modules/command-handler/commandHandler.ts @@ -53,11 +53,13 @@ const MEDIA_STATUS_KEY = "media:status"; export class CommandHandler { private redisSub: Redis; + private redisPub: Redis; private client: Client | null = null; private voiceController: VoiceController | null = null; constructor() { this.redisSub = new Redis(config.REDIS_URL); + this.redisPub = new Redis(config.REDIS_URL); this.redisSub.on("error", (err) => { logger.error({ error: err }, "Redis subscriber connection error"); @@ -99,7 +101,7 @@ export class CommandHandler { } async close(): Promise { - await this.redisSub.quit(); + await Promise.allSettled([this.redisSub.quit(), this.redisPub.quit()]); } // ---- Command dispatch ---- @@ -395,19 +397,14 @@ export class CommandHandler { } /** - * Fire-and-forget SET on a separate Redis connection so we never block the - * subscriber loop. + * Fire-and-forget SET using the persistent Redis publisher connection. */ private setKey(key: string, value: string): void { - const redis = new Redis(config.REDIS_URL); - redis + this.redisPub .set(key, value) .catch((err: unknown) => { const msg = err instanceof Error ? err.message : String(err); logger.warn({ key, error: msg }, "Failed to update Redis status key"); - }) - .finally(() => { - void redis.quit(); }); } } diff --git a/services/discord-gateway/src/modules/voice-recording/recorder.ts b/services/discord-gateway/src/modules/voice-recording/recorder.ts index 687edf8..8436c50 100644 --- a/services/discord-gateway/src/modules/voice-recording/recorder.ts +++ b/services/discord-gateway/src/modules/voice-recording/recorder.ts @@ -15,7 +15,6 @@ import type { Client, VoiceChannel } from "discord.js-selfbot-v13"; import { config } from "../../shared/config/config.js"; import type { PcmBroadcaster } from "../message-capture/types.js"; import { PacketFilter } from "./packetFilter.js"; -import { subscribeToAudioStream } from "./recorder/audioStream.js"; import { OpusDecoder } from "./recorder/decoder.js"; import { collectUserMetadata, @@ -92,9 +91,9 @@ export async function startRecording( config.VOICE_CONNECTION_TIMEOUT_MS, ), { - retries: 0, - minTimeout: 0, - maxTimeout: 0, + retries: 3, + minTimeout: 1000, + maxTimeout: 5000, }, ); logger.info("Connected to voice channel. Recording started"); @@ -237,29 +236,29 @@ export async function startRecording( logger.error({ userId, error: msg }, "File write error"); }); - // Feed Opus packets one-by-one - subscribeToAudioStream(receiver, userId, { - onPacket: (chunk) => { - if (chunk.length < 8) return; - segmentManager.rotateIfNeeded(oggPacketStream); - if (!broadcaster.broadcastPcmToWeb) return; - decoder.rotateIfNeeded(); - decoder.write(chunk); - }, - onEnd: () => { - segmentManager.close(oggPacketStream); - decoder.destroy(); - broadcaster.updateActiveUser?.(userId, { - username: userMetadata.username, - avatar: userMetadata.avatarUrl, - speaking: false, - }); - }, - onError: (error) => { - segmentManager.close(oggPacketStream); - decoder.destroy(); - logger.error({ userId, error: error.message }, "Audio stream error"); - }, + // Attach event handlers directly to the existing audioStream (no double subscription) + audioStream.on("data", (chunk: Buffer) => { + if (chunk.length < 8) return; + segmentManager.rotateIfNeeded(oggPacketStream); + if (!broadcaster.broadcastPcmToWeb) return; + decoder.rotateIfNeeded(); + decoder.write(chunk); + }); + + audioStream.on("end", () => { + segmentManager.close(oggPacketStream); + decoder.destroy(); + broadcaster.updateActiveUser?.(userId, { + username: userMetadata.username, + avatar: userMetadata.avatarUrl, + speaking: false, + }); + }); + + audioStream.on("error", (error: Error) => { + segmentManager.close(oggPacketStream); + decoder.destroy(); + logger.error({ userId, error: error.message }, "Audio stream error"); }); packetFilterForOgg.on("error", (err) => { diff --git a/services/discord-gateway/src/shared/config/config.ts b/services/discord-gateway/src/shared/config/config.ts index 029667b..6527b8c 100644 --- a/services/discord-gateway/src/shared/config/config.ts +++ b/services/discord-gateway/src/shared/config/config.ts @@ -21,7 +21,7 @@ const configSchema = z RECORDINGS_DIR: z.string().default("./recordings"), RECORDING_SEGMENT_MS: z.coerce.number().positive().default(5000), DECODER_ROTATE_MS: z.coerce.number().positive().default(5000), - DECODER_COOLDOWN_MS: z.coerce.number().positive().default(0), + DECODER_COOLDOWN_MS: z.coerce.number().positive().default(30000), WEBSERVER_PORT: z.coerce.number().positive().default(3000), VOICE_CONNECTION_TIMEOUT_MS: z.coerce.number().positive().default(15000), RECONNECT_TIMEOUT_MS: z.coerce.number().positive().default(5000),