From 7108f6bb47d084d392952c8c80315afb39c3b494 Mon Sep 17 00:00:00 2001 From: MythEclipse Date: Tue, 9 Jun 2026 16:56:44 +0700 Subject: [PATCH] feat(system): implement graceful shutdown and expand websocket events Improve system reliability and real-time capabilities by implementing a robust lifecycle management system and adding new broadcast events for attachments and voice recordings. - Implement asynchronous graceful shutdown in backend to close HTTP, WebSocket, Redis, and database connections. - Add new WebSocket broadcast events: `attachment_created`, `voice_recording_started`, `voice_recording_stopped`, `voice_recording_uploaded`, and `analysis_queue_status`. - Refactor media status handling to use boolean `playing` state instead of string-based status. - Centralize `PageResult` and `VoiceRecording` types to improve consistency between frontend and backend. - Update frontend API client to include `listRecordings` and handle new WebSocket event types. - Fix type mismatches in voice command handling and media status reporting. --- services/backend/src/index.ts | 53 ++++++++++++++----- .../src/modules/media/media.service.ts | 8 ++- .../modules/messages/messages.repository.ts | 6 +-- .../src/modules/voice/voice.controller.ts | 24 ++++----- services/backend/src/shared/redis/index.ts | 4 +- services/backend/src/ws/broadcast.ts | 20 +++++++ services/backend/src/ws/server.ts | 21 ++++++++ .../modules/command-handler/commandHandler.ts | 32 +++++------ services/frontend/src/App.tsx | 4 ++ .../frontend/src/entities/message/types.ts | 6 +-- .../live/components/RecordingsSubPanel.tsx | 27 ++-------- .../features/live/hooks/useVoiceControl.ts | 3 ++ services/frontend/src/shared/api/client.ts | 38 +++++++++---- services/frontend/src/shared/ws/socket.ts | 5 +- 14 files changed, 161 insertions(+), 90 deletions(-) diff --git a/services/backend/src/index.ts b/services/backend/src/index.ts index 903a0cf..c55aeea 100644 --- a/services/backend/src/index.ts +++ b/services/backend/src/index.ts @@ -1,6 +1,10 @@ import type { Server } from "node:http"; import { createChildLogger } from "@bete/shared/logger"; import { startHttpServer } from "./http/server.js"; +import { closeDatabase } from "./shared/database/index.js"; +import { stopCommandBridge } from "./shared/redis/index.js"; +import { stopRedisBridge as stopEventBridge } from "./ws/redis-bridge.js"; +import { closeWebSocketServer } from "./ws/server.js"; const logger = createChildLogger("backend"); @@ -17,22 +21,43 @@ async function main() { } } -function shutdown(signal: string) { +async function shutdown(signal: string) { logger.info({ signal }, "Shutting down gracefully"); - if (httpServer) { - httpServer.close(() => { - logger.info("HTTP server closed"); - process.exit(0); - }); + try { + // 1. Stop accepting new HTTP connections + if (httpServer) { + await new Promise((resolve) => { + httpServer!.close(() => { + logger.info("HTTP server closed"); + resolve(); + }); + }); + } - // Force exit after 10s if connections don't close - setTimeout(() => { - logger.error("Forced shutdown after timeout"); - process.exit(1); - }, 10_000).unref(); - } else { + // 2. Close WebSocket server + closeWebSocketServer(); + + // 3. Stop Redis bridges (event subscriptions + command channel) + await Promise.allSettled([ + stopEventBridge().catch((err) => + logger.warn({ err }, "Error stopping event bridge"), + ), + stopCommandBridge().catch((err) => + logger.warn({ err }, "Error stopping command bridge"), + ), + ]); + + // 4. Close database pool + await closeDatabase().catch((err) => + logger.warn({ err }, "Error closing database"), + ); + + logger.info("Graceful shutdown completed"); process.exit(0); + } catch (err) { + logger.error({ err }, "Error during graceful shutdown"); + process.exit(1); } } @@ -41,12 +66,12 @@ process.on("SIGTERM", () => shutdown("SIGTERM")); process.on("uncaughtException", (err) => { logger.error({ err }, "Uncaught exception"); - process.exit(1); + shutdown("uncaughtException"); }); process.on("unhandledRejection", (reason) => { logger.error({ reason }, "Unhandled rejection"); - process.exit(1); + shutdown("unhandledRejection"); }); main(); diff --git a/services/backend/src/modules/media/media.service.ts b/services/backend/src/modules/media/media.service.ts index 5984342..371de4d 100644 --- a/services/backend/src/modules/media/media.service.ts +++ b/services/backend/src/modules/media/media.service.ts @@ -47,8 +47,14 @@ export async function getStatus(): Promise { const cached = await readRedisStatus("media:status"); if (cached) { + const rawPlaying = cached.playing; + // Handle both boolean (new) and string (legacy from String(discordPlayer.getStatus())) + const playing = + rawPlaying === true || + rawPlaying === "playing" || + rawPlaying === "buffering"; return { - playing: Boolean(cached.playing), + playing, musicVolume: Number(cached.musicVolume ?? 1.0), current: (cached.current as MediaItem | null) ?? null, queue: (cached.queue as MediaItem[]) ?? [], diff --git a/services/backend/src/modules/messages/messages.repository.ts b/services/backend/src/modules/messages/messages.repository.ts index 0a3bbc7..5a3fe38 100644 --- a/services/backend/src/modules/messages/messages.repository.ts +++ b/services/backend/src/modules/messages/messages.repository.ts @@ -1,3 +1,4 @@ +import type { PageResult } from "@bete/shared"; import { createChildLogger } from "@bete/shared/logger"; import { getPool } from "../../shared/database/index.js"; import type { @@ -8,11 +9,6 @@ import type { const logger = createChildLogger("messages.repository"); -export interface PageResult { - data: T[]; - nextCursor: string | null; -} - export interface AttachmentResult { id: string; message_id: string; diff --git a/services/backend/src/modules/voice/voice.controller.ts b/services/backend/src/modules/voice/voice.controller.ts index 8df2a75..7f4e5d8 100644 --- a/services/backend/src/modules/voice/voice.controller.ts +++ b/services/backend/src/modules/voice/voice.controller.ts @@ -12,13 +12,15 @@ export async function handleGetVoiceStatus(_req: Request, res: Response) { res.json(status); } +/** Safely extract a string value that may be a single string or string array. */ +function asString(val: unknown): string { + if (Array.isArray(val)) return String(val[0] ?? ""); + return String(val ?? ""); +} + export async function handleConnectVoice(req: Request, res: Response) { - const guildId = Array.isArray(req.body.guildId) - ? req.body.guildId[0] - : req.body.guildId; - const channelId = Array.isArray(req.body.channelId) - ? req.body.channelId[0] - : req.body.channelId; + const guildId = asString(req.body.guildId); + const channelId = asString(req.body.channelId); if (!guildId || !channelId) { return res.status(400).json({ error: "VALIDATION_ERROR", @@ -35,17 +37,13 @@ export async function handleDisconnectVoice(_req: Request, res: Response) { } export async function handleGetVoiceChannels(req: Request, res: Response) { - const guildId = Array.isArray(req.params.guildId) - ? req.params.guildId[0] - : req.params.guildId; + const guildId = asString(req.params.guildId); const channels = await getVoiceChannels(guildId); res.json(channels); } export async function handleVoiceCommand(req: Request, res: Response) { - const command = Array.isArray(req.body.command) - ? req.body.command[0] - : req.body.command; + const command = asString(req.body.command); if (!command) { return res.status(400).json({ @@ -55,7 +53,7 @@ export async function handleVoiceCommand(req: Request, res: Response) { } try { - await publishCommandNoReply(command as string); + await publishCommandNoReply(command); res.json({ success: true, command }); } catch (err) { res.status(500).json({ diff --git a/services/backend/src/shared/redis/index.ts b/services/backend/src/shared/redis/index.ts index 6b24c48..c0a705c 100644 --- a/services/backend/src/shared/redis/index.ts +++ b/services/backend/src/shared/redis/index.ts @@ -260,7 +260,7 @@ export async function writeRedisStatus( // Lifecycle // --------------------------------------------------------------------------- -export async function startRedisBridge(): Promise { +export async function startCommandBridge(): Promise { if (!ensureRedisConfig()) { logger.info("Redis not configured, skipping command channel bridge"); return; @@ -272,7 +272,7 @@ export async function startRedisBridge(): Promise { logger.info("Redis command channel initialized"); } -export async function stopRedisBridge(): Promise { +export async function stopCommandBridge(): Promise { if (publisherClient) { await publisherClient.quit(); publisherClient = null; diff --git a/services/backend/src/ws/broadcast.ts b/services/backend/src/ws/broadcast.ts index 3379f13..7b19950 100644 --- a/services/backend/src/ws/broadcast.ts +++ b/services/backend/src/ws/broadcast.ts @@ -18,9 +18,14 @@ export interface BroadcastFunctions { 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; } @@ -53,18 +58,33 @@ export const broadcastMessageUpdated: BroadcastFn = (data) => export const broadcastMessageDeleted: BroadcastFn = (data) => (_fns?.messageDeleted ?? noop)(data); +export const broadcastAttachmentCreated: BroadcastFn = (data) => + (_fns?.attachmentCreated ?? noop)(data); + export const broadcastAttachmentUploaded: BroadcastFn = (data) => (_fns?.attachmentUploaded ?? noop)(data); export const broadcastMessageAnalyzed: BroadcastFn = (data) => (_fns?.messageAnalyzed ?? noop)(data); +export const broadcastVoiceRecordingStarted: BroadcastFn = (data) => + (_fns?.voiceRecordingStarted ?? noop)(data); + +export const broadcastVoiceRecordingStopped: BroadcastFn = (data) => + (_fns?.voiceRecordingStopped ?? noop)(data); + +export const broadcastVoiceRecordingUploaded: BroadcastFn = (data) => + (_fns?.voiceRecordingUploaded ?? noop)(data); + export const broadcastVoicePcmData: BroadcastFn = (data) => (_fns?.voicePcmData ?? noop)(data); export const broadcastVoiceActiveUser: BroadcastFn = (data) => (_fns?.voiceActiveUser ?? noop)(data); +export const broadcastAnalysisQueueStatus: BroadcastFn = (data) => + (_fns?.analysisQueueStatus ?? noop)(data); + export const broadcastRaw: BroadcastRawFn = (type, data) => (_fns?.raw ?? noopRaw)(type, data); diff --git a/services/backend/src/ws/server.ts b/services/backend/src/ws/server.ts index a17587c..c6891b5 100644 --- a/services/backend/src/ws/server.ts +++ b/services/backend/src/ws/server.ts @@ -11,6 +11,9 @@ interface BroadcastEvent { timestamp: string; } +// Track the active WebSocket server for lifecycle management +let _wss: WebSocketServer | null = null; + async function sendInitialStates(ws: WebSocket): Promise { // Send initial user state ws.send( @@ -51,10 +54,18 @@ async function sendInitialStates(ws: WebSocket): Promise { } } +export function closeWebSocketServer(): void { + if (!_wss) return; + logger.info("Closing WebSocket server"); + _wss.close(() => logger.info("WebSocket server closed")); + _wss = null; +} + export function createWebSocketServer(server: Server): WebSocketServer { const clients = new Set(); const wss = new WebSocketServer({ server, path: "/ws" }); + _wss = wss; wss.on("connection", (ws: WebSocket) => { clients.add(ws); @@ -188,12 +199,22 @@ export function createWebSocketServer(server: Server): WebSocketServer { 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, }); diff --git a/services/discord-gateway/src/modules/command-handler/commandHandler.ts b/services/discord-gateway/src/modules/command-handler/commandHandler.ts index 1cc33a3..a8b394c 100644 --- a/services/discord-gateway/src/modules/command-handler/commandHandler.ts +++ b/services/discord-gateway/src/modules/command-handler/commandHandler.ts @@ -35,7 +35,7 @@ interface VoiceStatusPayload { } interface MediaStatusPayload { - playing: string; + playing: boolean; musicVolume: number; current: unknown; queue: unknown[]; @@ -341,25 +341,34 @@ export class CommandHandler { } } - private async handleMediaQueue(_cmd: BackendCommand): Promise { + private getCurrentMediaStatus(): MediaStatusPayload { + return { + playing: discordPlayer.getStatus() === "playing", + musicVolume: discordPlayer.getMusicVolume(), + current: null, + queue: [], + }; + } + + private async handleMediaQueue(cmd: BackendCommand): Promise { // Media queueing is handled at a higher level (frontend / backend streams // audio directly). Log the request for now. logger.info("media:queue received — media queueing is handled externally"); return { - id: _cmd.id, + id: cmd.id, success: true, - data: { note: "media queueing handled externally" }, + data: this.getCurrentMediaStatus(), }; } private async handleMediaSkip(cmd: BackendCommand): Promise { discordPlayer.stop("music"); - return { id: cmd.id, success: true, data: { action: "skipped" } }; + return { id: cmd.id, success: true, data: this.getCurrentMediaStatus() }; } private async handleMediaStop(cmd: BackendCommand): Promise { discordPlayer.stop("music"); - return { id: cmd.id, success: true, data: { action: "stopped" } }; + return { id: cmd.id, success: true, data: this.getCurrentMediaStatus() }; } private async handleMediaVolume(cmd: BackendCommand): Promise { @@ -376,7 +385,7 @@ export class CommandHandler { return { id: cmd.id, success: true, - data: { volume: discordPlayer.getMusicVolume() }, + data: this.getCurrentMediaStatus(), }; } @@ -579,14 +588,7 @@ export class CommandHandler { } private publishMediaStatus(): void { - const status: MediaStatusPayload = { - playing: String(discordPlayer.getStatus()), - musicVolume: discordPlayer.getMusicVolume(), - current: null, - queue: [], - }; - - this.setKey(MEDIA_STATUS_KEY, JSON.stringify(status)); + this.setKey(MEDIA_STATUS_KEY, JSON.stringify(this.getCurrentMediaStatus())); } /** diff --git a/services/frontend/src/App.tsx b/services/frontend/src/App.tsx index 71e0969..df8d949 100644 --- a/services/frontend/src/App.tsx +++ b/services/frontend/src/App.tsx @@ -88,6 +88,10 @@ export default function App() { messages .fetchMessages(monitorGuildId || undefined) .catch(() => undefined), + onAttachmentCreated: () => + messages + .fetchMessages(monitorGuildId || undefined) + .catch(() => undefined), onMediaState: (state) => media.setMediaState(state as MediaState), onVoiceRecordingUploaded: (d) => window.dispatchEvent( diff --git a/services/frontend/src/entities/message/types.ts b/services/frontend/src/entities/message/types.ts index 31e3365..4bfa826 100644 --- a/services/frontend/src/entities/message/types.ts +++ b/services/frontend/src/entities/message/types.ts @@ -3,6 +3,7 @@ export type { AISeverity, AIStatus, MessageRecord, + PageResult, } from "@bete/shared"; export interface MessageMetadata { @@ -26,8 +27,3 @@ export function parseMetadata(value: string | null): MessageMetadata { return {}; } } - -export interface PageResult { - data: T[]; - nextCursor: string | null; -} diff --git a/services/frontend/src/features/live/components/RecordingsSubPanel.tsx b/services/frontend/src/features/live/components/RecordingsSubPanel.tsx index f07a910..a1f6aa4 100644 --- a/services/frontend/src/features/live/components/RecordingsSubPanel.tsx +++ b/services/frontend/src/features/live/components/RecordingsSubPanel.tsx @@ -1,43 +1,24 @@ -// ─── Recordings Sub-Panel — BUG 1 FIX: useEffect instead of useMemo for side effects ── +// ─── Recordings Sub-Panel ── import { Download, Mic } from "lucide-react"; import { useEffect, useState } from "react"; +import type { VoiceRecording } from "../../../shared/api/client"; +import { listRecordings } from "../../../shared/api/client"; import { formatBytes, formatDate } from "../../../shared/lib/utils"; import { Badge, Button, EmptyStateMascot, Skeleton } from "../../../shared/ui"; -interface VoiceRecording { - id: string; - user_id: string; - username: string; - avatar_url: string | null; - guild_id: string | null; - channel_id: string | null; - channel_name: string | null; - filename: string; - size_bytes: number; - download_url: string | null; - upload_status: "pending" | "uploaded" | "failed"; - upload_error: string | null; - created_at: number; - uploaded_at: number | null; -} - export function RecordingsSubPanel() { const [recordings, setRecordings] = useState([]); const [loading, setLoading] = useState(true); const [error, setError] = useState(null); - // BUG 1 FIX: proper useEffect for async data fetching useEffect(() => { let cancelled = false; async function loadRecordings() { try { setLoading(true); setError(null); - const response = await fetch("/api/recordings"); - if (!response.ok) - throw new Error(`Failed to load recordings: ${response.status}`); - const data = (await response.json()) as VoiceRecording[]; + const data = await listRecordings(); if (!cancelled) setRecordings(data); } catch (err) { if (!cancelled) diff --git a/services/frontend/src/features/live/hooks/useVoiceControl.ts b/services/frontend/src/features/live/hooks/useVoiceControl.ts index aeba682..5a76c63 100644 --- a/services/frontend/src/features/live/hooks/useVoiceControl.ts +++ b/services/frontend/src/features/live/hooks/useVoiceControl.ts @@ -15,6 +15,9 @@ export function useVoiceControl() { const [textChannels, setTextChannels] = useState([]); const [voiceStatus, setVoiceStatus] = useState({ connected: false, + activeGuildId: null, + activeChannelId: null, + activeChannelName: null, }); const [loading, setLoading] = useState(false); const [error, setError] = useState(null); diff --git a/services/frontend/src/shared/api/client.ts b/services/frontend/src/shared/api/client.ts index 68b5382..0e19d27 100644 --- a/services/frontend/src/shared/api/client.ts +++ b/services/frontend/src/shared/api/client.ts @@ -1,6 +1,6 @@ // ─── Shared HTTP client — all API endpoints in one file ────────────────────── -import type { MessageRecord } from "@bete/shared"; +import type { MessageRecord, PageResult } from "@bete/shared"; const BE_API_URL = import.meta.env.VITE_BE_API_URL || "http://localhost:3001"; const BE_WS_URL = import.meta.env.VITE_BE_WS_URL || "ws://localhost:3001"; @@ -54,12 +54,7 @@ export function getAPIURL(): string { // ─── Types ─────────────────────────────────────────────────────────────────── -export interface PageResult { - data: T[]; - nextCursor: string | null; -} - -export type { MessageRecord }; +export type { MessageRecord, PageResult }; export interface Guild { id: string; @@ -76,9 +71,9 @@ export interface Channel { export interface VoiceStatus { connected: boolean; - activeGuildId?: string | null; - activeChannelId?: string | null; - activeChannelName?: string | null; + activeGuildId: string | null; + activeChannelId: string | null; + activeChannelName: string | null; } export interface ActiveSpeaker { @@ -237,6 +232,29 @@ export function setMediaVolume(volume: number): Promise { }); } +// ─── Recordings ────────────────────────────────────────────────────────────── + +export interface VoiceRecording { + id: string; + user_id: string; + username: string; + avatar_url: string | null; + guild_id: string | null; + channel_id: string | null; + channel_name: string | null; + filename: string; + size_bytes: number; + download_url: string | null; + upload_status: "pending" | "uploaded" | "failed"; + upload_error: string | null; + created_at: number; + uploaded_at: number | null; +} + +export function listRecordings(limit = 50): Promise { + return request(`/api/recordings?limit=${limit}`); +} + // ─── Auth ──────────────────────────────────────────────────────────────────── export function login(password: string): Promise<{ ok: boolean }> { diff --git a/services/frontend/src/shared/ws/socket.ts b/services/frontend/src/shared/ws/socket.ts index fade1ea..502d7e4 100644 --- a/services/frontend/src/shared/ws/socket.ts +++ b/services/frontend/src/shared/ws/socket.ts @@ -11,6 +11,7 @@ export interface WsHandlers { onMessageUpdated?: (data: unknown) => void; onMessageDeleted?: (data: unknown) => void; onMessageAnalyzed?: (data: unknown) => void; + onAttachmentCreated?: (data: unknown) => void; onAttachmentUploaded?: (data: unknown) => void; onUserState?: (users: unknown[]) => void; onUiState?: (state: unknown) => void; @@ -99,8 +100,7 @@ function doConnect(): WebSocket { h.onVoiceActiveUser?.(msg.data); break; case "attachment_created": - // attachment_created is informational — same data shape as message_created - h.onMessageCreated?.(msg.data); + h.onAttachmentCreated?.(msg.data); break; case "analysis_queue_status": // analysis_queue_status is monitoring-only — no UI action needed @@ -147,6 +147,7 @@ export function useDashboardSocket(handlers: WsHandlers) { onMessageUpdated: (d) => handlersRef.current.onMessageUpdated?.(d), onMessageDeleted: (d) => handlersRef.current.onMessageDeleted?.(d), onMessageAnalyzed: (d) => handlersRef.current.onMessageAnalyzed?.(d), + onAttachmentCreated: (d) => handlersRef.current.onAttachmentCreated?.(d), onAttachmentUploaded: (d) => handlersRef.current.onAttachmentUploaded?.(d), onUserState: (u) => handlersRef.current.onUserState?.(u),