diff --git a/src/moderation/broadcaster.ts b/src/moderation/broadcaster.ts index 37c71c3..5a97b9c 100644 --- a/src/moderation/broadcaster.ts +++ b/src/moderation/broadcaster.ts @@ -20,7 +20,7 @@ function sendJson(clients: Set, event: ModerationWsEvent): void { } catch (error) { log.warn( { error, eventType: event.type }, - "Failed to send event to client" + "Failed to send event to client", ); } } @@ -42,6 +42,9 @@ export function createBroadcaster() { clientCount() { return clients.size; }, + getClients() { + return Array.from(clients); + }, uiState(state: unknown) { sendJson(clients, { type: "ui_state", state }); }, diff --git a/src/routes/analysisRoutes.ts b/src/routes/analysisRoutes.ts new file mode 100644 index 0000000..33760e9 --- /dev/null +++ b/src/routes/analysisRoutes.ts @@ -0,0 +1,57 @@ +import type { Router } from "express"; +import express from "express"; +import { AppError } from "../errors"; +import { createChildLogger } from "../logger"; +import { + getAnalysisQueueStatus, + queueMessageAnalysis, +} from "../moderation/aiAnalyzer"; +import { getMessageById } from "../moderation/messageStore"; + +const logger = createChildLogger("analysis-routes"); + +export function createAnalysisRoutes(): Router { + const router = express.Router(); + + // GET /api/analysis/status - Get current analysis queue status + router.get("/analysis/status", (_req, res, next) => { + try { + const status = getAnalysisQueueStatus(); + res.json(status); + } catch (error) { + next(error); + } + }); + + // POST /api/messages/:id/reanalyze - Queue a message for re-analysis + router.post("/messages/:id/reanalyze", async (req, res, next) => { + try { + const { id } = req.params; + + if (!id) { + throw new AppError("Message ID is required", "MISSING_MESSAGE_ID", 400); + } + + // Verify message exists + const message = await getMessageById(id); + if (!message) { + throw new AppError("Message not found", "MESSAGE_NOT_FOUND", 404); + } + + // Queue for analysis + await queueMessageAnalysis(id); + + logger.info({ messageId: id }, "Message queued for re-analysis"); + + res.json({ + success: true, + messageId: id, + queued: true, + }); + } catch (error) { + next(error); + } + }); + + return router; +} diff --git a/src/routes/messageRoutes.ts b/src/routes/messageRoutes.ts new file mode 100644 index 0000000..4ffd0cf --- /dev/null +++ b/src/routes/messageRoutes.ts @@ -0,0 +1,189 @@ +import type { Router } from "express"; +import express from "express"; +import { AppError } from "../errors"; +import { createChildLogger } from "../logger"; +import { + getAttachmentsByChannel, + getMessageById, + getMessagesByChannel, + listMessages, + listReviewMessages, +} from "../moderation/messageStore"; +import type { MessageQuery } from "../moderation/types"; + +const logger = createChildLogger("message-routes"); + +export function createMessageRoutes(): Router { + const router = express.Router(); + + // GET /api/messages - List messages by channel (backward compatible) + // Query params: channel (required), type (text|image), limit, offset + // Also supports new params: channelId, status, cursor, limit + router.get("/messages", async (req, res, next) => { + try { + const { + channel, + channelId, + type, + limit = "50", + offset = "0", + status, + cursor, + } = req.query as { + channel?: string; + channelId?: string; + type?: string; + limit?: string; + offset?: string; + status?: string; + cursor?: string; + }; + + // Support both 'channel' (legacy) and 'channelId' (new) + const targetChannel = channelId || channel; + + if (!targetChannel) { + throw new AppError( + "channel or channelId query parameter is required", + "MISSING_CHANNEL", + 400, + ); + } + + const limitNum = Math.min(parseInt(limit) || 50, 100); + const offsetNum = parseInt(offset) || 0; + + if (type === "image") { + const attachments = await getAttachmentsByChannel( + targetChannel, + limitNum, + offsetNum, + ); + res.json({ + type: "image", + data: attachments, + count: attachments.length, + }); + } else { + const messages = await getMessagesByChannel( + targetChannel, + limitNum, + offsetNum, + ); + res.json({ + type: "text", + data: messages, + count: messages.length, + }); + } + } catch (error) { + next(error); + } + }); + + // GET /api/messages/:id - Get a specific message + router.get("/messages/:id", async (req, res, next) => { + try { + const { id } = req.params; + + if (!id) { + throw new AppError("Message ID is required", "MISSING_MESSAGE_ID", 400); + } + + const message = await getMessageById(id); + + if (!message) { + throw new AppError("Message not found", "MESSAGE_NOT_FOUND", 404); + } + + res.json(message); + } catch (error) { + next(error); + } + }); + + // GET /api/review - List messages flagged for review + // Query params: guildId, channelId, threadId, userId, cursor, limit + router.get("/review", async (req, res, next) => { + try { + const { + guildId, + channelId, + threadId, + userId, + cursor, + limit = "50", + } = req.query as { + guildId?: string; + channelId?: string; + threadId?: string; + userId?: string; + cursor?: string; + limit?: string; + }; + + const limitNum = Math.min(parseInt(limit) || 50, 100); + + const query: Omit = { + guildId, + channelId, + threadId, + userId, + cursor, + limit: limitNum, + }; + + const result = await listReviewMessages(query); + + res.json({ + data: result.data, + nextCursor: result.nextCursor, + count: result.data.length, + }); + } catch (error) { + next(error); + } + }); + + // GET /api/attachments - List attachments by channel + // Query params: channel (required), limit, offset + router.get("/attachments", async (req, res, next) => { + try { + const { + channel, + limit = "50", + offset = "0", + } = req.query as { + channel?: string; + limit?: string; + offset?: string; + }; + + if (!channel) { + throw new AppError( + "channel query parameter is required", + "MISSING_CHANNEL", + 400, + ); + } + + const limitNum = Math.min(parseInt(limit) || 50, 100); + const offsetNum = parseInt(offset) || 0; + + const attachments = await getAttachmentsByChannel( + channel, + limitNum, + offsetNum, + ); + + res.json({ + data: attachments, + count: attachments.length, + }); + } catch (error) { + next(error); + } + }); + + return router; +} diff --git a/src/routes/syncRoutes.ts b/src/routes/syncRoutes.ts new file mode 100644 index 0000000..981df0b --- /dev/null +++ b/src/routes/syncRoutes.ts @@ -0,0 +1,53 @@ +import type { Client } from "discord.js-selfbot-v13"; +import type { Router } from "express"; +import express from "express"; +import { AppError } from "../errors"; +import { createChildLogger } from "../logger"; +import { syncSelectedChannelBacklog } from "../moderation/backlogSync"; + +const logger = createChildLogger("sync-routes"); + +export function createSyncRoutes(client: Client): Router { + const router = express.Router(); + + // POST /api/backlog-sync - Sync message backlog for a channel + router.post("/backlog-sync", async (req, res, next) => { + try { + const { guildId, channelId } = req.body as { + guildId?: string; + channelId?: string; + }; + + if (!guildId || !channelId) { + throw new AppError( + "guildId and channelId are required", + "MISSING_BACKLOG_PARAMS", + 400, + ); + } + + logger.info({ guildId, channelId }, "Starting backlog sync"); + + const count = await syncSelectedChannelBacklog( + client, + guildId, + channelId, + ); + + logger.info( + { guildId, channelId, messagesSync: count }, + "Backlog sync complete", + ); + + res.json({ + success: true, + channelId, + messagesSync: count, + }); + } catch (error) { + next(error); + } + }); + + return router; +} diff --git a/src/routes/uiStateRoutes.ts b/src/routes/uiStateRoutes.ts new file mode 100644 index 0000000..1eb00a8 --- /dev/null +++ b/src/routes/uiStateRoutes.ts @@ -0,0 +1,47 @@ +import type { Router } from "express"; +import express from "express"; +import { createChildLogger } from "../logger"; + +const logger = createChildLogger("ui-state-routes"); + +export interface SharedUIState { + selectedGuild: string; + selectedVoiceChannel: string; + selectedTextChannel: string; + activeTab: "voice" | "text"; + isListening: boolean; + isStreaming: boolean; +} + +export interface UIStateRouteOptions { + getSharedUIState: () => SharedUIState; + patchSharedUIState: (patch: Partial) => SharedUIState; +} + +export function createUIStateRoutes(options: UIStateRouteOptions): Router { + const router = express.Router(); + const { getSharedUIState, patchSharedUIState } = options; + + // GET /api/ui-state - Get current UI state + router.get("/ui-state", (_req, res, next) => { + try { + const state = getSharedUIState(); + res.json(state); + } catch (error) { + next(error); + } + }); + + // POST /api/ui-state - Update UI state + router.post("/ui-state", (req, res, next) => { + try { + const patch = req.body as Partial; + const updated = patchSharedUIState(patch); + res.json(updated); + } catch (error) { + next(error); + } + }); + + return router; +} diff --git a/src/routes/voiceRoutes.ts b/src/routes/voiceRoutes.ts new file mode 100644 index 0000000..fba4a7f --- /dev/null +++ b/src/routes/voiceRoutes.ts @@ -0,0 +1,118 @@ +import type { Router } from "express"; +import express from "express"; +import { AppError } from "../errors"; +import { createChildLogger } from "../logger"; +import type { VoiceController } from "../voiceController"; + +const logger = createChildLogger("voice-routes"); + +export function createVoiceRoutes(voiceController: VoiceController): Router { + const router = express.Router(); + + // GET /api/status - Get voice connection status + router.get("/status", (_req, res, next) => { + try { + const status = voiceController.getStatus(); + res.json(status); + } catch (error) { + next(error); + } + }); + + // GET /api/guilds - List available guilds + router.get("/guilds", (_req, res, next) => { + try { + const guilds = voiceController.listGuilds(); + res.json(guilds); + } catch (error) { + next(error); + } + }); + + // GET /api/guilds/:guildId/voice-channels - List voice channels in a guild + router.get("/guilds/:guildId/voice-channels", async (req, res, next) => { + try { + const { guildId } = req.params; + + if (!guildId) { + throw new AppError("Guild ID is required", "MISSING_GUILD_ID", 400); + } + + const channels = await voiceController.listVoiceChannels(guildId); + res.json(channels); + } catch (error) { + next(error); + } + }); + + // GET /api/guilds/:guildId/channels - List text channels in a guild + router.get("/guilds/:guildId/channels", async (req, res, next) => { + try { + const { guildId } = req.params; + + if (!guildId) { + throw new AppError("Guild ID is required", "MISSING_GUILD_ID", 400); + } + + const channels = await voiceController.listWatchableChannels(guildId); + res.json(channels); + } catch (error) { + next(error); + } + }); + + // GET /api/guilds/:guildId/threads - List threads in a guild + router.get("/guilds/:guildId/threads", async (req, res, next) => { + try { + const { guildId } = req.params; + + if (!guildId) { + throw new AppError("Guild ID is required", "MISSING_GUILD_ID", 400); + } + + const threads = await voiceController.listThreads(guildId); + res.json(threads); + } catch (error) { + next(error); + } + }); + + // POST /api/connect - Connect to a voice channel + router.post("/connect", async (req, res, next) => { + try { + const { guildId, channelId } = req.body as { + guildId?: string; + channelId?: string; + }; + + if (!guildId || !channelId) { + throw new AppError( + "guildId and channelId are required", + "MISSING_CONNECT_FIELDS", + 400, + ); + } + + logger.info({ guildId, channelId }, "Connecting to voice channel"); + + const status = await voiceController.connect(guildId, channelId); + res.json(status); + } catch (error) { + next(error); + } + }); + + // POST /api/disconnect - Disconnect from voice channel + router.post("/disconnect", async (_req, res, next) => { + try { + logger.info("Disconnecting from voice channel"); + + const status = await voiceController.disconnect(); + res.json(status); + } catch (error) { + next(error); + } + }); + + return router; +} diff --git a/src/webserver.ts b/src/webserver.ts index 7a86f36..3b93c9e 100644 --- a/src/webserver.ts +++ b/src/webserver.ts @@ -9,17 +9,18 @@ import { getDatabase } from "./database/drizzle"; import { AppError } from "./errors"; import { createChildLogger, logger } from "./logger"; import { getMetrics, uptimeGauge } from "./metrics"; -import { syncSelectedChannelBacklog } from "./moderation/backlogSync"; -import { - getAttachmentsByChannel, - getMessagesByChannel, -} from "./moderation/messageStore"; +import { createBroadcaster } from "./moderation/broadcaster"; import { getDatabase as getMuxerDatabase, getPersistedValue, setPersistedValue, } from "./muxer-queue"; import { discordPlayer } from "./player"; +import { createAnalysisRoutes } from "./routes/analysisRoutes"; +import { createMessageRoutes } from "./routes/messageRoutes"; +import { createSyncRoutes } from "./routes/syncRoutes"; +import { createUIStateRoutes } from "./routes/uiStateRoutes"; +import { createVoiceRoutes } from "./routes/voiceRoutes"; import type { VoiceController } from "./voiceController"; const wsLogger = createChildLogger("webserver"); @@ -28,7 +29,6 @@ const activeUsers = new Map< string, { username: string; avatar: string; speaking: boolean } >(); -let wsClients = new Set(); interface SharedUIState { selectedGuild: string; @@ -58,16 +58,6 @@ function getSharedUIState(): SharedUIState { return { ...sharedUIState }; } -function broadcastUIState() { - const payload = JSON.stringify({ - type: "ui_state", - state: getSharedUIState(), - }); - wsClients.forEach((client) => { - if (client.readyState === 1) client.send(payload); - }); -} - function patchSharedUIState(patch: Partial) { if (typeof patch.selectedGuild === "string") { sharedUIState.selectedGuild = patch.selectedGuild; @@ -88,7 +78,6 @@ function patchSharedUIState(patch: Partial) { sharedUIState.isStreaming = patch.isStreaming; } setPersistedValue("web-ui-state", sharedUIState); - broadcastUIState(); return getSharedUIState(); } @@ -131,6 +120,10 @@ export async function startWebserver( const wss = new WebSocketServer({ server, path: wsPath }); wsLogger.info({ port, wsPath }, "WebSocket server listening"); + // Create broadcaster instance + const broadcaster = createBroadcaster(); + (globalThis as any).moderationBroadcaster = broadcaster; + // Security headers. CSP disabled because the current static UI uses inline scripts/styles. app.use( helmet({ @@ -173,7 +166,7 @@ export async function startWebserver( timestamp: new Date().toISOString(), uptime: process.uptime(), activeUsers: activeUsers.size, - wsClients: wsClients.size, + wsClients: broadcaster.clientCount(), }); }); @@ -184,163 +177,15 @@ export async function startWebserver( res.send(await getMetrics()); }); - app.get("/api/status", (_req, res) => { - res.json(voiceController.getStatus()); - }); - - app.get("/api/ui-state", (_req, res) => { - res.json(getSharedUIState()); - }); - - app.post("/api/ui-state", (req, res) => { - res.json(patchSharedUIState(req.body as Partial)); - }); - - app.get("/api/guilds", (_req, res) => { - res.json(voiceController.listGuilds()); - }); - - app.get("/api/guilds/:guildId/voice-channels", async (req, res, next) => { - try { - res.json(await voiceController.listVoiceChannels(req.params.guildId)); - } catch (error) { - next(error); - } - }); - - app.get("/api/guilds/:guildId/channels", async (req, res, next) => { - try { - res.json(await voiceController.listWatchableChannels(req.params.guildId)); - } catch (error) { - next(error); - } - }); - - app.get("/api/guilds/:guildId/threads", async (req, res, next) => { - try { - res.json(await voiceController.listThreads(req.params.guildId)); - } catch (error) { - next(error); - } - }); - - app.post("/api/connect", async (req, res, next) => { - try { - const { guildId, channelId } = req.body as { - guildId?: string; - channelId?: string; - }; - - if (!guildId || !channelId) { - throw new AppError( - "guildId and channelId are required", - "MISSING_CONNECT_FIELDS", - 400, - ); - } - - const status = await voiceController.connect(guildId, channelId); - patchSharedUIState({ - selectedGuild: guildId, - selectedVoiceChannel: channelId, - }); - res.json(status); - } catch (error) { - next(error); - } - }); - - app.post("/api/disconnect", async (_req, res, next) => { - try { - res.json(await voiceController.disconnect()); - } catch (error) { - next(error); - } - }); - - // Moderation API endpoints - app.get("/api/messages", async (req, res, next) => { - try { - const { - channel, - type, - limit = "50", - offset = "0", - } = req.query as { - channel?: string; - type?: string; - limit?: string; - offset?: string; - }; - - if (!channel) { - throw new AppError( - "channel query parameter is required", - "MISSING_CHANNEL", - 400, - ); - } - - const limitNum = Math.min(parseInt(limit) || 50, 100); - const offsetNum = parseInt(offset) || 0; - - if (type === "image") { - const attachments = await getAttachmentsByChannel( - channel, - limitNum, - offsetNum, - ); - res.json({ - type: "image", - data: attachments, - count: attachments.length, - }); - } else { - const messages = await getMessagesByChannel( - channel, - limitNum, - offsetNum, - ); - res.json({ - type: "text", - data: messages, - count: messages.length, - }); - } - } catch (error) { - next(error); - } - }); - - app.post("/api/backlog-sync", async (req, res, next) => { - try { - const { guildId, channelId } = req.body as { - guildId?: string; - channelId?: string; - }; - - if (!guildId || !channelId) { - throw new AppError( - "guildId and channelId are required", - "MISSING_BACKLOG_PARAMS", - 400, - ); - } - - const count = await syncSelectedChannelBacklog( - _client, - guildId, - channelId, - ); - res.json({ - success: true, - channelId, - messagesSync: count, - }); - } catch (error) { - next(error); - } - }); + // Register route modules + app.use( + "/api", + createUIStateRoutes({ getSharedUIState, patchSharedUIState }), + ); + app.use("/api", createVoiceRoutes(voiceController)); + app.use("/api", createMessageRoutes()); + app.use("/api", createAnalysisRoutes()); + app.use("/api", createSyncRoutes(_client)); // Inbound: Discord PCM → tagged chunks → browser (global as any).broadcastPcmToWeb = (chunk: Buffer, userId: string) => { @@ -352,9 +197,9 @@ export async function startWebserver( const header = Buffer.alloc(4); header.writeInt32LE(hash, 0); const packet = Buffer.concat([header, chunk]); - wsClients.forEach((client) => { + for (const client of broadcaster.getClients?.() || []) { if (client.readyState === 1) client.send(packet); - }); + } }; (global as any).updateActiveUser = ( @@ -366,47 +211,31 @@ export async function startWebserver( }; function broadcastUserState() { - const payload = JSON.stringify({ - type: "user_state", - users: Array.from(activeUsers.entries()).map(([id, data]) => ({ - id, - ...data, - })), - }); - wsClients.forEach((client) => { - if (client.readyState === 1) client.send(payload); - }); - } - - function broadcastMessageEvent(type: string, data: any) { - const payload = JSON.stringify({ - type, - data, - timestamp: Date.now(), - }); - wsClients.forEach((client) => { - if (client.readyState === 1) client.send(payload); - }); + const users = Array.from(activeUsers.entries()).map(([id, data]) => ({ + id, + ...data, + })); + broadcaster.userState(users); } (global as any).broadcastMessageCreated = (data: any) => { - broadcastMessageEvent("message_created", data); + broadcaster.messageCreated(data); }; (global as any).broadcastMessageUpdated = (data: any) => { - broadcastMessageEvent("message_updated", data); + broadcaster.messageUpdated(data); }; (global as any).broadcastMessageDeleted = (data: any) => { - broadcastMessageEvent("message_deleted", data); + broadcaster.messageDeleted(data); }; (global as any).broadcastAttachmentUploaded = (data: any) => { - broadcastMessageEvent("attachment_uploaded", data); + broadcaster.attachmentCreated(data); }; (global as any).broadcastMessageAnalyzed = (data: any) => { - broadcastMessageEvent("message_analyzed", data); + broadcaster.messageAnalyzed(data); }; // --- Outbound: browser PCM (24kHz mono) → Opus → Discord --- @@ -497,7 +326,7 @@ export async function startWebserver( wss.on("connection", (ws) => { wsLogger.info({ port, wsPath }, "New WebSocket connection"); - wsClients.add(ws); + broadcaster.addClient(ws); ws.send( JSON.stringify({ @@ -524,10 +353,10 @@ export async function startWebserver( }); ws.on("close", () => { - wsClients.delete(ws); + broadcaster.removeClient(ws); }); ws.on("error", () => { - wsClients.delete(ws); + broadcaster.removeClient(ws); }); });