From f84b380f4ff9df3aadd9ec40e0d5a5296e2a123d Mon Sep 17 00:00:00 2001 From: MythEclipse Date: Tue, 9 Jun 2026 11:06:14 +0700 Subject: [PATCH] fix: resolve architecture disconnects and codebase weaknesses - fix(backend): replace raw .parse() with proper loadConfig() + ConfigError - fix(gateway): connect voice recording uploader to EventBroadcaster - fix(gateway): remove dead globalThis.moderationBroadcaster path in AI analyzer - fix(gateway): eliminate audioStream race condition by attaching handlers before pipe - fix(gateway): enable inlineVolume by default for setMusicVolume to work - fix(gateway): reuse persistent redisPub for command replies (no new connection per cmd) - fix(frontend): add missing voice_active_user/voice_pcm_data to WsEventMap - fix(frontend): correct onAttachmentUploaded handler signature to accept data - chore: move @types/pg from dependencies to devDependencies - chore: translate remaining Indonesian comments to English - chore: remove stale P3 TODO comment --- pnpm-lock.yaml | 6 +- services/backend/package.json | 2 +- services/backend/src/shared/config/index.ts | 29 ++++++- .../src/modules/ai-moderation/aiAnalyzer.ts | 11 --- .../modules/command-handler/commandHandler.ts | 9 +-- .../modules/message-capture/messageStore.ts | 1 - .../src/modules/voice-recording/mediaTypes.ts | 1 + .../modules/voice-recording/packetFilter.ts | 6 +- .../src/modules/voice-recording/player.ts | 3 +- .../src/modules/voice-recording/recorder.ts | 80 ++++++++++--------- .../voice-recording/recorder/uploader.ts | 38 +++------ services/frontend/src/shared/ws/events.ts | 4 +- services/frontend/src/shared/ws/socket.ts | 7 +- 13 files changed, 104 insertions(+), 93 deletions(-) diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 72be35a..d26159e 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -52,9 +52,6 @@ importers: '@discordjs/voice': specifier: ^0.19.2 version: 0.19.2(@discordjs/opus@0.10.0)(@emnapi/core@1.10.0)(@emnapi/runtime@1.10.0)(opusscript@0.0.8) - '@types/pg': - specifier: ^8.20.0 - version: 8.20.0 axios: specifier: ^1.16.1 version: 1.16.1 @@ -101,6 +98,9 @@ importers: '@types/node': specifier: ^25.9.0 version: 25.9.0 + '@types/pg': + specifier: ^8.20.0 + version: 8.20.0 '@types/ws': specifier: ^8.18.1 version: 8.18.1 diff --git a/services/backend/package.json b/services/backend/package.json index 76b1d7f..49e00a5 100644 --- a/services/backend/package.json +++ b/services/backend/package.json @@ -16,7 +16,6 @@ "dependencies": { "@bete/shared": "workspace:*", "@discordjs/voice": "^0.19.2", - "@types/pg": "^8.20.0", "axios": "^1.16.1", "dotenv": "^17.4.2", "drizzle-orm": "^0.45.2", @@ -37,6 +36,7 @@ "@types/ws": "^8.18.1", "tsx": "^4.22.2", "typescript": "^5.9.3", + "@types/pg": "^8.20.0", "vitest": "latest" } } diff --git a/services/backend/src/shared/config/index.ts b/services/backend/src/shared/config/index.ts index 11c93c8..3b99da8 100644 --- a/services/backend/src/shared/config/index.ts +++ b/services/backend/src/shared/config/index.ts @@ -1,4 +1,5 @@ import "dotenv/config"; +import { ConfigError } from "@bete/shared/errors"; import { z } from "zod"; const configSchema = z @@ -86,7 +87,31 @@ const configSchema = z .url() .default("https://upload.asepharyana.my.id/api/upload"), }) - .parse(process.env); + .superRefine((value, ctx) => { + if (!value.DATABASE_URL && !value.DATABASE_HOST) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + path: ["DATABASE_URL"], + message: "Either DATABASE_URL or DATABASE_HOST must be provided", + }); + } + }); -export const config = configSchema; +export function loadConfig( + env: NodeJS.ProcessEnv = process.env, +): z.infer { + try { + return configSchema.parse(env); + } catch (error) { + if (error instanceof z.ZodError) { + const messages = error.issues + .map((e) => `${e.path.join(".")}: ${e.message}`) + .join("\n"); + throw new ConfigError(`Configuration validation failed:\n${messages}`); + } + throw error; + } +} + +export const config = loadConfig(); export type Config = typeof config; diff --git a/services/discord-gateway/src/modules/ai-moderation/aiAnalyzer.ts b/services/discord-gateway/src/modules/ai-moderation/aiAnalyzer.ts index fa69b34..1f647b9 100644 --- a/services/discord-gateway/src/modules/ai-moderation/aiAnalyzer.ts +++ b/services/discord-gateway/src/modules/ai-moderation/aiAnalyzer.ts @@ -22,7 +22,6 @@ import type { AnalysisQueueStatus, AnalysisResult, MessageRecord, - ModerationBroadcaster, } from "../message-capture/types.js"; import { attemptAutoDeleteFlaggedMessage } from "./autoDeleteManager.js"; import { estimateTokens } from "./conversationContext.js"; @@ -30,22 +29,12 @@ import { logModerationError } from "./responseLogger.js"; const logger = createChildLogger("ai-analyzer"); -type ModerationGlobal = typeof globalThis & { - moderationBroadcaster?: ModerationBroadcaster; -}; - -function getModerationBroadcaster(): ModerationBroadcaster | undefined { - return (globalThis as ModerationGlobal).moderationBroadcaster; -} - // Redis EventBroadcaster — set by startPendingAIAnalysisWorker. // Used to publish analysis completion events so the backend // redis-bridge can forward them to frontend WebSocket clients. let _redisEventBroadcaster: EventBroadcaster | undefined; function broadcastAnalysisCompleted(row: MessageRecord): void { - // In-memory WS broadcast (direct-connected DG clients) - getModerationBroadcaster()?.messageAnalyzed(row); // Redis pub/sub broadcast → backend → frontend WebSocket if (_redisEventBroadcaster) { _redisEventBroadcaster.messageAnalyzed(row).catch((err: unknown) => diff --git a/services/discord-gateway/src/modules/command-handler/commandHandler.ts b/services/discord-gateway/src/modules/command-handler/commandHandler.ts index df2fc7f..44fdf05 100644 --- a/services/discord-gateway/src/modules/command-handler/commandHandler.ts +++ b/services/discord-gateway/src/modules/command-handler/commandHandler.ts @@ -178,12 +178,11 @@ export class CommandHandler { }; } - // Publish reply on the designated reply channel. - const redisPub = new Redis(config.REDIS_URL); + // Publish reply on the designated reply channel using the persistent publisher. try { - await redisPub.publish(cmd.replyChannel, JSON.stringify(reply)); - } finally { - await redisPub.quit(); + await this.redisPub.publish(cmd.replyChannel, JSON.stringify(reply)); + } catch (err) { + logger.error({ err }, "Failed to publish command reply"); } // Always refresh status keys after every command so the backend has diff --git a/services/discord-gateway/src/modules/message-capture/messageStore.ts b/services/discord-gateway/src/modules/message-capture/messageStore.ts index b843e7a..619e114 100644 --- a/services/discord-gateway/src/modules/message-capture/messageStore.ts +++ b/services/discord-gateway/src/modules/message-capture/messageStore.ts @@ -257,7 +257,6 @@ export async function getMessagesByChannel( .select() .from(messagesTable) .where(and(...conditions)) - // P3: add secondary sort by id for stable pagination .orderBy(desc(messagesTable.created_at), desc(messagesTable.id)) .limit(limit) .offset(offset); diff --git a/services/discord-gateway/src/modules/voice-recording/mediaTypes.ts b/services/discord-gateway/src/modules/voice-recording/mediaTypes.ts index 0694522..22b63ed 100644 --- a/services/discord-gateway/src/modules/voice-recording/mediaTypes.ts +++ b/services/discord-gateway/src/modules/voice-recording/mediaTypes.ts @@ -60,6 +60,7 @@ export type DiscordPlayerOwner = "none" | "browser-bridge" | "music" | "screen"; export interface DiscordPlayOptions { inputType?: StreamType; + /** Enable volume control via resource.volume (required for setMusicVolume to work). */ inlineVolume?: boolean; volume?: number; } diff --git a/services/discord-gateway/src/modules/voice-recording/packetFilter.ts b/services/discord-gateway/src/modules/voice-recording/packetFilter.ts index 566fc95..59d2902 100644 --- a/services/discord-gateway/src/modules/voice-recording/packetFilter.ts +++ b/services/discord-gateway/src/modules/voice-recording/packetFilter.ts @@ -1,8 +1,8 @@ import { Transform, TransformCallback } from "node:stream"; /** - * Transform stream untuk memfilter audio packets yang terlalu kecil - * Packet yang terlalu kecil kemungkinan gagal didekripsi oleh Discord + * Transform stream to filter out audio packets that are too small. + * Packets that are too small are likely to fail decryption by Discord. */ export class PacketFilter extends Transform { private minPacketSize: number; @@ -21,7 +21,7 @@ export class PacketFilter extends Transform { ): void { this.totalCount++; - // Filter packet yang terlalu kecil + // Filter out undersized packets if (chunk.length >= this.minPacketSize) { this.push(chunk); } else { diff --git a/services/discord-gateway/src/modules/voice-recording/player.ts b/services/discord-gateway/src/modules/voice-recording/player.ts index a33dd43..58602dc 100644 --- a/services/discord-gateway/src/modules/voice-recording/player.ts +++ b/services/discord-gateway/src/modules/voice-recording/player.ts @@ -59,7 +59,8 @@ export class DiscordPlayer { const resource = createAudioResource(stream, { inputType: options.inputType ?? StreamType.OggOpus, - inlineVolume: options.inlineVolume ?? false, + // Default to true so setMusicVolume/setResourceVolume works + inlineVolume: options.inlineVolume ?? true, }); if (this.owner === owner) { diff --git a/services/discord-gateway/src/modules/voice-recording/recorder.ts b/services/discord-gateway/src/modules/voice-recording/recorder.ts index 0f7a5e9..8909b3f 100644 --- a/services/discord-gateway/src/modules/voice-recording/recorder.ts +++ b/services/discord-gateway/src/modules/voice-recording/recorder.ts @@ -1,4 +1,4 @@ -import fs, { promises as fsPromises } from "node:fs"; +import { promises as fsPromises } from "node:fs"; import path from "node:path"; import { createChildLogger } from "@bete/shared/logger"; import { retryWithBackoff } from "@bete/shared/utils"; @@ -32,13 +32,16 @@ const logger = createChildLogger("recorder"); let _eventBroadcaster: EventBroadcaster | undefined; +/** @internal Export for uploader.ts to broadcast voice_recording_uploaded events */ +export { _eventBroadcaster }; + export function setEventBroadcaster(broadcaster: EventBroadcaster | undefined) { _eventBroadcaster = broadcaster; } const recordingsDir = config.RECORDINGS_DIR; -// Pastikan folder recordings ada +// Ensure recordings directory exists (async () => { try { await fsPromises.mkdir(recordingsDir, { recursive: true }); @@ -94,7 +97,7 @@ export async function startRecording( logger.error({ error: err }, "Voice connection error"); }); - // Tunggu sampai benar-benar terhubung dengan retry logic + // Wait until fully connected with retry logic try { await retryWithBackoff( () => @@ -148,7 +151,7 @@ export async function startRecording( speaking: true, }); - // Jangan record kalau sudah ada stream aktif untuk user ini + // Skip if user already has an active stream if (receiver.subscriptions.has(userId)) return; const userDir = path.join(recordingsDir, userId); @@ -157,17 +160,19 @@ export async function startRecording( }); try { - // --- OGG file recording with segment rotation --- - const packetFilterForOgg = new PacketFilter( - config.PACKET_FILTER_MIN_SIZE, - ); + // Subscribe to the audio stream FIRST, then immediately attach all event + // handlers before piping — prevents race condition where initial packets + // arrive before listeners are registered. const audioStream = receiver.subscribe(userId, { end: { behavior: EndBehaviorType.AfterSilence, duration: config.AUDIO_STREAM_SILENCE_DURATION_MS, }, }); - const oggPacketStream = audioStream.pipe(packetFilterForOgg); + + const packetFilterForOgg = new PacketFilter( + config.PACKET_FILTER_MIN_SIZE, + ); const segmentManager = new SegmentManager( userDir, config.RECORDING_SEGMENT_MS, @@ -187,6 +192,33 @@ export async function startRecording( }, }); + // Attach all audioStream event handlers BEFORE pipe() to avoid data loss + audioStream.on("data", (chunk: Buffer) => { + if (chunk.length < 8) return; + segmentManager.rotateIfNeeded(packetFilterForOgg); + decoder.rotateIfNeeded(); + decoder.write(chunk); + }); + + audioStream.on("end", () => { + segmentManager.close(packetFilterForOgg); + decoder.destroy(); + _eventBroadcaster?.voiceActiveUser(userId, { + username: userMetadata.username, + avatar: userMetadata.avatarUrl, + speaking: false, + }); + }); + + audioStream.on("error", (error: Error) => { + segmentManager.close(packetFilterForOgg); + decoder.destroy(); + logger.error({ userId, error: error.message }, "Audio stream error"); + }); + + // Now pipe for OGG recording (safe — event handlers already attached) + const oggPacketStream = audioStream.pipe(packetFilterForOgg); + const activeSession = activeSessions.get(channel.guild.id); let currentSegment = segmentManager.open(oggPacketStream); currentSegment.out.on("finish", () => { @@ -256,30 +288,6 @@ export async function startRecording( logger.error({ userId, error: msg }, "File write 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); - decoder.rotateIfNeeded(); - decoder.write(chunk); - }); - - audioStream.on("end", () => { - segmentManager.close(oggPacketStream); - decoder.destroy(); - _eventBroadcaster?.voiceActiveUser(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) => { segmentManager.close(oggPacketStream); logger.error({ userId, error: err.message }, "PacketFilter error"); @@ -292,7 +300,7 @@ export async function startRecording( } }); - // Handle disconnect yang tidak disengaja + // Handle unexpected disconnection connection.on(VoiceConnectionStatus.Disconnected, async () => { if (config.VERBOSE) { logger.warn("Disconnected from voice channel. Reconnecting..."); @@ -310,7 +318,7 @@ export async function startRecording( config.RECONNECT_TIMEOUT_MS, ), ]); - // Berhasil reconnect + // Reconnected successfully } catch { logger.error("Could not reconnect. Destroying connection"); connection.destroy(); @@ -328,7 +336,7 @@ export async function startRecording( } /** - * Hentikan recording dan disconnect dari voice channel. + * Stop recording and disconnect from voice channel. */ export function stopRecording(guildId: string): void { const connection = getVoiceConnection(guildId); diff --git a/services/discord-gateway/src/modules/voice-recording/recorder/uploader.ts b/services/discord-gateway/src/modules/voice-recording/recorder/uploader.ts index 84c1578..c16d7c3 100644 --- a/services/discord-gateway/src/modules/voice-recording/recorder/uploader.ts +++ b/services/discord-gateway/src/modules/voice-recording/recorder/uploader.ts @@ -68,12 +68,11 @@ export async function uploadRecordingSegment(input: { await updateVoiceRecordingAsUploaded(id, downloadUrl, Date.now()); logger.info({ id, downloadUrl }, "Recording segment uploaded successfully"); - // 4. Broadcast via WebSocket if broadcaster exists globally - const broadcaster = (globalThis as any).moderationBroadcaster; - if (broadcaster) { - const payload = JSON.stringify({ - type: "voice_recording_uploaded", - data: { + // 4. Broadcast via Redis EventBroadcaster (forwarded to WebSocket clients by backend) + const { _eventBroadcaster } = await import("../recorder.js"); + if (_eventBroadcaster) { + _eventBroadcaster + .voiceRecordingUploaded({ id, user_id: userId, username, @@ -87,26 +86,13 @@ export async function uploadRecordingSegment(input: { upload_status: "uploaded", created_at: Date.now(), uploaded_at: Date.now(), - }, - timestamp: Date.now(), - }); - - broadcaster - .getClients() - .forEach( - (client: { readyState: number; send: (data: string) => void }) => { - if (client.readyState === 1) { - try { - client.send(payload); - } catch (err) { - logger.warn( - { err }, - "Failed to send recording upload event to client", - ); - } - } - }, - ); + }) + .catch((err: unknown) => { + logger.warn( + { err }, + "Failed to broadcast voice recording upload event", + ); + }); } } catch (error) { const errorMsg = error instanceof Error ? error.message : String(error); diff --git a/services/frontend/src/shared/ws/events.ts b/services/frontend/src/shared/ws/events.ts index a5dec4d..04547e6 100644 --- a/services/frontend/src/shared/ws/events.ts +++ b/services/frontend/src/shared/ws/events.ts @@ -5,13 +5,15 @@ export interface WsEventMap { message_updated: { data: unknown }; message_deleted: { data: { id: string } }; message_analyzed: { data: unknown }; - attachment_uploaded: Record; + attachment_uploaded: { data: unknown }; user_state: { users: unknown[] }; ui_state: { state: unknown }; media_state: { state: unknown }; voice_recording_uploaded: { data: unknown }; voice_recording_started: { data: unknown }; voice_recording_stopped: { data: unknown }; + voice_pcm_data: { data: unknown }; + voice_active_user: { data: unknown }; attachment_created: { data: unknown }; analysis_queue_status: { data: unknown }; } diff --git a/services/frontend/src/shared/ws/socket.ts b/services/frontend/src/shared/ws/socket.ts index c06488c..f1bff69 100644 --- a/services/frontend/src/shared/ws/socket.ts +++ b/services/frontend/src/shared/ws/socket.ts @@ -11,7 +11,7 @@ export interface WsHandlers { onMessageUpdated?: (data: unknown) => void; onMessageDeleted?: (data: unknown) => void; onMessageAnalyzed?: (data: unknown) => void; - onAttachmentUploaded?: () => void; + onAttachmentUploaded?: (data: unknown) => void; onUserState?: (users: unknown[]) => void; onUiState?: (state: unknown) => void; onMediaState?: (state: unknown) => void; @@ -72,7 +72,7 @@ function doConnect(): WebSocket { h.onMessageAnalyzed?.(msg.data); break; case "attachment_uploaded": - h.onAttachmentUploaded?.(); + h.onAttachmentUploaded?.(msg.data); break; case "user_state": h.onUserState?.((msg.users as unknown[]) || []); @@ -147,7 +147,8 @@ export function useDashboardSocket(handlers: WsHandlers) { onMessageUpdated: (d) => handlersRef.current.onMessageUpdated?.(d), onMessageDeleted: (d) => handlersRef.current.onMessageDeleted?.(d), onMessageAnalyzed: (d) => handlersRef.current.onMessageAnalyzed?.(d), - onAttachmentUploaded: () => handlersRef.current.onAttachmentUploaded?.(), + onAttachmentUploaded: (d) => + handlersRef.current.onAttachmentUploaded?.(d), onUserState: (u) => handlersRef.current.onUserState?.(u), onUiState: (s) => handlersRef.current.onUiState?.(s), onMediaState: (s) => handlersRef.current.onMediaState?.(s),