refactor(discord-gateway): optimize redis usage and audio stream handling
- refactor(command-handler): replace per-call Redis connection creation with a persistent publisher connection to reduce overhead - refactor(recorder): switch from manual audio stream subscription to direct event listeners on the existing stream - feat(recorder): implement exponential backoff for voice connection retries - chore(config): update default DECODER_COOLDOWN_MS to 30000ms
This commit is contained in:
@@ -53,11 +53,13 @@ const MEDIA_STATUS_KEY = "media:status";
|
|||||||
|
|
||||||
export class CommandHandler {
|
export class CommandHandler {
|
||||||
private redisSub: Redis;
|
private redisSub: Redis;
|
||||||
|
private redisPub: Redis;
|
||||||
private client: Client | null = null;
|
private client: Client | null = null;
|
||||||
private voiceController: VoiceController | null = null;
|
private voiceController: VoiceController | null = null;
|
||||||
|
|
||||||
constructor() {
|
constructor() {
|
||||||
this.redisSub = new Redis(config.REDIS_URL);
|
this.redisSub = new Redis(config.REDIS_URL);
|
||||||
|
this.redisPub = new Redis(config.REDIS_URL);
|
||||||
|
|
||||||
this.redisSub.on("error", (err) => {
|
this.redisSub.on("error", (err) => {
|
||||||
logger.error({ error: err }, "Redis subscriber connection error");
|
logger.error({ error: err }, "Redis subscriber connection error");
|
||||||
@@ -99,7 +101,7 @@ export class CommandHandler {
|
|||||||
}
|
}
|
||||||
|
|
||||||
async close(): Promise<void> {
|
async close(): Promise<void> {
|
||||||
await this.redisSub.quit();
|
await Promise.allSettled([this.redisSub.quit(), this.redisPub.quit()]);
|
||||||
}
|
}
|
||||||
|
|
||||||
// ---- Command dispatch ----
|
// ---- Command dispatch ----
|
||||||
@@ -395,19 +397,14 @@ export class CommandHandler {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Fire-and-forget SET on a separate Redis connection so we never block the
|
* Fire-and-forget SET using the persistent Redis publisher connection.
|
||||||
* subscriber loop.
|
|
||||||
*/
|
*/
|
||||||
private setKey(key: string, value: string): void {
|
private setKey(key: string, value: string): void {
|
||||||
const redis = new Redis(config.REDIS_URL);
|
this.redisPub
|
||||||
redis
|
|
||||||
.set(key, value)
|
.set(key, value)
|
||||||
.catch((err: unknown) => {
|
.catch((err: unknown) => {
|
||||||
const msg = err instanceof Error ? err.message : String(err);
|
const msg = err instanceof Error ? err.message : String(err);
|
||||||
logger.warn({ key, error: msg }, "Failed to update Redis status key");
|
logger.warn({ key, error: msg }, "Failed to update Redis status key");
|
||||||
})
|
|
||||||
.finally(() => {
|
|
||||||
void redis.quit();
|
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -15,7 +15,6 @@ import type { Client, VoiceChannel } from "discord.js-selfbot-v13";
|
|||||||
import { config } from "../../shared/config/config.js";
|
import { config } from "../../shared/config/config.js";
|
||||||
import type { PcmBroadcaster } from "../message-capture/types.js";
|
import type { PcmBroadcaster } from "../message-capture/types.js";
|
||||||
import { PacketFilter } from "./packetFilter.js";
|
import { PacketFilter } from "./packetFilter.js";
|
||||||
import { subscribeToAudioStream } from "./recorder/audioStream.js";
|
|
||||||
import { OpusDecoder } from "./recorder/decoder.js";
|
import { OpusDecoder } from "./recorder/decoder.js";
|
||||||
import {
|
import {
|
||||||
collectUserMetadata,
|
collectUserMetadata,
|
||||||
@@ -92,9 +91,9 @@ export async function startRecording(
|
|||||||
config.VOICE_CONNECTION_TIMEOUT_MS,
|
config.VOICE_CONNECTION_TIMEOUT_MS,
|
||||||
),
|
),
|
||||||
{
|
{
|
||||||
retries: 0,
|
retries: 3,
|
||||||
minTimeout: 0,
|
minTimeout: 1000,
|
||||||
maxTimeout: 0,
|
maxTimeout: 5000,
|
||||||
},
|
},
|
||||||
);
|
);
|
||||||
logger.info("Connected to voice channel. Recording started");
|
logger.info("Connected to voice channel. Recording started");
|
||||||
@@ -237,16 +236,16 @@ export async function startRecording(
|
|||||||
logger.error({ userId, error: msg }, "File write error");
|
logger.error({ userId, error: msg }, "File write error");
|
||||||
});
|
});
|
||||||
|
|
||||||
// Feed Opus packets one-by-one
|
// Attach event handlers directly to the existing audioStream (no double subscription)
|
||||||
subscribeToAudioStream(receiver, userId, {
|
audioStream.on("data", (chunk: Buffer) => {
|
||||||
onPacket: (chunk) => {
|
|
||||||
if (chunk.length < 8) return;
|
if (chunk.length < 8) return;
|
||||||
segmentManager.rotateIfNeeded(oggPacketStream);
|
segmentManager.rotateIfNeeded(oggPacketStream);
|
||||||
if (!broadcaster.broadcastPcmToWeb) return;
|
if (!broadcaster.broadcastPcmToWeb) return;
|
||||||
decoder.rotateIfNeeded();
|
decoder.rotateIfNeeded();
|
||||||
decoder.write(chunk);
|
decoder.write(chunk);
|
||||||
},
|
});
|
||||||
onEnd: () => {
|
|
||||||
|
audioStream.on("end", () => {
|
||||||
segmentManager.close(oggPacketStream);
|
segmentManager.close(oggPacketStream);
|
||||||
decoder.destroy();
|
decoder.destroy();
|
||||||
broadcaster.updateActiveUser?.(userId, {
|
broadcaster.updateActiveUser?.(userId, {
|
||||||
@@ -254,12 +253,12 @@ export async function startRecording(
|
|||||||
avatar: userMetadata.avatarUrl,
|
avatar: userMetadata.avatarUrl,
|
||||||
speaking: false,
|
speaking: false,
|
||||||
});
|
});
|
||||||
},
|
});
|
||||||
onError: (error) => {
|
|
||||||
|
audioStream.on("error", (error: Error) => {
|
||||||
segmentManager.close(oggPacketStream);
|
segmentManager.close(oggPacketStream);
|
||||||
decoder.destroy();
|
decoder.destroy();
|
||||||
logger.error({ userId, error: error.message }, "Audio stream error");
|
logger.error({ userId, error: error.message }, "Audio stream error");
|
||||||
},
|
|
||||||
});
|
});
|
||||||
|
|
||||||
packetFilterForOgg.on("error", (err) => {
|
packetFilterForOgg.on("error", (err) => {
|
||||||
|
|||||||
@@ -21,7 +21,7 @@ const configSchema = z
|
|||||||
RECORDINGS_DIR: z.string().default("./recordings"),
|
RECORDINGS_DIR: z.string().default("./recordings"),
|
||||||
RECORDING_SEGMENT_MS: z.coerce.number().positive().default(5000),
|
RECORDING_SEGMENT_MS: z.coerce.number().positive().default(5000),
|
||||||
DECODER_ROTATE_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),
|
WEBSERVER_PORT: z.coerce.number().positive().default(3000),
|
||||||
VOICE_CONNECTION_TIMEOUT_MS: z.coerce.number().positive().default(15000),
|
VOICE_CONNECTION_TIMEOUT_MS: z.coerce.number().positive().default(15000),
|
||||||
RECONNECT_TIMEOUT_MS: z.coerce.number().positive().default(5000),
|
RECONNECT_TIMEOUT_MS: z.coerce.number().positive().default(5000),
|
||||||
|
|||||||
Reference in New Issue
Block a user