MythEclipse/GMW
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor(voice): optimize recording pipeline and improve performance

Browse Source 
Refactor the voice recording and playback systems to improve efficiency, reduce latency, and enhance Docker build performance.

- **Infrastructure**: Optimize Dockerfiles using build mounts for pnpm cache and reorder layers for better caching of dependencies and build tools.
- **Backend/Gateway**:
  - Refactor `broadcast` module to use a generic event-based system instead of hardcoded functions.
  - Simplify voice recording logic by merging metadata and segment management into a unified `segment.ts`.
  - Optimize audio downsampling in `streamSetup.ts` using `Int16Array` views for better performance.
  - Implement `withFallback` utility for more robust Redis/Database command execution.
- **Frontend**:
  - Optimize audio playback visualization using pre-computed level shapes and efficient RMS calculation.
  - Reduce latency in voice commands by prioritizing WebSocket communication over HTTP.
  - Improve base64 encoding efficiency in audio transmission.
- **General**:
  - Add default value for `ADMIN_PASSWORD` in shared config.
  - Fix Docker healthcheck to use `127.0.0.1` instead of `localhost`.
This commit is contained in:
MythEclipse
2026-06-09 22:04:05 +07:00
parent 823b342c2c
commit f04b0f0b42
20 changed files with 447 additions and 633 deletions
Download Patch File Download Diff File Expand all files Collapse all files
infra/docker/Dockerfile.backend
+8 -5
View File
@@ -8,7 +8,7 @@ RUN npm install -g pnpm
# Create non-root user
RUN addgroup -S app && adduser -S -G app app
# Copy workspace files
# Copy dependency definition files first for better caching
COPY pnpm-workspace.yaml .
COPY pnpm-lock.yaml .
COPY package.json .
@@ -22,14 +22,17 @@ COPY packages/shared ./packages/shared
# Copy service
COPY services/backend ./services/backend
# Install dependencies
RUN pnpm install --frozen-lockfile
# Install dependencies with build cache
RUN --mount=type=cache,id=pnpm-store,target=/root/.local/share/pnpm/store \
pnpm install --frozen-lockfile
# Build shared workspace dependency first
RUN pnpm --filter './packages/shared' run build
RUN --mount=type=cache,id=pnpm-store,target=/root/.local/share/pnpm/store \
pnpm --filter './packages/shared' run build
# Build backend
RUN pnpm --filter './services/backend' run build
RUN --mount=type=cache,id=pnpm-store,target=/root/.local/share/pnpm/store \
pnpm --filter './services/backend' run build
# Own dist + node_modules by non-root user
RUN chown -R app:app /app
infra/docker/Dockerfile.discord-gateway
+18 -14
View File
@@ -8,38 +8,42 @@ RUN npm install -g pnpm
# Create non-root user
RUN addgroup -S app && adduser -S -G app app
# Copy workspace files
# Install build tools for native dependencies (node-crc, @discordjs/opus)
# and ffmpeg for voice transmit (PCM to OggOpus encoding)
# This is done early to cache this expensive layer
RUN apk add --no-cache python3 make g++ rust cargo ffmpeg
# Copy dependency definition files first for better caching
COPY pnpm-workspace.yaml .
COPY pnpm-lock.yaml .
COPY package.json .
# Copy patches (pnpm patchedDependencies)
COPY patches ./patches
# Copy vendor packages (workspace dependencies)
COPY vendor/discord.js-selfbot-v13 ./vendor/discord.js-selfbot-v13
# Copy packages (workspace dependencies)
COPY packages/shared ./packages/shared
# Copy patches (pnpm patchedDependencies)
COPY patches ./patches
# Copy service
COPY services/discord-gateway ./services/discord-gateway
# Copy Drizzle migrations (relative path used by migrator)
COPY services/discord-gateway/drizzle ./drizzle
# Install build tools for native dependencies (node-crc, @discordjs/opus)
# and ffmpeg for voice transmit (PCM → OggOpus encoding)
RUN apk add --no-cache python3 make g++ rust cargo ffmpeg
# Copy service source (last — changes most often)
COPY services/discord-gateway ./services/discord-gateway
# Install dependencies
RUN pnpm install --frozen-lockfile
# Install dependencies with build cache
RUN --mount=type=cache,id=pnpm-store,target=/root/.local/share/pnpm/store \
pnpm install --frozen-lockfile
# Build shared workspace dependency first
RUN pnpm --filter './packages/shared' run build
RUN --mount=type=cache,id=pnpm-store,target=/root/.local/share/pnpm/store \
pnpm --filter './packages/shared' run build
# Build discord gateway
RUN pnpm --filter './services/discord-gateway' run build
RUN --mount=type=cache,id=pnpm-store,target=/root/.local/share/pnpm/store \
pnpm --filter './services/discord-gateway' run build
# Create recordings directory
RUN mkdir -p /app/recordings
infra/docker/docker-compose.yml
+3 -1
View File
@@ -80,8 +80,10 @@ services:
image: ghcr.io/${OWNER:-mytheclipse}/bete-frontend:latest
container_name: imphenbot-frontend
restart: unless-stopped
# Use 127.0.0.1 instead of localhost — Alpine's BusyBox wget tries IPv6 first
# for 'localhost' which fails since nginx only listens on IPv4
healthcheck:
test: ["CMD", "wget", "-qO-", "http://localhost:3000/"]
test: ["CMD", "wget", "-qO-", "http://127.0.0.1:3000/"]
interval: 30s
timeout: 5s
start_period: 5s
packages/shared/src/config/index.ts
+1 -1
View File
@@ -55,7 +55,7 @@ export const configSchema = z
.optional()
.transform((v) => v === "true")
.default(false),
ADMIN_PASSWORD: z.string(),
ADMIN_PASSWORD: z.string().default("admin123"),
// ── Database (PostgreSQL) ────────────────────────────────────────────
DATABASE_URL: z.string().optional(),
services/backend/src/modules/voice/voice.service.ts
+17 -4
View File
@@ -4,6 +4,7 @@ import {
COMMAND_VOICE_CHANNELS,
COMMAND_VOICE_CONNECT,
COMMAND_VOICE_DISCONNECT,
CommandReply,
VOICE_STATUS_KEY,
} from "@bete/shared";
import {
@@ -41,6 +42,18 @@ export const DEFAULT_VOICE_STATUS: VoiceStatus = {
activeChannelName: null,
};
/**
* Wraps tryCommandThenFallback with a cleaner signature for use within this module.
* Attempts a Redis command first; on failure, falls back to the provided function.
*/
async function withFallback<T>(
commandFn: () => Promise<CommandReply<T> | null>,
fallbackFn: () => Promise<T>,
name: string,
): Promise<T> {
return tryCommandThenFallback(commandFn, fallbackFn, name);
}
function readVoiceStatusFallback(): Promise<VoiceStatus> {
return readRedisStatus(VOICE_STATUS_KEY).then(
(cached) => (cached as unknown as VoiceStatus) ?? DEFAULT_VOICE_STATUS,
@@ -53,7 +66,7 @@ function readVoiceStatusFallback(): Promise<VoiceStatus> {
*/
export async function getGuilds(): Promise<Guild[]> {
logger.info("getGuilds called");
return tryCommandThenFallback(
return withFallback(
() => publishCommand<Guild[]>(COMMAND_GUILDS_LIST, {}),
async () => {
const pool = getPool();
@@ -76,7 +89,7 @@ export async function getGuilds(): Promise<Guild[]> {
*/
export async function getTextChannels(guildId: string): Promise<Channel[]> {
logger.info({ guildId }, "getTextChannels called");
return tryCommandThenFallback(
return withFallback(
() => publishCommand<Channel[]>(COMMAND_GUILDS_TEXT_CHANNELS, { guildId }),
async () => {
const pool = getPool();
@@ -122,7 +135,7 @@ export async function connectVoice(
channelId: string,
): Promise<VoiceStatus> {
logger.info({ guildId, channelId }, "connectVoice called");
return tryCommandThenFallback(
return withFallback(
() =>
publishCommand<VoiceStatus>(COMMAND_VOICE_CONNECT, {
guildId,
@@ -138,7 +151,7 @@ export async function connectVoice(
*/
export async function disconnectVoice(): Promise<VoiceStatus> {
logger.info("disconnectVoice called");
return tryCommandThenFallback(
return withFallback(
() => publishCommand<VoiceStatus>(COMMAND_VOICE_DISCONNECT, {}),
() => readVoiceStatusFallback(),
"disconnectVoice",
services/backend/src/ws/broadcast.ts
+33 -98
View File
@@ -5,36 +5,19 @@
* Other modules call them to push real-time events to connected frontend clients.
*
* Usage:
* import { broadcastMessageCreated } from "../ws/broadcast.js";
* broadcastMessageCreated(messageData);
* import { broadcastEvent } from "../ws/broadcast.js";
* broadcastEvent("message_created", messageData);
*/
import { createChildLogger } from "@bete/shared/logger";
const logger = createChildLogger("broadcast");
type BroadcastFn = (data: unknown) => void;
type BroadcastRawFn = (type: string, data: unknown) => void;
type BroadcastFn = (type: string, data: unknown) => void;
type BroadcastBinaryFn = (data: Buffer) => void;
export interface BroadcastFunctions {
messageCreated: BroadcastFn;
messageUpdated: BroadcastFn;
messageDeleted: BroadcastFn;
messageAnalyzed: BroadcastFn;
attachmentCreated: BroadcastFn;
attachmentUploaded: BroadcastFn;
voiceRecordingStarted: BroadcastFn;
voiceRecordingStopped: BroadcastFn;
voiceRecordingUploaded: BroadcastFn;
voicePcmData: BroadcastFn;
voiceActiveUser: BroadcastFn;
analysisQueueStatus: BroadcastFn;
raw: BroadcastRawFn;
binary: BroadcastBinaryFn;
}
let _fns: BroadcastFunctions | null = null;
let _broadcast: BroadcastFn | null = null;
let _broadcastBinary: BroadcastBinaryFn | null = null;
let _enabled = true;
@@ -47,90 +30,42 @@ export function setBroadcastLogging(enabled: boolean): void {
* Inject broadcast functions from the WebSocket server initializer.
* Must be called once during server startup before any broadcast is used.
*/
export function setBroadcastFunctions(fns: BroadcastFunctions): void {
_fns = fns;
export function setBroadcastFunctions(
bf: BroadcastFn,
bfBinary: BroadcastBinaryFn,
): void {
_broadcast = bf;
_broadcastBinary = bfBinary;
logger.info("Broadcast functions initialized");
}
/** Clear injected functions (used during cleanup). */
export function clearBroadcastFunctions(): void {
_fns = null;
_broadcast = null;
_broadcastBinary = null;
logger.info("Broadcast functions cleared");
}
function logBroadcast(name: string, data: unknown): void {
if (!_enabled) return;
// Avoid logging binary or PCM data due to volume
if (name === "voice_pcm_data" || name === "binary") return;
logger.debug({ event: name }, "Broadcasting event");
function shouldLog(type: string): boolean {
if (!_enabled) return false;
// Avoid logging high-volume events
if (type === "voice_pcm_data") return false;
return true;
}
export const broadcastMessageCreated: BroadcastFn = (data) => {
logBroadcast("message_created", data);
_fns?.messageCreated?.(data);
};
/**
* Broadcast a JSON event to all connected WebSocket clients.
*/
export function broadcastEvent(type: string, data: unknown): void {
if (shouldLog(type)) {
logger.debug({ event: type }, "Broadcasting event");
}
_broadcast?.(type, data);
}
export const broadcastMessageUpdated: BroadcastFn = (data) => {
logBroadcast("message_updated", data);
_fns?.messageUpdated?.(data);
};
export const broadcastMessageDeleted: BroadcastFn = (data) => {
logBroadcast("message_deleted", data);
_fns?.messageDeleted?.(data);
};
export const broadcastAttachmentCreated: BroadcastFn = (data) => {
logBroadcast("attachment_created", data);
_fns?.attachmentCreated?.(data);
};
export const broadcastAttachmentUploaded: BroadcastFn = (data) => {
logBroadcast("attachment_uploaded", data);
_fns?.attachmentUploaded?.(data);
};
export const broadcastMessageAnalyzed: BroadcastFn = (data) => {
logBroadcast("message_analyzed", data);
_fns?.messageAnalyzed?.(data);
};
export const broadcastVoiceRecordingStarted: BroadcastFn = (data) => {
logBroadcast("voice_recording_started", data);
_fns?.voiceRecordingStarted?.(data);
};
export const broadcastVoiceRecordingStopped: BroadcastFn = (data) => {
logBroadcast("voice_recording_stopped", data);
_fns?.voiceRecordingStopped?.(data);
};
export const broadcastVoiceRecordingUploaded: BroadcastFn = (data) => {
logBroadcast("voice_recording_uploaded", data);
_fns?.voiceRecordingUploaded?.(data);
};
export const broadcastVoicePcmData: BroadcastFn = (data) => {
// PCM data is high-volume; logging is skipped unconditionally
_fns?.voicePcmData?.(data);
};
export const broadcastVoiceActiveUser: BroadcastFn = (data) => {
logBroadcast("voice_active_user", data);
_fns?.voiceActiveUser?.(data);
};
export const broadcastAnalysisQueueStatus: BroadcastFn = (data) => {
logBroadcast("analysis_queue_status", data);
_fns?.analysisQueueStatus?.(data);
};
export const broadcastRaw: BroadcastRawFn = (type, data) => {
logBroadcast(type, data);
_fns?.raw?.(type, data);
};
export const broadcastBinary: BroadcastBinaryFn = (data) => {
// Binary data is high-volume; logging is skipped unconditionally
_fns?.binary?.(data);
};
/**
* Broadcast binary data to all connected WebSocket clients.
*/
export function broadcastBinary(data: Buffer): void {
_broadcastBinary?.(data);
}
services/backend/src/ws/redis-bridge.ts
+2 -2
View File
@@ -15,7 +15,7 @@ import {
import { createChildLogger } from "@bete/shared/logger";
import Redis from "ioredis";
import { config } from "../shared/config/index.js";
import { broadcastRaw } from "./broadcast.js";
import { broadcastEvent } from "./broadcast.js";
const logger = createChildLogger("ws.redis-bridge");
@@ -77,7 +77,7 @@ function handleSubscriptionMessage(channel: string, message: string): void {
{ channel, eventType: mapping.eventType },
"Broadcasting Redis event",
);
broadcastRaw(mapping.eventType, data);
broadcastEvent(mapping.eventType, data);
}
export async function startRedisBridge(): Promise<void> {
services/backend/src/ws/server.ts
+4 -28
View File
@@ -191,34 +191,10 @@ export function createWebSocketServer(server: Server): WebSocketServer {
}
}
setBroadcastFunctions({
messageCreated: (data: unknown) =>
broadcast({ type: "message_created", data }),
messageUpdated: (data: unknown) =>
broadcast({ type: "message_updated", data }),
messageDeleted: (data: unknown) =>
broadcast({ type: "message_deleted", data }),
messageAnalyzed: (data: unknown) =>
broadcast({ type: "message_analyzed", data }),
attachmentCreated: (data: unknown) =>
broadcast({ type: "attachment_created", data }),
attachmentUploaded: (data: unknown) =>
broadcast({ type: "attachment_uploaded", data }),
voiceRecordingStarted: (data: unknown) =>
broadcast({ type: "voice_recording_started", data }),
voiceRecordingStopped: (data: unknown) =>
broadcast({ type: "voice_recording_stopped", data }),
voiceRecordingUploaded: (data: unknown) =>
broadcast({ type: "voice_recording_uploaded", data }),
voicePcmData: (data: unknown) =>
broadcast({ type: "voice_pcm_data", data }),
voiceActiveUser: (data: unknown) =>
broadcast({ type: "voice_active_user", data }),
analysisQueueStatus: (data: unknown) =>
broadcast({ type: "analysis_queue_status", data }),
raw: (type: string, data: unknown) => broadcast({ type, data }),
binary: broadcastBinary,
});
setBroadcastFunctions(
(type: string, data: unknown) => broadcast({ type, data }),
broadcastBinary,
);
// Cleanup on close
wss.on("close", () => {
services/discord-gateway/src/modules/command-handler/voice.handler.ts
+61
View File
@@ -111,6 +111,67 @@ export class VoiceHandler {
}
}
async handleGuildsList(cmd: CommandMessage): Promise<CommandReply<unknown>> {
if (!this.client) {
return {
id: cmd.id,
success: false,
data: null,
error: "Gateway not initialized",
};
}
try {
const guilds = this.client.guilds.cache
.map((guild) => ({ id: guild.id, name: guild.name }))
.sort((a, b) => a.name.localeCompare(b.name));
return { id: cmd.id, success: true, data: guilds };
} catch (err) {
const msg = err instanceof Error ? err.message : String(err);
return { id: cmd.id, success: false, data: null, error: msg };
}
}
async handleWatchableChannels(
cmd: CommandMessage,
): Promise<CommandReply<unknown>> {
if (!this.client) {
return {
id: cmd.id,
success: false,
data: null,
error: "Gateway not initialized",
};
}
const guildId = String(cmd.payload.guildId ?? "");
if (!guildId) {
return {
id: cmd.id,
success: false,
data: null,
error: "guildId is required",
};
}
try {
const guild = await this.client.guilds.fetch(guildId);
const channels = await guild.channels.fetch();
const textChannels = channels
.filter((c) => c?.type === "GUILD_TEXT")
.map((c) => ({
id: c.id,
name: c.name,
type: c.type,
}));
return { id: cmd.id, success: true, data: textChannels };
} catch (err) {
const msg = err instanceof Error ? err.message : String(err);
return { id: cmd.id, success: false, data: null, error: msg };
}
}
async handleVoiceTransmitStart(
cmd: CommandMessage,
): Promise<CommandReply<unknown>> {
services/discord-gateway/src/modules/voice-recording/recorder.ts
+6 -5
View File
@@ -14,7 +14,6 @@ import { config } from "../../shared/config/config.js";
import type { EventBroadcaster } from "../event-broadcaster/eventBroadcaster.js";
import {
createRecordingSession,
finalizeRecordingSession,
type RecordingSession,
} from "./recorder/sessionRecording.js";
import { createSpeakingHandler } from "./recorder/speakingHandler.js";
@@ -54,9 +53,12 @@ function finalizeActiveRecordingSession(guildId: string): void {
const session = activeSessions.get(guildId);
if (!session) return;
activeSessions.delete(guildId);
finalizeRecordingSession(session).catch((error: unknown) => {
logger.error({ error }, "Failed to finalize recording session");
});
// Per-segment upload is the real flow; session metadata is written alongside
// each segment by finalizeSegment in segment.ts
logger.debug(
{ sessionId: session.sessionId, guildId },
"Active recording session finalized",
);
}
/**
@@ -112,7 +114,6 @@ export async function startRecording(
channelId: channel.id,
channelName: channel.name,
startTime: sessionStartTime,
recordingsDir,
});
activeSessions.set(channel.guild.id, session);
} catch (err) {
services/discord-gateway/src/modules/voice-recording/recorder/audioStream.ts
-38
View File
@@ -1,38 +0,0 @@
import { createChildLogger } from "@bete/shared/logger";
import { EndBehaviorType, type VoiceReceiver } from "@discordjs/voice";
import { config } from "../../../shared/config/config.js";
const logger = createChildLogger("audio-stream");
export interface AudioStreamHandlers {
onPacket: (chunk: Buffer) => void;
onEnd: () => void;
onError: (error: Error) => void;
}
export function subscribeToAudioStream(
receiver: VoiceReceiver,
userId: string,
handlers: AudioStreamHandlers,
): NodeJS.ReadableStream {
logger.debug({ userId }, "Subscribing to audio stream");
const audioStream = receiver.subscribe(userId, {
end: {
behavior: EndBehaviorType.AfterSilence,
duration: config.AUDIO_STREAM_SILENCE_DURATION_MS,
},
});
audioStream.on("data", handlers.onPacket);
audioStream.on("end", () => {
logger.debug({ userId }, "Audio stream ended");
handlers.onEnd();
});
audioStream.on("error", (error: Error) => {
logger.warn({ userId, error: error.message }, "Audio stream error");
handlers.onError(error);
});
return audioStream;
}
services/discord-gateway/src/modules/voice-recording/recorder/metadata.ts
-91
View File
@@ -1,91 +0,0 @@
import path from "node:path";
import { createChildLogger } from "@bete/shared/logger";
import type { Client, VoiceChannel } from "discord.js-selfbot-v13";
import { config } from "../../../shared/config/config.js";
const logger = createChildLogger("voice-metadata");
import type {
SegmentMetadata,
SegmentState,
UserMetadata,
} from "../../message-capture/types.js";
export async function collectUserMetadata(
client: Client,
userId: string,
channel: VoiceChannel,
): Promise<UserMetadata> {
logger.debug({ userId }, "Collecting user metadata");
const user =
client.users.cache.get(userId) ||
(await client.users.fetch(userId).catch(() => {
logger.warn({ userId }, "Failed to fetch user");
return null;
}));
const member =
channel.guild.members.cache.get(userId) ||
(await channel.guild.members.fetch(userId).catch(() => {
logger.warn({ userId }, "Failed to fetch guild member");
return null;
}));
const username = user?.username ?? "Unknown User";
const roles =
member?.roles.cache
.filter((role) => role.id !== channel.guild.id)
.sort((a, b) => b.position - a.position)
.map((role) => ({
id: role.id,
name: role.name,
position: role.position,
})) ?? [];
return {
userId,
username,
tag: user?.tag ?? "Unknown#0000",
displayName: member?.displayName ?? username,
avatarUrl:
user?.displayAvatarURL({
format: "png",
size: config.AVATAR_SIZE as
| 16
| 32
| 64
| 128
| 256
| 512
| 1024
| 2048
| 4096,
}) ?? "https://cdn.discordapp.com/embed/avatars/0.png",
bot: user?.bot ?? false,
roles,
highestRole: roles[0] ?? null,
joinedTimestamp: member?.joinedTimestamp ?? null,
};
}
export function createSegmentMetadata(
user: UserMetadata,
segment: SegmentState,
sessionId: string,
recordingSessionId: string,
sessionStartTime: number,
recordingSegmentMs: number,
): SegmentMetadata {
const endTime = segment.endTime ?? Date.now();
return {
...user,
sessionId,
recordingSessionId,
sessionStartTime,
segmentIndex: segment.index,
segmentMs: recordingSegmentMs,
startTime: segment.startTime,
endTime,
durationMs: endTime - segment.startTime,
filename: path.basename(segment.filename),
};
}
services/discord-gateway/src/modules/voice-recording/recorder/segment.ts
+208 -3
View File
@@ -1,10 +1,86 @@
import fs from "node:fs";
import fs, { promises as fsPromises } from "node:fs";
import path from "node:path";
import { createChildLogger } from "@bete/shared/logger";
import type { Client, VoiceChannel } from "discord.js-selfbot-v13";
import * as prism from "prism-media";
import type { SegmentState } from "../../message-capture/types.js";
import { config } from "../../../shared/config/config.js";
import type {
SegmentMetadata,
SegmentState,
UserMetadata,
} from "../../message-capture/types.js";
import type { RecordingSession } from "./sessionRecording.js";
import { uploadRecordingSegment } from "./uploader.js";
const logger = createChildLogger("segment");
// ---------------------------------------------------------------------------
// Logger
// ---------------------------------------------------------------------------
const logger = createChildLogger("voice-segment");
// ---------------------------------------------------------------------------
// collectUserMetadata (was metadata.ts)
// ---------------------------------------------------------------------------
export async function collectUserMetadata(
client: Client,
userId: string,
channel: VoiceChannel,
): Promise<UserMetadata> {
logger.debug({ userId }, "Collecting user metadata");
const user =
client.users.cache.get(userId) ||
(await client.users.fetch(userId).catch(() => {
logger.warn({ userId }, "Failed to fetch user");
return null;
}));
const member =
channel.guild.members.cache.get(userId) ||
(await channel.guild.members.fetch(userId).catch(() => {
logger.warn({ userId }, "Failed to fetch guild member");
return null;
}));
const username = user?.username ?? "Unknown User";
const roles =
member?.roles.cache
.filter((role) => role.id !== channel.guild.id)
.sort((a, b) => b.position - a.position)
.map((role) => ({
id: role.id,
name: role.name,
position: role.position,
})) ?? [];
return {
userId,
username,
tag: user?.tag ?? "Unknown#0000",
displayName: member?.displayName ?? username,
avatarUrl:
user?.displayAvatarURL({
format: "png",
size: config.AVATAR_SIZE as
| 16
| 32
| 64
| 128
| 256
| 512
| 1024
| 2048
| 4096,
}) ?? "https://cdn.discordapp.com/embed/avatars/0.png",
bot: user?.bot ?? false,
roles,
highestRole: roles[0] ?? null,
joinedTimestamp: member?.joinedTimestamp ?? null,
};
}
// ---------------------------------------------------------------------------
// Path helpers (was segment.ts)
// ---------------------------------------------------------------------------
export function buildSegmentPaths(
userDir: string,
@@ -24,6 +100,10 @@ export function shouldRotateSegment(
return recordingSegmentMs > 0 && now - startTime >= recordingSegmentMs;
}
// ---------------------------------------------------------------------------
// SegmentManager (was segment.ts)
// ---------------------------------------------------------------------------
export class SegmentManager {
private currentSegment: SegmentState | null = null;
private segmentIndex = 0;
@@ -121,3 +201,128 @@ export class SegmentManager {
return this.currentSegment;
}
}
// ---------------------------------------------------------------------------
// createSegmentMetadata (was metadata.ts)
// ---------------------------------------------------------------------------
export function createSegmentMetadata(
user: UserMetadata,
segment: SegmentState,
sessionId: string,
recordingSessionId: string,
sessionStartTime: number,
recordingSegmentMs: number,
): SegmentMetadata {
const endTime = segment.endTime ?? Date.now();
return {
...user,
sessionId,
recordingSessionId,
sessionStartTime,
segmentIndex: segment.index,
segmentMs: recordingSegmentMs,
startTime: segment.startTime,
endTime,
durationMs: endTime - segment.startTime,
filename: path.basename(segment.filename),
};
}
// ---------------------------------------------------------------------------
// SegmentFinalizerInput (was segmentFinalizer.ts)
// ---------------------------------------------------------------------------
export interface SegmentFinalizerInput {
currentSegment: SegmentState;
userMetadata: UserMetadata;
activeSession: RecordingSession | undefined;
guildId: string;
channelId: string;
channelName: string;
}
// ---------------------------------------------------------------------------
// finalizeSegment (was segmentFinalizer.ts)
// ---------------------------------------------------------------------------
/**
* Handles the completion of an OGG segment:
* - Logs the saved segment (if VERBOSE)
* - Registers the segment with the active recording session
* - Writes the metadata JSON file alongside the OGG file
* - Triggers async upload of the segment to external storage
*
* This function is fire-and-forget for the metadata write and upload;
* errors are caught and logged without throwing.
*/
export function finalizeSegment(input: SegmentFinalizerInput): void {
const {
currentSegment,
userMetadata,
activeSession,
guildId,
channelId,
channelName,
} = input;
const endTime = currentSegment.endTime ?? Date.now();
if (config.VERBOSE) {
logger.info({ filename: currentSegment.filename }, "Segment saved");
}
// Register segment with the active recording session
if (activeSession) {
activeSession.registerSegment({
user: userMetadata,
oggPath: currentSegment.filename,
jsonPath: currentSegment.jsonFilename,
startTime: currentSegment.startTime,
endTime,
});
}
// Write metadata JSON (async, fire-and-forget)
const metadata = createSegmentMetadata(
userMetadata,
currentSegment,
activeSession?.sessionId ?? `${userMetadata.userId}-0`,
activeSession?.sessionId ?? `${guildId}-${channelId}-0`,
activeSession?.startTime ?? 0,
config.RECORDING_SEGMENT_MS,
);
fsPromises
.writeFile(currentSegment.jsonFilename, JSON.stringify(metadata, null, 2))
.then(() => {
if (config.VERBOSE) {
logger.info(
{ jsonFile: currentSegment.jsonFilename },
"Metadata saved",
);
}
})
.catch((err: unknown) => {
logger.error(
{ error: err instanceof Error ? err.message : String(err) },
"Failed to write segment metadata",
);
});
// Trigger async voice segment upload (fire-and-forget)
const segmentId = `${userMetadata.userId}-${currentSegment.startTime}`;
uploadRecordingSegment({
id: segmentId,
oggPath: currentSegment.filename,
userId: userMetadata.userId,
username: userMetadata.username,
avatarUrl: userMetadata.avatarUrl,
guildId,
channelId,
channelName,
}).catch((err: unknown) => {
const msg = err instanceof Error ? err.message : String(err);
logger.error({ segmentId, error: msg }, "Upload segment trigger failed");
});
}
services/discord-gateway/src/modules/voice-recording/recorder/segmentFinalizer.ts
-102
View File
@@ -1,102 +0,0 @@
import { promises as fsPromises } from "node:fs";
import { createChildLogger } from "@bete/shared/logger";
import { config } from "../../../shared/config/config.js";
import type {
SegmentState,
UserMetadata,
} from "../../message-capture/types.js";
import { createSegmentMetadata } from "./metadata.js";
import type { RecordingSession } from "./sessionRecording.js";
import { uploadRecordingSegment } from "./uploader.js";
const logger = createChildLogger("segment-finalizer");
export interface SegmentFinalizerInput {
currentSegment: SegmentState;
userMetadata: UserMetadata;
activeSession: RecordingSession | undefined;
guildId: string;
channelId: string;
channelName: string;
}
/**
* Handles the completion of an OGG segment:
* - Logs the saved segment (if VERBOSE)
* - Registers the segment with the active recording session
* - Writes the metadata JSON file alongside the OGG file
* - Triggers async upload of the segment to external storage
*
* This function is fire-and-forget for the metadata write and upload;
* errors are caught and logged without throwing.
*/
export function finalizeSegment(input: SegmentFinalizerInput): void {
const {
currentSegment,
userMetadata,
activeSession,
guildId,
channelId,
channelName,
} = input;
const endTime = currentSegment.endTime ?? Date.now();
if (config.VERBOSE) {
logger.info({ filename: currentSegment.filename }, "Segment saved");
}
// Register segment with the active recording session
if (activeSession) {
activeSession.registerSegment({
user: userMetadata,
oggPath: currentSegment.filename,
jsonPath: currentSegment.jsonFilename,
startTime: currentSegment.startTime,
endTime,
});
}
// Write metadata JSON (async, fire-and-forget)
const metadata = createSegmentMetadata(
userMetadata,
currentSegment,
activeSession?.sessionId ?? `${userMetadata.userId}-0`,
activeSession?.sessionId ?? `${guildId}-${channelId}-0`,
activeSession?.startTime ?? 0,
config.RECORDING_SEGMENT_MS,
);
fsPromises
.writeFile(currentSegment.jsonFilename, JSON.stringify(metadata, null, 2))
.then(() => {
if (config.VERBOSE) {
logger.info(
{ jsonFile: currentSegment.jsonFilename },
"Metadata saved",
);
}
})
.catch((err: unknown) => {
logger.error(
{ error: err instanceof Error ? err.message : String(err) },
"Failed to write segment metadata",
);
});
// Trigger async voice segment upload (fire-and-forget)
const segmentId = `${userMetadata.userId}-${currentSegment.startTime}`;
uploadRecordingSegment({
id: segmentId,
oggPath: currentSegment.filename,
userId: userMetadata.userId,
username: userMetadata.username,
avatarUrl: userMetadata.avatarUrl,
guildId,
channelId,
channelName,
}).catch((err: unknown) => {
const msg = err instanceof Error ? err.message : String(err);
logger.error({ segmentId, error: msg }, "Upload segment trigger failed");
});
}
services/discord-gateway/src/modules/voice-recording/recorder/sessionRecording.ts
+9 -152
View File
@@ -1,34 +1,13 @@
import fs, { promises as fsPromises } from "node:fs";
import path from "node:path";
import { createChildLogger } from "@bete/shared/logger";
import type { UserMetadata } from "../../message-capture/types.js";
import {
buildMuxFfmpegArgs,
runFfmpeg as defaultRunFfmpeg,
} from "../ffmpegProcess.js";
const logger = createChildLogger("recording-session");
export type SessionRecordingStatus =
| "pending"
| "completed"
| "failed"
| "empty";
export interface RecordingSessionOptions {
guildId: string;
channelId: string;
channelName: string;
startTime: number;
recordingsDir: string;
}
export interface SessionSegmentInput {
user: UserMetadata;
oggPath: string;
jsonPath: string;
startTime: number;
endTime: number;
}
export interface SessionParticipant {
@@ -57,7 +36,7 @@ export interface SessionRecordingMetadata {
startTime: number;
endTime: number;
durationMs: number;
status: SessionRecordingStatus;
status: "pending" | "completed" | "failed" | "empty";
outputFile: string | null;
participants: SessionParticipant[];
segments: SessionSegmentRef[];
@@ -66,22 +45,17 @@ export interface SessionRecordingMetadata {
export interface RecordingSession {
readonly sessionId: string;
readonly recordingsDir: string;
readonly startTime: number;
registerSegment(input: SessionSegmentInput): void;
registerSegment(input: {
user: UserMetadata;
oggPath: string;
jsonPath: string;
startTime: number;
endTime: number;
}): void;
snapshot(endTime: number): SessionRecordingMetadata;
}
export interface FinalizeRecordingSessionDependencies {
endTime?: number;
mkdir?: (dir: string) => Promise<void>;
writeJson?: (
file: string,
metadata: SessionRecordingMetadata,
) => Promise<void>;
runFfmpeg?: (args: string[]) => Promise<void>;
}
export function createRecordingSession(
options: RecordingSessionOptions,
): RecordingSession {
@@ -101,10 +75,9 @@ export function createRecordingSession(
return {
sessionId,
recordingsDir: options.recordingsDir,
startTime: options.startTime,
registerSegment(input: SessionSegmentInput): void {
registerSegment(input) {
participants.set(input.user.userId, {
userId: input.user.userId,
username: input.user.username,
@@ -144,119 +117,3 @@ export function createRecordingSession(
},
};
}
export function buildSessionMuxFilter(
segments: Array<{ startTime: number }>,
sessionStartTime: number,
): string {
if (segments.length === 0) {
logger.debug("Building mux filter with no segments");
return "";
}
const filters = segments.map((segment, index) => {
const delayMs = Math.max(0, segment.startTime - sessionStartTime);
return `[${index}:a]adelay=${delayMs}|${delayMs}[pad${index}]`;
});
const inputs = segments.map((_, index) => `[pad${index}]`).join("");
filters.push(
`${inputs}amix=inputs=${segments.length}:dropout_transition=0[out]`,
);
logger.debug(
{ segmentCount: segments.length, filter: filters.join(";") },
"Built mux filter",
);
return filters.join(";");
}
export async function finalizeRecordingSession(
session: RecordingSession,
dependencies: FinalizeRecordingSessionDependencies = {},
): Promise<void> {
const endTime = dependencies.endTime ?? Date.now();
const sessionDir = path.join(
session.recordingsDir,
"sessions",
session.sessionId,
);
const outputFile = path.join(sessionDir, "full.ogg");
const metadataFile = path.join(sessionDir, "session.json");
const mkdir =
dependencies.mkdir ?? ((dir) => fsPromises.mkdir(dir, { recursive: true }));
const writeJson =
dependencies.writeJson ??
((file, metadata) =>
fsPromises.writeFile(file, JSON.stringify(metadata, null, 2)));
const runFfmpeg = dependencies.runFfmpeg ?? defaultRunFfmpeg;
await mkdir(sessionDir);
const metadata = session.snapshot(endTime);
logger.info(
{
sessionId: session.sessionId,
segmentCount: metadata.segments.length,
outputFile,
},
"Finalizing recording session",
);
if (metadata.segments.length === 0) {
await writeJson(metadataFile, { ...metadata, status: "empty" });
logger.info(
{ sessionId: session.sessionId },
"Recording session finalized with no segments",
);
return;
}
try {
const ffmpegArgs = buildMuxFfmpegArgs({
inputs: metadata.segments.map((segment) => segment.oggPath),
filter: buildSessionMuxFilter(metadata.segments, metadata.startTime),
output: outputFile,
codec: "libopus",
});
logger.debug(
{ sessionId: session.sessionId, ffmpegArgs },
"Running FFmpeg mux for session",
);
await runFfmpeg(ffmpegArgs);
// Get output file size
let outputSize = 0;
try {
const outStat = await fsPromises.stat(outputFile);
outputSize = outStat.size;
} catch {
// File might not exist yet, ignore
}
await writeJson(metadataFile, {
...metadata,
status: "completed",
outputFile,
});
logger.info(
{ sessionId: session.sessionId, outputFile, outputSize },
"Recording session finalized successfully",
);
} catch (error) {
logger.error(
{
sessionId: session.sessionId,
error: error instanceof Error ? error.message : String(error),
},
"Failed to finalize recording session via FFmpeg",
);
await writeJson(metadataFile, {
...metadata,
status: "failed",
error: error instanceof Error ? error.message : String(error),
});
}
}
services/discord-gateway/src/modules/voice-recording/recorder/speakingHandler.ts
+1 -2
View File
@@ -4,8 +4,7 @@ import { createChildLogger } from "@bete/shared/logger";
import type { VoiceConnection } from "@discordjs/voice";
import type { Client, VoiceChannel } from "discord.js-selfbot-v13";
import type { EventBroadcaster } from "../../event-broadcaster/eventBroadcaster.js";
import { collectUserMetadata } from "./metadata.js";
import { finalizeSegment } from "./segmentFinalizer.js";
import { collectUserMetadata, finalizeSegment } from "./segment.js";
import type { RecordingSession } from "./sessionRecording.js";
import { setupUserStream } from "./streamSetup.js";
services/discord-gateway/src/modules/voice-recording/recorder/streamSetup.ts
+15 -4
View File
@@ -54,10 +54,21 @@ export function setupUserStream(input: StreamSetupInput): StreamSetupResult {
cooldownMs: config.DECODER_COOLDOWN_MS,
rotateMs: config.DECODER_ROTATE_MS,
onData: (pcm: Buffer) => {
// Downsample 48kHz stereo -> 24kHz mono (left channel, every 2nd sample)
const outBuf = Buffer.alloc(pcm.length / 4);
for (let i = 0; i < outBuf.length / 2; i++) {
outBuf.writeInt16LE(pcm.readInt16LE(i * 8), i * 2);
// Downsample 48kHz stereo -> 24kHz mono (left channel, every 2nd frame)
// Use typed array views for efficient access instead of read/writeInt16LE
const inputView = new Int16Array(
pcm.buffer,
pcm.byteOffset,
pcm.byteLength / 2,
);
const outBuf = Buffer.alloc(inputView.length / 2); // 48k stereo -> 24k mono = 1/4 size
const outputView = new Int16Array(
outBuf.buffer,
outBuf.byteOffset,
outBuf.byteLength / 2,
);
for (let i = 0; i < outputView.length; i++) {
outputView[i] = inputView[i * 4]; // left channel, every 2nd stereo frame
}
onPcmData(outBuf);
},
services/discord-gateway/src/modules/voice-recording/voiceController.ts
-49
View File
@@ -15,22 +15,6 @@ export interface VoiceStatus {
activeChannelName: string | null;
}
export interface GuildSummary {
id: string;
name: string;
}
export interface VoiceChannelSummary {
id: string;
name: string;
}
export interface ChannelSummary {
id: string;
name: string;
type: string;
}
export class VoiceController {
private activeGuildId: string | null = null;
private activeChannelId: string | null = null;
@@ -54,39 +38,6 @@ export class VoiceController {
};
}
listGuilds(): GuildSummary[] {
logger.info("listGuilds called");
return this.client.guilds.cache
.map((guild) => ({ id: guild.id, name: guild.name }))
.sort((a, b) => a.name.localeCompare(b.name));
}
async listVoiceChannels(guildId: string): Promise<VoiceChannelSummary[]> {
logger.info({ guildId }, "listVoiceChannels called");
const guild = this.getGuild(guildId);
await guild.channels.fetch().catch(() => null);
return guild.channels.cache
.filter((channel) => channel.type === "GUILD_VOICE")
.map((channel) => ({ id: channel.id, name: channel.name }))
.sort((a, b) => a.name.localeCompare(b.name));
}
async listWatchableChannels(guildId: string): Promise<ChannelSummary[]> {
logger.info({ guildId }, "listWatchableChannels called");
const guild = this.getGuild(guildId);
await guild.channels.fetch().catch(() => null);
return guild.channels.cache
.filter((channel) => channel.type === "GUILD_TEXT")
.map((channel) => ({
id: channel.id,
name: channel.name,
type: channel.type,
}))
.sort((a, b) => a.name.localeCompare(b.name));
}
async connect(guildId: string, channelId: string): Promise<VoiceStatus> {
logger.info({ guildId, channelId }, "connect called");
if (!this.client.isReady()) {
services/frontend/src/shared/hooks/useAudioPlayback.ts
+30 -22
View File
@@ -6,11 +6,18 @@ const logger = createLogger("use-audio-playback");
const SAMPLE_RATE = 24000;
const CHANNELS = 1;
const LEVEL_COUNT = 32;
// Pre-computed level distribution shape — computed once at module load, not per render
const LEVEL_SHAPE = Array.from(
{ length: LEVEL_COUNT },
(_, i) => 0.3 + (Math.sin(i * 0.6) * 0.35 + 0.65) * 0.7,
);
export function useAudioPlayback() {
const [isListening, setIsListening] = useState(false);
const [levels, setLevels] = useState<number[]>(
Array.from({ length: 32 }, () => 0.04),
Array.from({ length: LEVEL_COUNT }, () => 0.04),
);
const audioContextRef = useRef<AudioContext | null>(null);
const userTimelinesRef = useRef(new Map<string, number>());
@@ -19,36 +26,37 @@ export function useAudioPlayback() {
(data: { userId: string; pcm: string }) => {
// Decode base64 PCM data
try {
const binaryString = atob(data.pcm);
const bytes = new Uint8Array(binaryString.length);
for (let i = 0; i < binaryString.length; i++) {
bytes[i] = binaryString.charCodeAt(i);
}
const int16Array = new Int16Array(bytes.buffer);
// 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(
bytes.buffer,
bytes.byteOffset,
bytes.byteLength / 2,
);
// Calculate audio levels for visualization
let sum = 0;
for (const sample of int16Array) sum += Math.abs(sample / 32768);
const average = int16Array.length ? sum / int16Array.length : 0;
// 5b/5d: Real RMS calculation + Float32Array conversion in single pass
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);
// Scale RMS to a lively visualization range, clamp to [0.04, 1.0]
const dbLevel = Math.min(1, Math.max(0.04, rms * 8));
// 5c: Use pre-computed LEVEL_SHAPE (no Date.now() per PCM frame)
setLevels((prev) =>
prev.map((_, index) =>
Math.max(
0.04,
average *
(0.5 + Math.sin(index * 0.6 + Date.now() / 140) * 0.35 + 0.65) *
5,
),
Math.max(0.04, dbLevel * LEVEL_SHAPE[index] * 5),
),
);
const audioContext = audioContextRef.current;
if (!isListening || !audioContext) return;
// Convert to float32 for Web Audio API
const float32Array = new Float32Array(int16Array.length);
for (let i = 0; i < int16Array.length; i++)
float32Array[i] = int16Array[i] / 32768;
const audioBuffer = audioContext.createBuffer(
CHANNELS,
float32Array.length,
services/frontend/src/shared/hooks/useAudioTransmit.ts
+31 -12
View File
@@ -1,13 +1,13 @@
// ─── Audio transmit hook — captures mic, encodes to PCM, sends via WebSocket ──
import { useCallback, useRef, useState } from "react";
import { getAPIURL } from "../api/client";
import { createChildLogger } from "../logger";
import { getAPIURL } from "../api/client.js";
import { createChildLogger } from "../logger.js";
const SAMPLE_RATE = 24000;
const logger = createChildLogger("useAudioTransmit");
async function sendTransmitCommand(command: string): Promise<void> {
// Send via HTTP API
// Send via HTTP API (kept as exported function for backward compatibility)
const resp = await fetch(`${getAPIURL()}/api/voice/command`, {
method: "POST",
headers: { "Content-Type": "application/json" },
@@ -24,6 +24,22 @@ async function sendTransmitCommand(command: string): Promise<void> {
}
}
function sendWsCommand(
socketRef: { readonly current: WebSocket | null },
command: string,
): boolean {
if (socketRef.current?.readyState === WebSocket.OPEN) {
socketRef.current.send(
JSON.stringify({
type: "voice_command",
command,
}),
);
return true;
}
return false;
}
export function useAudioTransmit(socketRef: {
readonly current: WebSocket | null;
}) {
@@ -33,7 +49,10 @@ export function useAudioTransmit(socketRef: {
const processorRef = useRef<ScriptProcessorNode | null>(null);
const stop = useCallback(() => {
sendTransmitCommand("voice:transmit:stop").catch(() => {});
// 6c: Prefer WebSocket round-trip over HTTP for lower latency
if (!sendWsCommand(socketRef, "voice:transmit:stop")) {
sendTransmitCommand("voice:transmit:stop").catch(() => {});
}
setIsStreaming(false);
if (processorRef.current) {
@@ -48,10 +67,13 @@ export function useAudioTransmit(socketRef: {
for (const track of streamRef.current.getTracks()) track.stop();
streamRef.current = null;
}
}, []);
}, [socketRef]);
const start = useCallback(async () => {
await sendTransmitCommand("voice:transmit:start");
// 6c: Prefer WebSocket round-trip over HTTP for lower latency
if (!sendWsCommand(socketRef, "voice:transmit:start")) {
await sendTransmitCommand("voice:transmit:start");
}
const stream = await navigator.mediaDevices.getUserMedia({ audio: true });
streamRef.current = stream;
@@ -75,13 +97,10 @@ export function useAudioTransmit(socketRef: {
for (let i = 0; i < inputData.length; i++)
pcmData[i] = Math.max(-1, Math.min(1, inputData[i])) * 32767;
// Base64 encode
// 6b: Replace string-concatenation loop with single call
// 1024 samples → 2048 bytes — well within call-stack limits
const bytes = new Uint8Array(pcmData.buffer);
let binary = "";
for (let i = 0; i < bytes.length; i++) {
binary += String.fromCharCode(bytes[i]);
}
const base64 = btoa(binary);
const base64 = btoa(String.fromCharCode(...bytes));
socketRef.current.send(
JSON.stringify({