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

fix: resolve architecture disconnects and codebase weaknesses

Browse Source 
- 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
This commit is contained in:
MythEclipse
2026-06-09 11:06:14 +07:00
parent 4becf0d6f1
commit f84b380f4f
13 changed files with 104 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/backend/package.json
+1 -1
View File
@@ -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"
}
}
services/backend/src/shared/config/index.ts
+27 -2
View File
@@ -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<typeof configSchema> {
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;
services/discord-gateway/src/modules/ai-moderation/aiAnalyzer.ts
-11
View File
@@ -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) =>
services/discord-gateway/src/modules/command-handler/commandHandler.ts
+4 -5
View File
@@ -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
services/discord-gateway/src/modules/message-capture/messageStore.ts
-1
View File
@@ -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);
services/discord-gateway/src/modules/voice-recording/mediaTypes.ts
+1
View File
@@ -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;
}
services/discord-gateway/src/modules/voice-recording/packetFilter.ts
+3 -3
View File
@@ -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 {
services/discord-gateway/src/modules/voice-recording/player.ts
+2 -1
View File
@@ -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) {
services/discord-gateway/src/modules/voice-recording/recorder.ts
+44 -36
View File
@@ -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);
services/discord-gateway/src/modules/voice-recording/recorder/uploader.ts
+12 -26
View File
@@ -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);
services/frontend/src/shared/ws/events.ts
+3 -1
View File
@@ -5,13 +5,15 @@ export interface WsEventMap {
message_updated: { data: unknown };
message_deleted: { data: { id: string } };
message_analyzed: { data: unknown };
attachment_uploaded: Record<string, never>;
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 };
}
services/frontend/src/shared/ws/socket.ts
+4 -3
View File
@@ -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),