refactor: comprehensive codebase cleanup and architecture hardening
- Sprint 1 (Quick Wins): Remove dead analytics modules, fix 4 unresolved imports, replace 3 console.warn with logger, remove mock-crc import - Sprint 2 (Architecture): Create MascotChatRepository, AnalysisRepository, 3 Zod schemas (mascot-chat, analysis, voice), deduplicate error classes, move 3 SQL queries from routes to repository - Sprint 3 (Complexity): Replace 7 any types with proper interfaces, extract 6 helpers from prepareMediaMessage (CC 85 -> ~15) - Sprint 4 (Config): Remove 22 dead env vars from .env, add 30 missing vars to .env.example, standardize naming Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
MythEclipse
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
d0d9e1669e
commit
4becf0d6f1
@@ -1,3 +1,4 @@
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
import express, {
|
||||
type Express,
|
||||
type NextFunction,
|
||||
@@ -6,7 +7,6 @@ import express, {
|
||||
} from "express";
|
||||
import helmet from "helmet";
|
||||
import { createAnalysisRouter } from "../modules/analysis/analysis.routes.js";
|
||||
import { createAnalyticsRouter } from "../modules/analytics/analytics.routes.js";
|
||||
import { createAuthRouter } from "../modules/auth/auth.routes.js";
|
||||
import { createConfigRouter } from "../modules/config/config.routes.js";
|
||||
import { createHealthRouter } from "../modules/health/health.routes.js";
|
||||
@@ -15,9 +15,8 @@ import { createMediaRouter } from "../modules/media/media.routes.js";
|
||||
import { createMessagesRouter } from "../modules/messages/messages.routes.js";
|
||||
import { createRecordingsRouter } from "../modules/recordings/recordings.routes.js";
|
||||
import { createUiStateRouter } from "../modules/ui-state/ui-state.routes.js";
|
||||
import { createVoiceRouter } from "../modules/voice/voice.routes.js";
|
||||
import { createGuildsRouter } from "../modules/voice/guilds.routes.js";
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
import { createVoiceRouter } from "../modules/voice/voice.routes.js";
|
||||
import { errorHandler } from "../shared/middlewares/index.js";
|
||||
|
||||
const logger = createChildLogger("http.app");
|
||||
@@ -66,7 +65,6 @@ export function createHttpApp(): Express {
|
||||
app.use("/api", createConfigRouter());
|
||||
app.use("/api", createMessagesRouter());
|
||||
app.use("/api", createAnalysisRouter());
|
||||
app.use("/api", createAnalyticsRouter());
|
||||
app.use("/api", createMascotChatRouter());
|
||||
app.use("/api", createMediaRouter());
|
||||
app.use("/api", createVoiceRouter());
|
||||
|
||||
@@ -1,10 +1,10 @@
|
||||
import { createServer, type Server } from "node:http";
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
import { config } from "../shared/config/index.js";
|
||||
import { initializeDatabase } from "../shared/database/index.js";
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
import { createHttpApp } from "./app.js";
|
||||
import { createWebSocketServer } from "../ws/server.js";
|
||||
import { startRedisBridge } from "../ws/redis-bridge.js";
|
||||
import { createWebSocketServer } from "../ws/server.js";
|
||||
import { createHttpApp } from "./app.js";
|
||||
|
||||
const logger = createChildLogger("http.server");
|
||||
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import { startHttpServer } from "./http/server.js";
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
import type { Server } from "node:http";
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
import { startHttpServer } from "./http/server.js";
|
||||
|
||||
const logger = createChildLogger("backend");
|
||||
|
||||
|
||||
@@ -0,0 +1,114 @@
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
import { getPool } from "../../shared/database/index.js";
|
||||
|
||||
const logger = createChildLogger("analysis.repository");
|
||||
|
||||
export interface AnalysisSearchQuery {
|
||||
q?: string;
|
||||
channelId?: string;
|
||||
guildId?: string;
|
||||
limit?: number;
|
||||
}
|
||||
|
||||
export interface AnalysisSearchResult {
|
||||
id: string;
|
||||
guild_id: string;
|
||||
channel_id: string;
|
||||
thread_id: string | null;
|
||||
user_id: string;
|
||||
username: string;
|
||||
avatar_url: string | null;
|
||||
content: string;
|
||||
edited_content: string | null;
|
||||
created_at: number;
|
||||
edited_at: number | null;
|
||||
deleted_at: number | null;
|
||||
type: string;
|
||||
metadata: string | null;
|
||||
ai_status: string | null;
|
||||
ai_moderation_flags: string | null;
|
||||
ai_moderation_score: number | null;
|
||||
ai_analysis: string | null;
|
||||
ai_categories: string | null;
|
||||
ai_severity: string | null;
|
||||
ai_confidence: number | null;
|
||||
ai_recommended_action: string | null;
|
||||
ai_analyzed_at: number | null;
|
||||
ai_error: string | null;
|
||||
}
|
||||
|
||||
function mapSearchResult(row: Record<string, unknown>): AnalysisSearchResult {
|
||||
return {
|
||||
id: String(row.id ?? ""),
|
||||
guild_id: String(row.guild_id ?? ""),
|
||||
channel_id: String(row.channel_id ?? ""),
|
||||
thread_id: (row.thread_id as string | null) ?? null,
|
||||
user_id: String(row.user_id ?? ""),
|
||||
username: String(row.username ?? ""),
|
||||
avatar_url: (row.avatar_url as string | null) ?? null,
|
||||
content: String(row.content ?? ""),
|
||||
edited_content: (row.edited_content as string | null) ?? null,
|
||||
created_at: Number(row.created_at ?? 0),
|
||||
edited_at: (row.edited_at as number | null) ?? null,
|
||||
deleted_at: (row.deleted_at as number | null) ?? null,
|
||||
type: String(row.type ?? "text"),
|
||||
metadata: (row.metadata as string | null) ?? null,
|
||||
ai_status: (row.ai_status as string | null) ?? null,
|
||||
ai_moderation_flags: (row.ai_moderation_flags as string | null) ?? null,
|
||||
ai_moderation_score: (row.ai_moderation_score as number | null) ?? null,
|
||||
ai_analysis: (row.ai_analysis as string | null) ?? null,
|
||||
ai_categories: (row.ai_categories as string | null) ?? null,
|
||||
ai_severity: (row.ai_severity as string | null) ?? null,
|
||||
ai_confidence: (row.ai_confidence as number | null) ?? null,
|
||||
ai_recommended_action: (row.ai_recommended_action as string | null) ?? null,
|
||||
ai_analyzed_at: (row.ai_analyzed_at as number | null) ?? null,
|
||||
ai_error: (row.ai_error as string | null) ?? null,
|
||||
};
|
||||
}
|
||||
|
||||
export class AnalysisRepository {
|
||||
async search(query: AnalysisSearchQuery): Promise<AnalysisSearchResult[]> {
|
||||
const pool = getPool();
|
||||
const { q = "", channelId, guildId, limit = 20 } = query;
|
||||
|
||||
logger.debug({ q, channelId, guildId, limit }, "Searching analysis");
|
||||
|
||||
const searchPattern = `%${q}%`;
|
||||
const clauses: string[] = ["content ILIKE $1"];
|
||||
const params: (string | number)[] = [searchPattern];
|
||||
let p = 2;
|
||||
|
||||
if (guildId) {
|
||||
clauses.push(`guild_id = $${p}`);
|
||||
params.push(guildId);
|
||||
p++;
|
||||
}
|
||||
|
||||
if (channelId) {
|
||||
clauses.push(`channel_id = $${p}`);
|
||||
params.push(channelId);
|
||||
p++;
|
||||
}
|
||||
|
||||
const where = clauses.join(" AND ");
|
||||
const { rows } = await pool.query(
|
||||
`SELECT
|
||||
id, guild_id, channel_id, thread_id,
|
||||
user_id, username, avatar_url,
|
||||
content, edited_content, created_at, edited_at, deleted_at,
|
||||
type, metadata,
|
||||
ai_status, ai_moderation_flags, ai_moderation_score,
|
||||
ai_analysis, ai_categories, ai_severity, ai_confidence,
|
||||
ai_recommended_action, ai_analyzed_at, ai_error
|
||||
FROM messages
|
||||
WHERE ${where}
|
||||
ORDER BY created_at DESC
|
||||
LIMIT $${p}`,
|
||||
[...params, limit],
|
||||
);
|
||||
|
||||
return rows.map((r) => mapSearchResult(r as Record<string, unknown>));
|
||||
}
|
||||
}
|
||||
|
||||
export const analysisRepository = new AnalysisRepository();
|
||||
@@ -1,6 +1,6 @@
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
import type { Request, Response, Router } from "express";
|
||||
import express from "express";
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
import { asyncHandler } from "../../shared/middlewares/index.js";
|
||||
import { analysisService } from "./analysis.service.js";
|
||||
|
||||
|
||||
@@ -0,0 +1,10 @@
|
||||
import { z } from "zod";
|
||||
|
||||
export const searchQuerySchema = z.object({
|
||||
q: z.string().default(""),
|
||||
channelId: z.string().optional(),
|
||||
guildId: z.string().optional(),
|
||||
limit: z.coerce.number().int().positive().max(100).default(20),
|
||||
});
|
||||
|
||||
export type SearchQuery = z.infer<typeof searchQuerySchema>;
|
||||
@@ -1,78 +1,26 @@
|
||||
import { sql } from "drizzle-orm";
|
||||
import { config } from "../../shared/config/index.js";
|
||||
import { getDatabase } from "../../shared/database/index.js";
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
import { config } from "../../shared/config/index.js";
|
||||
import type { AnalysisSearchQuery } from "./analysis.repository.js";
|
||||
import { analysisRepository } from "./analysis.repository.js";
|
||||
|
||||
const logger = createChildLogger("analysis.service");
|
||||
|
||||
export interface AnalysisSearchQuery {
|
||||
q?: string;
|
||||
channelId?: string;
|
||||
limit?: number;
|
||||
}
|
||||
|
||||
/** Full message columns for search results — matches MessageRecord from client.ts */
|
||||
const FULL_COLUMNS = sql.raw(`
|
||||
id, guild_id, channel_id, thread_id,
|
||||
user_id, username, avatar_url,
|
||||
content, edited_content, created_at, edited_at, deleted_at,
|
||||
type, metadata,
|
||||
ai_status, ai_moderation_flags, ai_moderation_score,
|
||||
ai_analysis, ai_categories, ai_severity, ai_confidence,
|
||||
ai_recommended_action, ai_analyzed_at, ai_error
|
||||
`);
|
||||
export type { AnalysisSearchQuery };
|
||||
|
||||
export class AnalysisService {
|
||||
async search(query: AnalysisSearchQuery) {
|
||||
const db = getDatabase();
|
||||
const { q = "", channelId, limit = 20 } = query;
|
||||
const guildId = config.MONITOR_GUILD_ID;
|
||||
|
||||
logger.debug({ q, channelId, limit, guildId }, "Searching analysis");
|
||||
|
||||
const searchPattern = `%${q}%`;
|
||||
const limitVal = limit;
|
||||
const rows = await analysisRepository.search({
|
||||
q,
|
||||
channelId,
|
||||
guildId,
|
||||
limit,
|
||||
});
|
||||
|
||||
let sqlQuery;
|
||||
if (channelId && guildId) {
|
||||
sqlQuery = sql`
|
||||
SELECT ${FULL_COLUMNS}
|
||||
FROM messages
|
||||
WHERE guild_id = ${guildId}
|
||||
AND channel_id = ${channelId}
|
||||
AND content ILIKE ${searchPattern}
|
||||
ORDER BY created_at DESC
|
||||
LIMIT ${limitVal}
|
||||
`;
|
||||
} else if (channelId) {
|
||||
sqlQuery = sql`
|
||||
SELECT ${FULL_COLUMNS}
|
||||
FROM messages
|
||||
WHERE channel_id = ${channelId}
|
||||
AND content ILIKE ${searchPattern}
|
||||
ORDER BY created_at DESC
|
||||
LIMIT ${limitVal}
|
||||
`;
|
||||
} else if (guildId) {
|
||||
sqlQuery = sql`
|
||||
SELECT ${FULL_COLUMNS}
|
||||
FROM messages
|
||||
WHERE guild_id = ${guildId}
|
||||
AND content ILIKE ${searchPattern}
|
||||
ORDER BY created_at DESC
|
||||
LIMIT ${limitVal}
|
||||
`;
|
||||
} else {
|
||||
sqlQuery = sql`
|
||||
SELECT ${FULL_COLUMNS}
|
||||
FROM messages
|
||||
WHERE content ILIKE ${searchPattern}
|
||||
ORDER BY created_at DESC
|
||||
LIMIT ${limitVal}
|
||||
`;
|
||||
}
|
||||
|
||||
const { rows } = await db.execute(sqlQuery);
|
||||
return { results: rows };
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,197 +0,0 @@
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
import type { NextFunction, Request, Response } from "express";
|
||||
import { asyncHandler, requireParam } from "../../shared/middlewares/index.js";
|
||||
import { analyticsQuerySchema } from "./analytics.schema.js";
|
||||
import { analyticsService } from "./analytics.service.js";
|
||||
|
||||
const logger = createChildLogger("analytics.controller");
|
||||
|
||||
export function handleGetOverview(
|
||||
req: Request,
|
||||
res: Response,
|
||||
next: NextFunction,
|
||||
) {
|
||||
return asyncHandler(async (req: Request, res: Response) => {
|
||||
const query = analyticsQuerySchema.parse(req.query);
|
||||
logger.debug({ query }, "Handling get overview");
|
||||
const result = await analyticsService.getOverview(query);
|
||||
res.json(result);
|
||||
})(req, res, next);
|
||||
}
|
||||
|
||||
export function handleGetDailyTrend(
|
||||
req: Request,
|
||||
res: Response,
|
||||
next: NextFunction,
|
||||
) {
|
||||
return asyncHandler(async (req: Request, res: Response) => {
|
||||
const guildId = requireParam(
|
||||
req.query.guildId,
|
||||
"query parameter",
|
||||
"guildId",
|
||||
);
|
||||
const hours = req.query.hours ? Number(req.query.hours) : 24;
|
||||
logger.debug({ guildId, hours }, "Handling get daily trend");
|
||||
const result = await analyticsService.getDailyTrend(guildId, hours);
|
||||
res.json(result);
|
||||
})(req, res, next);
|
||||
}
|
||||
|
||||
export function handleGetHourlyStats(
|
||||
req: Request,
|
||||
res: Response,
|
||||
next: NextFunction,
|
||||
) {
|
||||
return asyncHandler(async (req: Request, res: Response) => {
|
||||
const query = analyticsQuerySchema.parse(req.query);
|
||||
logger.debug({ query }, "Handling get hourly stats");
|
||||
const result = await analyticsService.getHourlyStats(
|
||||
query.guildId,
|
||||
query.channelId,
|
||||
query.hours,
|
||||
);
|
||||
res.json(result);
|
||||
})(req, res, next);
|
||||
}
|
||||
|
||||
export function handleGetTopViolators(
|
||||
req: Request,
|
||||
res: Response,
|
||||
next: NextFunction,
|
||||
) {
|
||||
return asyncHandler(async (req: Request, res: Response) => {
|
||||
const query = analyticsQuerySchema.parse(req.query);
|
||||
const limit = req.query.limit ? Number(req.query.limit) : 10;
|
||||
logger.debug({ query, limit }, "Handling get top violators");
|
||||
const result = await analyticsService.getTopViolators(
|
||||
query.guildId,
|
||||
query.channelId,
|
||||
query.hours,
|
||||
limit,
|
||||
);
|
||||
res.json(result);
|
||||
})(req, res, next);
|
||||
}
|
||||
|
||||
export function handleGetUserLeaderboard(
|
||||
req: Request,
|
||||
res: Response,
|
||||
next: NextFunction,
|
||||
) {
|
||||
return asyncHandler(async (req: Request, res: Response) => {
|
||||
const query = analyticsQuerySchema.parse(req.query);
|
||||
const limit = req.query.limit ? Number(req.query.limit) : 10;
|
||||
logger.debug({ query, limit }, "Handling get user leaderboard");
|
||||
const result = await analyticsService.getUserLeaderboard(
|
||||
query.guildId,
|
||||
query.channelId,
|
||||
query.hours,
|
||||
limit,
|
||||
);
|
||||
res.json(result);
|
||||
})(req, res, next);
|
||||
}
|
||||
|
||||
export function handleGetModerationStats(
|
||||
req: Request,
|
||||
res: Response,
|
||||
next: NextFunction,
|
||||
) {
|
||||
return asyncHandler(async (req: Request, res: Response) => {
|
||||
const query = analyticsQuerySchema.parse(req.query);
|
||||
logger.debug({ query }, "Handling get moderation stats");
|
||||
const result = await analyticsService.getModerationStats(
|
||||
query.guildId,
|
||||
query.channelId,
|
||||
query.hours,
|
||||
);
|
||||
res.json(result);
|
||||
})(req, res, next);
|
||||
}
|
||||
|
||||
export function handleGetHeatmap(
|
||||
req: Request,
|
||||
res: Response,
|
||||
next: NextFunction,
|
||||
) {
|
||||
return asyncHandler(async (req: Request, res: Response) => {
|
||||
const query = analyticsQuerySchema.parse(req.query);
|
||||
logger.debug({ query }, "Handling get heatmap");
|
||||
const result = await analyticsService.getHeatmap(
|
||||
query.guildId,
|
||||
query.channelId,
|
||||
query.hours,
|
||||
);
|
||||
res.json(result);
|
||||
})(req, res, next);
|
||||
}
|
||||
|
||||
export function handleGetTopics(
|
||||
req: Request,
|
||||
res: Response,
|
||||
next: NextFunction,
|
||||
) {
|
||||
return asyncHandler(async (req: Request, res: Response) => {
|
||||
const query = analyticsQuerySchema.parse(req.query);
|
||||
logger.debug({ query }, "Handling get topics");
|
||||
const result = await analyticsService.getTopics(
|
||||
query.guildId,
|
||||
query.channelId,
|
||||
query.hours,
|
||||
);
|
||||
res.json(result);
|
||||
})(req, res, next);
|
||||
}
|
||||
|
||||
export function handleGetModerationActions(
|
||||
req: Request,
|
||||
res: Response,
|
||||
next: NextFunction,
|
||||
) {
|
||||
return asyncHandler(async (req: Request, res: Response) => {
|
||||
const query = analyticsQuerySchema.parse(req.query);
|
||||
const limit = req.query.limit ? Number(req.query.limit) : 20;
|
||||
logger.debug({ query, limit }, "Handling get moderation actions");
|
||||
const result = await analyticsService.getModerationActions(
|
||||
query.guildId,
|
||||
query.channelId,
|
||||
query.hours,
|
||||
limit,
|
||||
);
|
||||
res.json(result);
|
||||
})(req, res, next);
|
||||
}
|
||||
|
||||
export function handleGetAIStats(
|
||||
req: Request,
|
||||
res: Response,
|
||||
next: NextFunction,
|
||||
) {
|
||||
return asyncHandler(async (req: Request, res: Response) => {
|
||||
const query = analyticsQuerySchema.parse(req.query);
|
||||
logger.debug({ query }, "Handling get AI stats");
|
||||
const result = await analyticsService.getAIStats(
|
||||
query.guildId,
|
||||
query.channelId,
|
||||
query.hours,
|
||||
);
|
||||
res.json(result);
|
||||
})(req, res, next);
|
||||
}
|
||||
|
||||
export function handleGetAttachmentStats(
|
||||
req: Request,
|
||||
res: Response,
|
||||
next: NextFunction,
|
||||
) {
|
||||
return asyncHandler(async (req: Request, res: Response) => {
|
||||
const query = analyticsQuerySchema.parse(req.query);
|
||||
logger.debug({ query }, "Handling get attachment stats");
|
||||
const result = await analyticsService.getAttachmentStats(
|
||||
query.guildId,
|
||||
query.channelId,
|
||||
query.hours,
|
||||
);
|
||||
res.json(result);
|
||||
})(req, res, next);
|
||||
}
|
||||
@@ -1,551 +0,0 @@
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
import { getPool } from "../../shared/database/index.js";
|
||||
|
||||
const logger = createChildLogger("analytics.repository");
|
||||
|
||||
interface TimeFilter {
|
||||
where: string;
|
||||
params: Array<string | number>;
|
||||
paramOffset: number;
|
||||
}
|
||||
|
||||
function buildTimeFilter(
|
||||
guildId: string,
|
||||
channelId: string | undefined,
|
||||
hours: number,
|
||||
offset = 1,
|
||||
): TimeFilter {
|
||||
const clauses: string[] = ["guild_id = $" + offset];
|
||||
const params: Array<string | number> = [guildId];
|
||||
let p = offset + 1;
|
||||
|
||||
if (channelId) {
|
||||
clauses.push("channel_id = $" + p);
|
||||
params.push(channelId);
|
||||
p++;
|
||||
}
|
||||
|
||||
clauses.push("created_at > (EXTRACT(EPOCH FROM NOW()) * 1000 - $" + p + ")");
|
||||
params.push(hours * 3_600_000);
|
||||
|
||||
return { where: "WHERE " + clauses.join(" AND "), params, paramOffset: p };
|
||||
}
|
||||
|
||||
export class AnalyticsRepository {
|
||||
async getOverview(guildId: string, channelId?: string, hours = 24) {
|
||||
logger.debug({ guildId, channelId, hours }, "Getting analytics overview");
|
||||
const pool = getPool();
|
||||
const now = Date.now();
|
||||
const start = now - hours * 3_600_000;
|
||||
const filter = buildTimeFilter(guildId, channelId, hours);
|
||||
|
||||
const { rows } = await pool.query(
|
||||
`
|
||||
SELECT
|
||||
COUNT(*)::int AS total_messages,
|
||||
COUNT(DISTINCT user_id)::int AS active_users_count,
|
||||
COUNT(DISTINCT channel_id)::int AS total_channels,
|
||||
COUNT(*) FILTER (WHERE ai_status = 'clean')::int AS clean,
|
||||
COUNT(*) FILTER (WHERE ai_status = 'warn')::int AS warned,
|
||||
COUNT(*) FILTER (WHERE ai_status = 'flagged')::int AS flagged,
|
||||
COUNT(*) FILTER (WHERE ai_status = 'error')::int AS error,
|
||||
COUNT(*) FILTER (WHERE ai_status = 'pending')::int AS pending,
|
||||
COALESCE(AVG(ai_moderation_score), 0)::real AS average_score
|
||||
FROM messages
|
||||
${filter.where}
|
||||
`,
|
||||
filter.params,
|
||||
);
|
||||
|
||||
const row = rows[0] as Record<string, unknown> | undefined;
|
||||
|
||||
// Fetch hourly stats, topics, and top violators to include in overview
|
||||
const hourly = await this.getHourlyStats(guildId, channelId, hours);
|
||||
const topics = await this.getTopics(guildId, channelId, hours);
|
||||
const topUsers = await this.getTopViolators(guildId, channelId, hours, 5);
|
||||
|
||||
return {
|
||||
period: { start, end: now },
|
||||
messages: {
|
||||
total: Number(row?.total_messages ?? 0),
|
||||
clean: Number(row?.clean ?? 0),
|
||||
warned: Number(row?.warned ?? 0),
|
||||
flagged: Number(row?.flagged ?? 0),
|
||||
error: Number(row?.error ?? 0),
|
||||
pending: Number(row?.pending ?? 0),
|
||||
average_score: Number(row?.average_score ?? 0),
|
||||
},
|
||||
hourly,
|
||||
topics,
|
||||
top_users: topUsers,
|
||||
active_users_count: Number(row?.active_users_count ?? 0),
|
||||
total_channels: Number(row?.total_channels ?? 0),
|
||||
};
|
||||
}
|
||||
|
||||
async getDailyTrend(guildId: string, hours = 24) {
|
||||
logger.debug({ guildId, hours }, "Getting daily trend");
|
||||
const pool = getPool();
|
||||
const filter = buildTimeFilter(guildId, undefined, hours);
|
||||
|
||||
const { rows } = await pool.query(
|
||||
`
|
||||
SELECT
|
||||
TO_CHAR(to_timestamp(created_at / 1000), 'YYYY-MM-DD') AS date,
|
||||
COUNT(*)::int AS count,
|
||||
COUNT(*) FILTER (WHERE ai_status = 'clean')::int AS clean,
|
||||
COUNT(*) FILTER (WHERE ai_status = 'warn')::int AS warned,
|
||||
COUNT(*) FILTER (WHERE ai_status = 'flagged')::int AS flagged,
|
||||
COUNT(*) FILTER (WHERE ai_status = 'error')::int AS error
|
||||
FROM messages
|
||||
${filter.where}
|
||||
GROUP BY date
|
||||
ORDER BY date ASC
|
||||
`,
|
||||
filter.params,
|
||||
);
|
||||
|
||||
return rows.map((r) => ({
|
||||
date: r.date as string,
|
||||
count: Number(r.count ?? 0),
|
||||
clean: Number(r.clean ?? 0),
|
||||
warned: Number(r.warned ?? 0),
|
||||
flagged: Number(r.flagged ?? 0),
|
||||
error: Number(r.error ?? 0),
|
||||
}));
|
||||
}
|
||||
|
||||
async getHourlyStats(guildId: string, channelId?: string, hours = 24) {
|
||||
logger.debug({ guildId, channelId, hours }, "Getting hourly stats");
|
||||
const pool = getPool();
|
||||
const filter = buildTimeFilter(guildId, channelId, hours);
|
||||
|
||||
const { rows } = await pool.query(
|
||||
`
|
||||
SELECT
|
||||
TO_CHAR(to_timestamp(created_at / 1000), 'HH24') AS hour,
|
||||
COUNT(*)::int AS count,
|
||||
COUNT(*) FILTER (WHERE ai_status = 'clean')::int AS clean,
|
||||
COUNT(*) FILTER (WHERE ai_status = 'warn')::int AS warned,
|
||||
COUNT(*) FILTER (WHERE ai_status = 'flagged')::int AS flagged,
|
||||
COUNT(*) FILTER (WHERE ai_status = 'error')::int AS error
|
||||
FROM messages
|
||||
${filter.where}
|
||||
GROUP BY hour
|
||||
ORDER BY hour ASC
|
||||
`,
|
||||
filter.params,
|
||||
);
|
||||
|
||||
return rows.map((r) => ({
|
||||
hour: r.hour as string,
|
||||
count: Number(r.count ?? 0),
|
||||
clean: Number(r.clean ?? 0),
|
||||
warned: Number(r.warned ?? 0),
|
||||
flagged: Number(r.flagged ?? 0),
|
||||
error: Number(r.error ?? 0),
|
||||
}));
|
||||
}
|
||||
|
||||
async getTopViolators(
|
||||
guildId: string,
|
||||
channelId?: string,
|
||||
hours = 24,
|
||||
limit = 10,
|
||||
) {
|
||||
logger.debug({ guildId, channelId, hours, limit }, "Getting top violators");
|
||||
const pool = getPool();
|
||||
const filter = buildTimeFilter(guildId, channelId, hours);
|
||||
|
||||
const { rows } = await pool.query(
|
||||
`
|
||||
SELECT
|
||||
user_id,
|
||||
MAX(username) AS username,
|
||||
MAX(avatar_url) AS avatar_url,
|
||||
COUNT(*)::int AS total_messages,
|
||||
COUNT(*) FILTER (WHERE ai_status IN ('warn', 'flagged', 'error'))::int AS flagged_count,
|
||||
COUNT(*) FILTER (WHERE ai_status = 'warn')::int AS warned_count,
|
||||
COUNT(*) FILTER (WHERE ai_status = 'flagged')::int AS hard_flagged_count,
|
||||
COUNT(*) FILTER (WHERE ai_status = 'error')::int AS error_count,
|
||||
COALESCE(AVG(ai_moderation_score), 0)::real AS violation_score,
|
||||
MAX(ai_moderation_flags) AS worst_flags,
|
||||
MAX(created_at) AS last_violation
|
||||
FROM messages
|
||||
${filter.where}
|
||||
AND ai_status IN ('warn', 'flagged', 'error')
|
||||
GROUP BY user_id
|
||||
ORDER BY flagged_count DESC
|
||||
LIMIT $${filter.params.length + 1}
|
||||
`,
|
||||
[...filter.params, limit],
|
||||
);
|
||||
|
||||
return rows.map((r) => ({
|
||||
user_id: r.user_id as string,
|
||||
username: (r.username as string) ?? "",
|
||||
avatar_url: (r.avatar_url as string | null) ?? null,
|
||||
total_messages: Number(r.total_messages ?? 0),
|
||||
flagged_count: Number(r.flagged_count ?? 0),
|
||||
warned_count: Number(r.warned_count ?? 0),
|
||||
violation_score: Number(r.violation_score ?? 0),
|
||||
worst_flags: (r.worst_flags as string | null)
|
||||
? (r.worst_flags as string)
|
||||
.split(",")
|
||||
.map((s) => s.trim())
|
||||
.filter(Boolean)
|
||||
: [],
|
||||
last_violation: Number(r.last_violation ?? 0),
|
||||
}));
|
||||
}
|
||||
|
||||
async getUserLeaderboard(
|
||||
guildId: string,
|
||||
channelId?: string,
|
||||
hours = 24,
|
||||
limit = 10,
|
||||
) {
|
||||
logger.debug(
|
||||
{ guildId, channelId, hours, limit },
|
||||
"Getting user leaderboard",
|
||||
);
|
||||
const pool = getPool();
|
||||
const filter = buildTimeFilter(guildId, channelId, hours);
|
||||
|
||||
const { rows } = await pool.query(
|
||||
`
|
||||
SELECT
|
||||
user_id,
|
||||
MAX(username) AS username,
|
||||
MAX(avatar_url) AS avatar_url,
|
||||
COUNT(*)::int AS message_count,
|
||||
COUNT(*) FILTER (WHERE type = 'edited')::int AS edited_count,
|
||||
COUNT(*) FILTER (WHERE type = 'deleted')::int AS deleted_count,
|
||||
COUNT(*) FILTER (WHERE ai_status IN ('warn', 'flagged', 'error'))::int AS flagged_count,
|
||||
MAX(created_at) AS last_active
|
||||
FROM messages
|
||||
${filter.where}
|
||||
GROUP BY user_id
|
||||
ORDER BY message_count DESC
|
||||
LIMIT $${filter.params.length + 1}
|
||||
`,
|
||||
[...filter.params, limit],
|
||||
);
|
||||
|
||||
return rows.map((r) => ({
|
||||
user_id: r.user_id as string,
|
||||
username: (r.username as string) ?? "",
|
||||
avatar_url: (r.avatar_url as string | null) ?? null,
|
||||
message_count: Number(r.message_count ?? 0),
|
||||
edited_count: Number(r.edited_count ?? 0),
|
||||
deleted_count: Number(r.deleted_count ?? 0),
|
||||
flagged_count: Number(r.flagged_count ?? 0),
|
||||
last_active: Number(r.last_active ?? 0),
|
||||
}));
|
||||
}
|
||||
|
||||
async getModerationStats(guildId: string, channelId?: string, hours = 24) {
|
||||
logger.debug({ guildId, channelId, hours }, "Getting moderation stats");
|
||||
const pool = getPool();
|
||||
const filter = buildTimeFilter(guildId, channelId, hours);
|
||||
|
||||
const { rows } = await pool.query(
|
||||
`
|
||||
SELECT
|
||||
COUNT(*)::int AS total,
|
||||
COUNT(*) FILTER (WHERE ai_status = 'clean')::int AS clean,
|
||||
COUNT(*) FILTER (WHERE ai_status = 'warn')::int AS warned,
|
||||
COUNT(*) FILTER (WHERE ai_status = 'flagged')::int AS flagged,
|
||||
COUNT(*) FILTER (WHERE ai_status = 'error')::int AS error,
|
||||
COUNT(*) FILTER (WHERE ai_status = 'pending')::int AS pending,
|
||||
COALESCE(AVG(ai_moderation_score), 0)::real AS average_score
|
||||
FROM messages
|
||||
${filter.where}
|
||||
`,
|
||||
filter.params,
|
||||
);
|
||||
|
||||
const row = rows[0] as Record<string, unknown> | undefined;
|
||||
|
||||
return {
|
||||
total: Number(row?.total ?? 0),
|
||||
clean: Number(row?.clean ?? 0),
|
||||
warned: Number(row?.warned ?? 0),
|
||||
flagged: Number(row?.flagged ?? 0),
|
||||
error: Number(row?.error ?? 0),
|
||||
pending: Number(row?.pending ?? 0),
|
||||
average_score: Number(row?.average_score ?? 0),
|
||||
};
|
||||
}
|
||||
|
||||
async getHeatmap(guildId: string, channelId?: string, hours = 24) {
|
||||
logger.debug({ guildId, channelId, hours }, "Getting heatmap data");
|
||||
const pool = getPool();
|
||||
const filter = buildTimeFilter(guildId, channelId, hours);
|
||||
|
||||
const { rows } = await pool.query(
|
||||
`
|
||||
SELECT
|
||||
EXTRACT(DOW FROM to_timestamp(created_at / 1000))::int AS day_of_week,
|
||||
EXTRACT(HOUR FROM to_timestamp(created_at / 1000))::int AS hour,
|
||||
COUNT(*)::int AS count,
|
||||
COUNT(*) FILTER (WHERE ai_status = 'clean')::int AS clean,
|
||||
COUNT(*) FILTER (WHERE ai_status = 'warn')::int AS warned,
|
||||
COUNT(*) FILTER (WHERE ai_status = 'flagged')::int AS flagged
|
||||
FROM messages
|
||||
${filter.where}
|
||||
GROUP BY day_of_week, hour
|
||||
ORDER BY day_of_week, hour
|
||||
`,
|
||||
filter.params,
|
||||
);
|
||||
|
||||
return rows.map((r) => ({
|
||||
dayOfWeek: Number(r.day_of_week ?? 0),
|
||||
hour: Number(r.hour ?? 0),
|
||||
count: Number(r.count ?? 0),
|
||||
clean: Number(r.clean ?? 0),
|
||||
warned: Number(r.warned ?? 0),
|
||||
flagged: Number(r.flagged ?? 0),
|
||||
}));
|
||||
}
|
||||
|
||||
async getTopics(guildId: string, channelId?: string, hours = 24) {
|
||||
logger.debug({ guildId, channelId, hours }, "Getting topics");
|
||||
const pool = getPool();
|
||||
const filter = buildTimeFilter(guildId, channelId, hours);
|
||||
|
||||
try {
|
||||
const { rows } = await pool.query(
|
||||
`
|
||||
WITH word_list AS (
|
||||
SELECT
|
||||
LOWER(TRIM(BOTH '.,!?;:\"()[]{}' FROM word)) AS word,
|
||||
ai_moderation_score
|
||||
FROM messages,
|
||||
LATERAL UNNEST(STRING_TO_ARRAY(content, ' ')) AS word
|
||||
${filter.where}
|
||||
AND content IS NOT NULL AND content != ''
|
||||
AND LENGTH(TRIM(BOTH '.,!?;:\"()[]{}' FROM word)) >= 4
|
||||
AND word !~ '^<.+:\d+>$'
|
||||
AND word !~ '^\['
|
||||
AND word !~ '^https?://'
|
||||
AND word !~ '^discord\.'
|
||||
AND word !~ '^cdn\.'
|
||||
)
|
||||
SELECT
|
||||
word AS topic,
|
||||
COUNT(*)::int AS count,
|
||||
COALESCE(AVG(ai_moderation_score), 0)::real AS score
|
||||
FROM word_list
|
||||
WHERE word NOT IN (
|
||||
'yang','dan','di','ke','dari','dengan','untuk','pada','ini','itu',
|
||||
'ada','akan','telah','sudah','bisa','dapat','tidak','nggak','enggak',
|
||||
'gak','gk','ga','aku','saya','kamu','dia','kami','kita','mereka',
|
||||
'iya','ya','yah','oh','ah','eh','lah','pun','juga','masih',
|
||||
'saja','hanya','sama','atau','tapi','namun','sedang','sangat',
|
||||
'begitu','karena','sebab','kalau','jika','maka','lalu','setelah',
|
||||
'seperti','antara','oleh','sebagai','secara','melalui','dalam',
|
||||
'the','and','for','are','but','not','you','all','can','has',
|
||||
'was','were','been','like','just','that','this','with','your',
|
||||
'from','they','have','what','when','where','which','their',
|
||||
'about','would','could','should','very','also','than','then',
|
||||
'mau','lagi','jadi','aja','nya','apa','orang',
|
||||
'lihat','kak','bro','bang','mas','pack','sih','dong','kok',
|
||||
'nih','deh','kali','loh','lho','doang',
|
||||
'gue','lo','lu','gua','elo','ane','wkwk','wkwkwk',
|
||||
'wkwkwkwk','wkwkwkwkwk','haha','hahaha','hehe','wk','wkwk',
|
||||
'kalo','buat','udah','jir','kan','tuh','pake','dulu',
|
||||
'banget','kayak','kya','kyk','klo','karna','soalnya',
|
||||
'bikin','bilang','makan','minum','tidur','main','pergi','pulang',
|
||||
'mana','cuma','kah','udh','gitu','gini','gtu','gni',
|
||||
'mending','wkakak','wakak'
|
||||
)
|
||||
AND word NOT LIKE '%.jpg'
|
||||
AND word NOT LIKE '%.jpeg'
|
||||
AND word NOT LIKE '%.png'
|
||||
AND word NOT LIKE '%.gif'
|
||||
AND word NOT LIKE '%.webp'
|
||||
AND word NOT LIKE '%.mp4'
|
||||
AND word NOT LIKE '%.mp3'
|
||||
AND word NOT LIKE '%.pdf'
|
||||
AND word NOT LIKE '%.zip'
|
||||
GROUP BY word
|
||||
ORDER BY COUNT(*) DESC
|
||||
LIMIT 10
|
||||
`,
|
||||
filter.params,
|
||||
);
|
||||
|
||||
return rows.map((r) => ({
|
||||
topic: (r.topic as string) ?? "",
|
||||
count: Number(r.count ?? 0),
|
||||
score: Number(r.score ?? 0),
|
||||
}));
|
||||
} catch (error) {
|
||||
logger.error(
|
||||
{ error: error instanceof Error ? error.message : String(error), guildId },
|
||||
"getTopics query failed — returning empty",
|
||||
);
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
// ── New endpoints ─────────────────────────────────────────────────────────
|
||||
|
||||
async getModerationActions(
|
||||
guildId: string,
|
||||
channelId?: string,
|
||||
hours = 24,
|
||||
limit = 20,
|
||||
) {
|
||||
logger.debug(
|
||||
{ guildId, channelId, hours, limit },
|
||||
"Getting moderation actions",
|
||||
);
|
||||
const pool = getPool();
|
||||
const filter = buildTimeFilter(guildId, channelId, hours);
|
||||
|
||||
const { rows } = await pool.query(
|
||||
`
|
||||
SELECT
|
||||
ma.id,
|
||||
ma.message_id,
|
||||
ma.user_id,
|
||||
ma.guild_id,
|
||||
ma.action_type,
|
||||
ma.reason,
|
||||
ma.executed_by,
|
||||
ma.status,
|
||||
ma.error,
|
||||
ma.created_at,
|
||||
ma.executed_at,
|
||||
m.username,
|
||||
m.content
|
||||
FROM moderation_actions ma
|
||||
LEFT JOIN messages m ON m.id = ma.message_id
|
||||
${filter.where.replace("guild_id", "ma.guild_id").replace("created_at", "ma.created_at")}
|
||||
ORDER BY ma.created_at DESC
|
||||
LIMIT $${filter.params.length + 1}
|
||||
`,
|
||||
[...filter.params, limit],
|
||||
);
|
||||
|
||||
return rows.map((r) => ({
|
||||
id: r.id as string,
|
||||
message_id: (r.message_id as string) ?? null,
|
||||
user_id: r.user_id as string,
|
||||
guild_id: r.guild_id as string,
|
||||
action_type: r.action_type as string,
|
||||
reason: (r.reason as string) ?? null,
|
||||
executed_by: (r.executed_by as string) ?? null,
|
||||
status: r.status as string,
|
||||
error: (r.error as string) ?? null,
|
||||
created_at: Number(r.created_at ?? 0),
|
||||
executed_at: r.executed_at ? Number(r.executed_at) : null,
|
||||
username: (r.username as string) ?? "Unknown",
|
||||
content: (r.content as string) ?? null,
|
||||
}));
|
||||
}
|
||||
|
||||
async getAIStats(guildId: string, channelId?: string, hours = 24) {
|
||||
logger.debug({ guildId, channelId, hours }, "Getting AI stats");
|
||||
const pool = getPool();
|
||||
const filter = buildTimeFilter(guildId, channelId, hours);
|
||||
|
||||
const { rows } = await pool.query(
|
||||
`
|
||||
SELECT
|
||||
COUNT(*)::int AS total_analyzed,
|
||||
COUNT(*) FILTER (WHERE ai_severity = 'none')::int AS severity_none,
|
||||
COUNT(*) FILTER (WHERE ai_severity = 'low')::int AS severity_low,
|
||||
COUNT(*) FILTER (WHERE ai_severity = 'medium')::int AS severity_medium,
|
||||
COUNT(*) FILTER (WHERE ai_severity = 'high')::int AS severity_high,
|
||||
COUNT(*) FILTER (WHERE ai_severity = 'critical')::int AS severity_critical,
|
||||
COUNT(*) FILTER (WHERE ai_recommended_action = 'none')::int AS action_none,
|
||||
COUNT(*) FILTER (WHERE ai_recommended_action = 'monitor')::int AS action_monitor,
|
||||
COUNT(*) FILTER (WHERE ai_recommended_action = 'warn')::int AS action_warn,
|
||||
COUNT(*) FILTER (WHERE ai_recommended_action = 'review')::int AS action_review,
|
||||
COUNT(*) FILTER (WHERE ai_recommended_action = 'delete')::int AS action_delete,
|
||||
COUNT(*) FILTER (WHERE ai_recommended_action = 'escalate')::int AS action_escalate,
|
||||
COUNT(*) FILTER (WHERE ai_status = 'error')::int AS analysis_errors,
|
||||
COUNT(*) FILTER (WHERE ai_status = 'pending')::int AS analysis_pending,
|
||||
COALESCE(AVG(ai_confidence), 0)::real AS avg_confidence,
|
||||
COALESCE(AVG(ai_moderation_score), 0)::real AS avg_score
|
||||
FROM messages
|
||||
${filter.where}
|
||||
`,
|
||||
filter.params,
|
||||
);
|
||||
|
||||
const row = rows[0] as Record<string, unknown> | undefined;
|
||||
|
||||
return {
|
||||
total_analyzed: Number(row?.total_analyzed ?? 0),
|
||||
severity: {
|
||||
none: Number(row?.severity_none ?? 0),
|
||||
low: Number(row?.severity_low ?? 0),
|
||||
medium: Number(row?.severity_medium ?? 0),
|
||||
high: Number(row?.severity_high ?? 0),
|
||||
critical: Number(row?.severity_critical ?? 0),
|
||||
},
|
||||
recommended_actions: {
|
||||
none: Number(row?.action_none ?? 0),
|
||||
monitor: Number(row?.action_monitor ?? 0),
|
||||
warn: Number(row?.action_warn ?? 0),
|
||||
review: Number(row?.action_review ?? 0),
|
||||
delete: Number(row?.action_delete ?? 0),
|
||||
escalate: Number(row?.action_escalate ?? 0),
|
||||
},
|
||||
analysis_errors: Number(row?.analysis_errors ?? 0),
|
||||
analysis_pending: Number(row?.analysis_pending ?? 0),
|
||||
avg_confidence: Number(row?.avg_confidence ?? 0),
|
||||
avg_score: Number(row?.avg_score ?? 0),
|
||||
};
|
||||
}
|
||||
|
||||
async getAttachmentStats(guildId: string, channelId?: string, hours = 24) {
|
||||
logger.debug({ guildId, channelId, hours }, "Getting attachment stats");
|
||||
const pool = getPool();
|
||||
const filter = buildTimeFilter(guildId, channelId, hours);
|
||||
|
||||
const { rows } = await pool.query(
|
||||
`
|
||||
SELECT
|
||||
COUNT(*)::int AS total_attachments,
|
||||
COUNT(*) FILTER (WHERE a.upload_status = 'uploaded')::int AS uploaded,
|
||||
COUNT(*) FILTER (WHERE a.upload_status = 'pending')::int AS pending,
|
||||
COUNT(*) FILTER (WHERE a.upload_status = 'failed')::int AS failed,
|
||||
COALESCE(SUM(a.size), 0)::bigint AS total_size_bytes,
|
||||
COUNT(DISTINCT a.user_id)::int AS unique_uploaders,
|
||||
(SELECT a2.type FROM attachments a2
|
||||
WHERE a2.guild_id = $1
|
||||
${channelId ? "AND a2.channel_id = $" + (filter.paramOffset - 1) : ""}
|
||||
AND a2.created_at > (EXTRACT(EPOCH FROM NOW()) * 1000 - $${filter.paramOffset})
|
||||
GROUP BY a2.type ORDER BY COUNT(*) DESC LIMIT 1
|
||||
) AS top_mime_type
|
||||
FROM attachments a
|
||||
WHERE a.guild_id = $1
|
||||
${channelId ? "AND a.channel_id = $" + (filter.paramOffset - 1) : ""}
|
||||
AND a.created_at > (EXTRACT(EPOCH FROM NOW()) * 1000 - $${filter.paramOffset})
|
||||
`,
|
||||
channelId
|
||||
? [guildId, channelId, hours * 3_600_000]
|
||||
: [guildId, hours * 3_600_000],
|
||||
);
|
||||
|
||||
const row = rows[0] as Record<string, unknown> | undefined;
|
||||
|
||||
return {
|
||||
total_attachments: Number(row?.total_attachments ?? 0),
|
||||
uploaded: Number(row?.uploaded ?? 0),
|
||||
pending: Number(row?.pending ?? 0),
|
||||
failed: Number(row?.failed ?? 0),
|
||||
total_size_bytes: Number(row?.total_size_bytes ?? 0),
|
||||
unique_uploaders: Number(row?.unique_uploaders ?? 0),
|
||||
top_mime_type: (row?.top_mime_type as string) ?? null,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
export const analyticsRepository = new AnalyticsRepository();
|
||||
@@ -1,33 +0,0 @@
|
||||
import type { Router } from "express";
|
||||
import express from "express";
|
||||
import {
|
||||
handleGetAIStats,
|
||||
handleGetAttachmentStats,
|
||||
handleGetDailyTrend,
|
||||
handleGetHeatmap,
|
||||
handleGetHourlyStats,
|
||||
handleGetModerationActions,
|
||||
handleGetModerationStats,
|
||||
handleGetOverview,
|
||||
handleGetTopics,
|
||||
handleGetTopViolators,
|
||||
handleGetUserLeaderboard,
|
||||
} from "./analytics.controller.js";
|
||||
|
||||
export function createAnalyticsRouter(): Router {
|
||||
const router = express.Router();
|
||||
|
||||
router.get("/analytics/overview", handleGetOverview);
|
||||
router.get("/analytics/trend", handleGetDailyTrend);
|
||||
router.get("/analytics/hourly", handleGetHourlyStats);
|
||||
router.get("/analytics/violators", handleGetTopViolators);
|
||||
router.get("/analytics/leaderboard", handleGetUserLeaderboard);
|
||||
router.get("/analytics/stats", handleGetModerationStats);
|
||||
router.get("/analytics/heatmap", handleGetHeatmap);
|
||||
router.get("/analytics/topics", handleGetTopics);
|
||||
router.get("/analytics/moderation-actions", handleGetModerationActions);
|
||||
router.get("/analytics/ai-stats", handleGetAIStats);
|
||||
router.get("/analytics/attachment-stats", handleGetAttachmentStats);
|
||||
|
||||
return router;
|
||||
}
|
||||
@@ -1,9 +0,0 @@
|
||||
import { z } from "zod";
|
||||
|
||||
export const analyticsQuerySchema = z.object({
|
||||
guildId: z.string(),
|
||||
channelId: z.string().optional(),
|
||||
hours: z.coerce.number().int().positive().default(24),
|
||||
});
|
||||
|
||||
export type AnalyticsQuery = z.infer<typeof analyticsQuerySchema>;
|
||||
@@ -1,127 +0,0 @@
|
||||
import { ForbiddenError, ValidationError } from "@bete/shared/errors";
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
import { config } from "../../shared/config/index.js";
|
||||
import { analyticsRepository } from "./analytics.repository.js";
|
||||
import type { AnalyticsQuery } from "./analytics.schema.js";
|
||||
|
||||
const logger = createChildLogger("analytics.service");
|
||||
|
||||
export class AnalyticsService {
|
||||
private assertMonitorGuild(guildId: string) {
|
||||
if (!config.MONITOR_GUILD_ID) {
|
||||
throw new ValidationError("MONITOR_GUILD_ID is not configured");
|
||||
}
|
||||
|
||||
if (guildId !== config.MONITOR_GUILD_ID) {
|
||||
throw new ForbiddenError("Analytics are restricted to the monitor guild");
|
||||
}
|
||||
}
|
||||
|
||||
async getOverview(query: AnalyticsQuery) {
|
||||
this.assertMonitorGuild(query.guildId);
|
||||
logger.debug({ query }, "Getting analytics overview");
|
||||
return analyticsRepository.getOverview(
|
||||
query.guildId,
|
||||
query.channelId,
|
||||
query.hours,
|
||||
);
|
||||
}
|
||||
|
||||
async getDailyTrend(guildId: string, hours = 24) {
|
||||
this.assertMonitorGuild(guildId);
|
||||
logger.debug({ guildId, hours }, "Getting daily trend");
|
||||
return analyticsRepository.getDailyTrend(guildId, hours);
|
||||
}
|
||||
|
||||
async getHourlyStats(guildId: string, channelId?: string, hours = 24) {
|
||||
this.assertMonitorGuild(guildId);
|
||||
logger.debug({ guildId, channelId, hours }, "Getting hourly stats");
|
||||
return analyticsRepository.getHourlyStats(guildId, channelId, hours);
|
||||
}
|
||||
|
||||
async getTopViolators(
|
||||
guildId: string,
|
||||
channelId?: string,
|
||||
hours = 24,
|
||||
limit = 10,
|
||||
) {
|
||||
this.assertMonitorGuild(guildId);
|
||||
logger.debug({ guildId, channelId, hours, limit }, "Getting top violators");
|
||||
return analyticsRepository.getTopViolators(
|
||||
guildId,
|
||||
channelId,
|
||||
hours,
|
||||
limit,
|
||||
);
|
||||
}
|
||||
|
||||
async getUserLeaderboard(
|
||||
guildId: string,
|
||||
channelId?: string,
|
||||
hours = 24,
|
||||
limit = 10,
|
||||
) {
|
||||
this.assertMonitorGuild(guildId);
|
||||
logger.debug(
|
||||
{ guildId, channelId, hours, limit },
|
||||
"Getting user leaderboard",
|
||||
);
|
||||
return analyticsRepository.getUserLeaderboard(
|
||||
guildId,
|
||||
channelId,
|
||||
hours,
|
||||
limit,
|
||||
);
|
||||
}
|
||||
|
||||
async getModerationStats(guildId: string, channelId?: string, hours = 24) {
|
||||
this.assertMonitorGuild(guildId);
|
||||
logger.debug({ guildId, channelId, hours }, "Getting moderation stats");
|
||||
return analyticsRepository.getModerationStats(guildId, channelId, hours);
|
||||
}
|
||||
|
||||
async getHeatmap(guildId: string, channelId?: string, hours = 24) {
|
||||
this.assertMonitorGuild(guildId);
|
||||
logger.debug({ guildId, channelId, hours }, "Getting heatmap");
|
||||
return analyticsRepository.getHeatmap(guildId, channelId, hours);
|
||||
}
|
||||
|
||||
async getTopics(guildId: string, channelId?: string, hours = 24) {
|
||||
this.assertMonitorGuild(guildId);
|
||||
logger.debug({ guildId, channelId, hours }, "Getting topics");
|
||||
return analyticsRepository.getTopics(guildId, channelId, hours);
|
||||
}
|
||||
|
||||
async getModerationActions(
|
||||
guildId: string,
|
||||
channelId?: string,
|
||||
hours = 24,
|
||||
limit = 20,
|
||||
) {
|
||||
this.assertMonitorGuild(guildId);
|
||||
logger.debug(
|
||||
{ guildId, channelId, hours, limit },
|
||||
"Getting moderation actions",
|
||||
);
|
||||
return analyticsRepository.getModerationActions(
|
||||
guildId,
|
||||
channelId,
|
||||
hours,
|
||||
limit,
|
||||
);
|
||||
}
|
||||
|
||||
async getAIStats(guildId: string, channelId?: string, hours = 24) {
|
||||
this.assertMonitorGuild(guildId);
|
||||
logger.debug({ guildId, channelId, hours }, "Getting AI stats");
|
||||
return analyticsRepository.getAIStats(guildId, channelId, hours);
|
||||
}
|
||||
|
||||
async getAttachmentStats(guildId: string, channelId?: string, hours = 24) {
|
||||
this.assertMonitorGuild(guildId);
|
||||
logger.debug({ guildId, channelId, hours }, "Getting attachment stats");
|
||||
return analyticsRepository.getAttachmentStats(guildId, channelId, hours);
|
||||
}
|
||||
}
|
||||
|
||||
export const analyticsService = new AnalyticsService();
|
||||
@@ -1,8 +1,8 @@
|
||||
import { UnauthorizedError } from "@bete/shared/errors";
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
import type { Request, Response, Router } from "express";
|
||||
import express from "express";
|
||||
import { config } from "../../shared/config/index.js";
|
||||
import { UnauthorizedError } from "@bete/shared/errors";
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
import { asyncHandler } from "../../shared/middlewares/index.js";
|
||||
|
||||
const logger = createChildLogger("auth.routes");
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { getPool } from "../../shared/database/index.js";
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
import { getPool } from "../../shared/database/index.js";
|
||||
|
||||
const logger = createChildLogger("health.repository");
|
||||
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import type { Request, Response } from "express";
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
import type { Request, Response } from "express";
|
||||
import { mascotChatService } from "./mascot-chat.service.js";
|
||||
|
||||
const logger = createChildLogger("mascot-chat.controller");
|
||||
@@ -20,11 +20,15 @@ export async function handleMascotChat(req: Request, res: Response) {
|
||||
|
||||
logger.debug(
|
||||
{ userId, messageLength: message.length, context },
|
||||
"Received mascot chat message"
|
||||
"Received mascot chat message",
|
||||
);
|
||||
|
||||
// Process message & generate response
|
||||
const response = await mascotChatService.processMessage(message, context, userId);
|
||||
const response = await mascotChatService.processMessage(
|
||||
message,
|
||||
context,
|
||||
userId,
|
||||
);
|
||||
|
||||
// Save conversation to database
|
||||
await mascotChatService.saveConversation({
|
||||
|
||||
@@ -0,0 +1,180 @@
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
import { getPool } from "../../shared/database/index.js";
|
||||
|
||||
const logger = createChildLogger("mascot-chat.repository");
|
||||
|
||||
export interface MascotChatContext {
|
||||
messageCount?: number;
|
||||
activeParticipants?: number;
|
||||
lastActivity?: string;
|
||||
topicsDiscussed?: string[];
|
||||
guildId?: string;
|
||||
channelId?: string;
|
||||
}
|
||||
|
||||
export interface SaveConversationInput {
|
||||
userId: string;
|
||||
userMessage: string;
|
||||
mascotResponse: string;
|
||||
context?: MascotChatContext;
|
||||
timestamp: Date;
|
||||
}
|
||||
|
||||
export interface MascotChatHistoryRow {
|
||||
id: string;
|
||||
user_id: string;
|
||||
user_message: string;
|
||||
mascot_response: string;
|
||||
context: MascotChatContext | null;
|
||||
created_at: string;
|
||||
}
|
||||
|
||||
export interface ServerInsights {
|
||||
total_messages: number;
|
||||
active_users: number;
|
||||
flagged: number;
|
||||
warned: number;
|
||||
}
|
||||
|
||||
export class MascotChatRepository {
|
||||
private initialized = false;
|
||||
|
||||
async ensureSchema(): Promise<void> {
|
||||
if (this.initialized) return;
|
||||
|
||||
const pool = getPool();
|
||||
await pool.query(`
|
||||
CREATE TABLE IF NOT EXISTS mascot_chat_messages (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
user_id TEXT NOT NULL,
|
||||
user_message TEXT NOT NULL,
|
||||
mascot_response TEXT NOT NULL,
|
||||
context JSONB NOT NULL DEFAULT '{}'::jsonb,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
)
|
||||
`);
|
||||
await pool.query(`
|
||||
CREATE INDEX IF NOT EXISTS idx_mascot_chat_messages_user_created
|
||||
ON mascot_chat_messages (user_id, created_at DESC)
|
||||
`);
|
||||
|
||||
this.initialized = true;
|
||||
logger.info("Mascot chat schema ready");
|
||||
}
|
||||
|
||||
async saveConversation(input: SaveConversationInput): Promise<void> {
|
||||
await this.ensureSchema();
|
||||
const pool = getPool();
|
||||
|
||||
await pool.query(
|
||||
`
|
||||
INSERT INTO mascot_chat_messages
|
||||
(user_id, user_message, mascot_response, context, created_at)
|
||||
VALUES ($1, $2, $3, $4::jsonb, $5)
|
||||
`,
|
||||
[
|
||||
input.userId,
|
||||
input.userMessage,
|
||||
input.mascotResponse,
|
||||
JSON.stringify(input.context ?? {}),
|
||||
input.timestamp.toISOString(),
|
||||
],
|
||||
);
|
||||
|
||||
logger.debug({ userId: input.userId }, "Conversation saved");
|
||||
}
|
||||
|
||||
async getChatHistory(
|
||||
userId: string,
|
||||
limit: number,
|
||||
): Promise<MascotChatHistoryRow[]> {
|
||||
await this.ensureSchema();
|
||||
const pool = getPool();
|
||||
|
||||
const { rows } = await pool.query<MascotChatHistoryRow>(
|
||||
`
|
||||
SELECT id, user_id, user_message, mascot_response, context, created_at
|
||||
FROM mascot_chat_messages
|
||||
WHERE user_id = $1
|
||||
ORDER BY created_at DESC
|
||||
LIMIT $2
|
||||
`,
|
||||
[userId, limit],
|
||||
);
|
||||
|
||||
logger.debug({ userId, count: rows.length }, "Chat history fetched");
|
||||
return rows.reverse();
|
||||
}
|
||||
|
||||
async clearChatHistory(userId: string): Promise<void> {
|
||||
await this.ensureSchema();
|
||||
const pool = getPool();
|
||||
|
||||
const { rowCount } = await pool.query(
|
||||
`DELETE FROM mascot_chat_messages WHERE user_id = $1`,
|
||||
[userId],
|
||||
);
|
||||
|
||||
logger.info({ userId, deletedRows: rowCount ?? 0 }, "Chat history cleared");
|
||||
}
|
||||
|
||||
async getServerInsights(
|
||||
guildId?: string,
|
||||
channelId?: string,
|
||||
): Promise<ServerInsights> {
|
||||
await this.ensureSchema();
|
||||
const pool = getPool();
|
||||
|
||||
try {
|
||||
const params: string[] = [];
|
||||
const clauses: string[] = [];
|
||||
|
||||
if (guildId) {
|
||||
params.push(guildId);
|
||||
clauses.push(`guild_id = $${params.length}`);
|
||||
}
|
||||
if (channelId) {
|
||||
params.push(channelId);
|
||||
clauses.push(`channel_id = $${params.length}`);
|
||||
}
|
||||
|
||||
const where = clauses.length ? `WHERE ${clauses.join(" AND ")}` : "";
|
||||
|
||||
const { rows } = await pool.query<ServerInsights>(
|
||||
`
|
||||
SELECT
|
||||
COUNT(*)::int AS total_messages,
|
||||
COUNT(DISTINCT user_id)::int AS active_users,
|
||||
COUNT(*) FILTER (WHERE ai_status = 'flagged')::int AS flagged,
|
||||
COUNT(*) FILTER (WHERE ai_status = 'warn')::int AS warned
|
||||
FROM messages
|
||||
${where}
|
||||
`,
|
||||
params,
|
||||
);
|
||||
|
||||
const insights = rows[0] ?? {
|
||||
total_messages: 0,
|
||||
active_users: 0,
|
||||
flagged: 0,
|
||||
warned: 0,
|
||||
};
|
||||
|
||||
logger.debug({ guildId, channelId, insights }, "Server insights fetched");
|
||||
return insights;
|
||||
} catch (error) {
|
||||
logger.warn(
|
||||
{ error, guildId, channelId },
|
||||
"Failed to load server insights",
|
||||
);
|
||||
return {
|
||||
total_messages: 0,
|
||||
active_users: 0,
|
||||
flagged: 0,
|
||||
warned: 0,
|
||||
};
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export const mascotChatRepository = new MascotChatRepository();
|
||||
@@ -0,0 +1,29 @@
|
||||
import { z } from "zod";
|
||||
|
||||
export const contextSchema = z.object({
|
||||
messageCount: z.number().int().nonnegative().optional(),
|
||||
activeParticipants: z.number().int().nonnegative().optional(),
|
||||
lastActivity: z.string().datetime().optional(),
|
||||
topicsDiscussed: z.array(z.string()).optional(),
|
||||
guildId: z.string().optional(),
|
||||
channelId: z.string().optional(),
|
||||
});
|
||||
|
||||
export const chatRequestSchema = z.object({
|
||||
message: z.string().min(1, "Message is required"),
|
||||
context: contextSchema.optional(),
|
||||
});
|
||||
|
||||
export const chatResponseSchema = z.object({
|
||||
response: z.string(),
|
||||
timestamp: z.string(),
|
||||
});
|
||||
|
||||
export const chatHistoryQuerySchema = z.object({
|
||||
limit: z.coerce.number().int().positive().max(100).default(50),
|
||||
});
|
||||
|
||||
export type ChatRequest = z.infer<typeof chatRequestSchema>;
|
||||
export type ChatResponse = z.infer<typeof chatResponseSchema>;
|
||||
export type ChatContext = z.infer<typeof contextSchema>;
|
||||
export type ChatHistoryQuery = z.infer<typeof chatHistoryQuerySchema>;
|
||||
@@ -1,47 +1,25 @@
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
import { config } from "../../shared/config/index.js";
|
||||
import { getPool } from "../../shared/database/index.js";
|
||||
import type {
|
||||
MascotChatContext,
|
||||
MascotChatHistoryRow,
|
||||
SaveConversationInput,
|
||||
} from "./mascot-chat.repository.js";
|
||||
import { mascotChatRepository } from "./mascot-chat.repository.js";
|
||||
|
||||
const logger = createChildLogger("mascot-chat.service");
|
||||
|
||||
export interface MascotChatContext {
|
||||
messageCount?: number;
|
||||
activeParticipants?: number;
|
||||
lastActivity?: string;
|
||||
topicsDiscussed?: string[];
|
||||
guildId?: string;
|
||||
channelId?: string;
|
||||
}
|
||||
|
||||
export interface SaveConversationInput {
|
||||
userId: string;
|
||||
userMessage: string;
|
||||
mascotResponse: string;
|
||||
context?: MascotChatContext;
|
||||
timestamp: Date;
|
||||
}
|
||||
|
||||
export interface MascotChatHistoryRow {
|
||||
id: string;
|
||||
user_id: string;
|
||||
user_message: string;
|
||||
mascot_response: string;
|
||||
context: MascotChatContext | null;
|
||||
created_at: string;
|
||||
}
|
||||
|
||||
class MascotChatService {
|
||||
private initialized = false;
|
||||
|
||||
async processMessage(
|
||||
message: string,
|
||||
context: MascotChatContext | undefined,
|
||||
userId: string,
|
||||
): Promise<string> {
|
||||
await this.ensureSchema();
|
||||
|
||||
const recentContext = await this.getRecentConversationContext(userId);
|
||||
const serverInsights = await this.getServerInsights(context);
|
||||
const serverInsights = await mascotChatRepository.getServerInsights(
|
||||
context?.guildId,
|
||||
context?.channelId,
|
||||
);
|
||||
|
||||
// Build LLM messages
|
||||
const systemPrompt = this.buildSystemPrompt(serverInsights);
|
||||
@@ -56,142 +34,30 @@ class MascotChatService {
|
||||
}
|
||||
|
||||
async saveConversation(input: SaveConversationInput): Promise<void> {
|
||||
await this.ensureSchema();
|
||||
const pool = getPool();
|
||||
|
||||
await pool.query(
|
||||
`
|
||||
INSERT INTO mascot_chat_messages
|
||||
(user_id, user_message, mascot_response, context, created_at)
|
||||
VALUES ($1, $2, $3, $4::jsonb, $5)
|
||||
`,
|
||||
[
|
||||
input.userId,
|
||||
input.userMessage,
|
||||
input.mascotResponse,
|
||||
JSON.stringify(input.context ?? {}),
|
||||
input.timestamp.toISOString(),
|
||||
],
|
||||
);
|
||||
await mascotChatRepository.saveConversation(input);
|
||||
}
|
||||
|
||||
async getChatHistory(
|
||||
userId: string,
|
||||
limit: number,
|
||||
): Promise<MascotChatHistoryRow[]> {
|
||||
await this.ensureSchema();
|
||||
const pool = getPool();
|
||||
|
||||
const { rows } = await pool.query<MascotChatHistoryRow>(
|
||||
`
|
||||
SELECT id, user_id, user_message, mascot_response, context, created_at
|
||||
FROM mascot_chat_messages
|
||||
WHERE user_id = $1
|
||||
ORDER BY created_at DESC
|
||||
LIMIT $2
|
||||
`,
|
||||
[userId, limit],
|
||||
);
|
||||
|
||||
return rows.reverse();
|
||||
return mascotChatRepository.getChatHistory(userId, limit);
|
||||
}
|
||||
|
||||
async clearChatHistory(userId: string): Promise<void> {
|
||||
await this.ensureSchema();
|
||||
const pool = getPool();
|
||||
await pool.query(`DELETE FROM mascot_chat_messages WHERE user_id = $1`, [
|
||||
userId,
|
||||
]);
|
||||
}
|
||||
|
||||
private async ensureSchema(): Promise<void> {
|
||||
if (this.initialized) return;
|
||||
|
||||
const pool = getPool();
|
||||
await pool.query(`
|
||||
CREATE TABLE IF NOT EXISTS mascot_chat_messages (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
user_id TEXT NOT NULL,
|
||||
user_message TEXT NOT NULL,
|
||||
mascot_response TEXT NOT NULL,
|
||||
context JSONB NOT NULL DEFAULT '{}'::jsonb,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
)
|
||||
`);
|
||||
await pool.query(`
|
||||
CREATE INDEX IF NOT EXISTS idx_mascot_chat_messages_user_created
|
||||
ON mascot_chat_messages (user_id, created_at DESC)
|
||||
`);
|
||||
|
||||
this.initialized = true;
|
||||
logger.info("Mascot chat schema ready");
|
||||
await mascotChatRepository.clearChatHistory(userId);
|
||||
}
|
||||
|
||||
private async getRecentConversationContext(
|
||||
userId: string,
|
||||
): Promise<string[]> {
|
||||
const history = await this.getChatHistory(userId, 3);
|
||||
const history = await mascotChatRepository.getChatHistory(userId, 3);
|
||||
return history.flatMap((row) => [
|
||||
`User: ${row.user_message}`,
|
||||
`Mascot: ${row.mascot_response}`,
|
||||
]);
|
||||
}
|
||||
|
||||
private async getServerInsights(context?: MascotChatContext) {
|
||||
const pool = getPool();
|
||||
const guildId = context?.guildId;
|
||||
const channelId = context?.channelId;
|
||||
|
||||
try {
|
||||
const params: string[] = [];
|
||||
const clauses: string[] = [];
|
||||
if (guildId) {
|
||||
params.push(guildId);
|
||||
clauses.push(`guild_id = $${params.length}`);
|
||||
}
|
||||
if (channelId) {
|
||||
params.push(channelId);
|
||||
clauses.push(`channel_id = $${params.length}`);
|
||||
}
|
||||
const where = clauses.length ? `WHERE ${clauses.join(" AND ")}` : "";
|
||||
|
||||
const { rows } = await pool.query<{
|
||||
total_messages: number;
|
||||
active_users: number;
|
||||
flagged: number;
|
||||
warned: number;
|
||||
}>(
|
||||
`
|
||||
SELECT
|
||||
COUNT(*)::int AS total_messages,
|
||||
COUNT(DISTINCT user_id)::int AS active_users,
|
||||
COUNT(*) FILTER (WHERE ai_status = 'flagged')::int AS flagged,
|
||||
COUNT(*) FILTER (WHERE ai_status = 'warn')::int AS warned
|
||||
FROM messages
|
||||
${where}
|
||||
`,
|
||||
params,
|
||||
);
|
||||
|
||||
return (
|
||||
rows[0] ?? {
|
||||
total_messages: 0,
|
||||
active_users: 0,
|
||||
flagged: 0,
|
||||
warned: 0,
|
||||
}
|
||||
);
|
||||
} catch (error) {
|
||||
logger.warn({ error }, "Failed to load mascot server insights");
|
||||
return {
|
||||
total_messages: context?.messageCount ?? 0,
|
||||
active_users: context?.activeParticipants ?? 0,
|
||||
flagged: 0,
|
||||
warned: 0,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
private buildSystemPrompt(insights: {
|
||||
total_messages: number;
|
||||
active_users: number;
|
||||
|
||||
@@ -1,8 +1,8 @@
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
import type { Request, Response, Router } from "express";
|
||||
import express from "express";
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
import { asyncHandler } from "../../shared/middlewares/index.js";
|
||||
import { queue, skip, stop, setVolume, getStatus } from "./media.service.js";
|
||||
import { getStatus, queue, setVolume, skip, stop } from "./media.service.js";
|
||||
|
||||
const logger = createChildLogger("media.routes");
|
||||
|
||||
|
||||
@@ -1,9 +1,6 @@
|
||||
import type { NextFunction, Request, Response } from "express";
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
import {
|
||||
asyncHandler,
|
||||
requireParam,
|
||||
} from "../../shared/middlewares/index.js";
|
||||
import type { NextFunction, Request, Response } from "express";
|
||||
import { asyncHandler, requireParam } from "../../shared/middlewares/index.js";
|
||||
import { messageQuerySchema } from "./messages.schema.js";
|
||||
import { messagesService } from "./messages.service.js";
|
||||
|
||||
@@ -28,7 +25,11 @@ export function handleGetMessagesByChannel(
|
||||
next: NextFunction,
|
||||
) {
|
||||
return asyncHandler(async (req: Request, res: Response) => {
|
||||
const channelId = requireParam(req.params.channelId, "route parameter", "channelId");
|
||||
const channelId = requireParam(
|
||||
req.params.channelId,
|
||||
"route parameter",
|
||||
"channelId",
|
||||
);
|
||||
const query = messageQuerySchema.parse(req.query);
|
||||
logger.debug({ channelId, query }, "Handling get messages by channel");
|
||||
const result = await messagesService.getMessagesByChannel(channelId, query);
|
||||
@@ -55,7 +56,11 @@ export function handleGetAttachmentsByChannel(
|
||||
next: NextFunction,
|
||||
) {
|
||||
return asyncHandler(async (req: Request, res: Response) => {
|
||||
const channelId = requireParam(req.params.channelId, "route parameter", "channelId");
|
||||
const channelId = requireParam(
|
||||
req.params.channelId,
|
||||
"route parameter",
|
||||
"channelId",
|
||||
);
|
||||
const query = messageQuerySchema.parse(req.query);
|
||||
logger.debug({ channelId, query }, "Handling get attachments by channel");
|
||||
const result = await messagesService.getAttachmentsByChannel(
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { getPool } from "../../shared/database/index.js";
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
import { getPool } from "../../shared/database/index.js";
|
||||
import type {
|
||||
MessageCreate,
|
||||
MessageQuery,
|
||||
@@ -61,7 +61,9 @@ function mapMessageRow(row: Record<string, unknown>) {
|
||||
}
|
||||
|
||||
export class MessagesRepository {
|
||||
async findMany(query: MessageQuery): Promise<PageResult<ReturnType<typeof mapMessageRow>>> {
|
||||
async findMany(
|
||||
query: MessageQuery,
|
||||
): Promise<PageResult<ReturnType<typeof mapMessageRow>>> {
|
||||
const pool = getPool();
|
||||
const limit = query.limit ?? 50;
|
||||
const clauses: string[] = [];
|
||||
@@ -101,7 +103,8 @@ export class MessagesRepository {
|
||||
);
|
||||
|
||||
const data = rows.slice(0, limit).map(mapMessageRow);
|
||||
const nextCursor = rows.length > limit ? String(rows[limit].created_at) : null;
|
||||
const nextCursor =
|
||||
rows.length > limit ? String(rows[limit].created_at) : null;
|
||||
|
||||
logger.debug({ count: data.length, nextCursor }, "Found messages");
|
||||
return { data, nextCursor };
|
||||
@@ -109,10 +112,9 @@ export class MessagesRepository {
|
||||
|
||||
async findById(id: string) {
|
||||
const pool = getPool();
|
||||
const { rows } = await pool.query(
|
||||
`SELECT * FROM messages WHERE id = $1`,
|
||||
[id],
|
||||
);
|
||||
const { rows } = await pool.query(`SELECT * FROM messages WHERE id = $1`, [
|
||||
id,
|
||||
]);
|
||||
|
||||
if (rows.length === 0) return null;
|
||||
return mapMessageRow(rows[0] as Record<string, unknown>);
|
||||
@@ -140,7 +142,8 @@ export class MessagesRepository {
|
||||
);
|
||||
|
||||
const data = rows.slice(0, limit).map(mapMessageRow);
|
||||
const nextCursor = rows.length > limit ? String(rows[limit].created_at) : null;
|
||||
const nextCursor =
|
||||
rows.length > limit ? String(rows[limit].created_at) : null;
|
||||
|
||||
return { data, nextCursor };
|
||||
}
|
||||
@@ -260,6 +263,58 @@ export class MessagesRepository {
|
||||
return rowCount ?? 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* Mark a single message for re-analysis by resetting ai_status to 'pending'.
|
||||
* Skips messages already in 'pending' state to avoid write amplification.
|
||||
*/
|
||||
async markForReanalysis(id: string): Promise<void> {
|
||||
const pool = getPool();
|
||||
await pool.query(
|
||||
`UPDATE messages SET ai_status = 'pending'
|
||||
WHERE id = $1 AND ai_status != 'pending'`,
|
||||
[id],
|
||||
);
|
||||
logger.debug({ id }, "Message marked for re-analysis");
|
||||
}
|
||||
|
||||
/**
|
||||
* Retrieve messages flagged for review (ai_status IN ('warn', 'flagged')).
|
||||
* Optionally filtered by channelId, with configurable limit.
|
||||
*/
|
||||
async getReviewMessages(
|
||||
channelId?: string,
|
||||
limit: number = 20,
|
||||
): Promise<Record<string, unknown>[]> {
|
||||
const pool = getPool();
|
||||
|
||||
if (channelId) {
|
||||
const { rows } = await pool.query(
|
||||
`SELECT id, guild_id, channel_id, user_id, username, avatar_url,
|
||||
content, type, created_at, ai_status, ai_severity,
|
||||
ai_confidence, ai_analysis
|
||||
FROM messages
|
||||
WHERE ai_status IN ('warn', 'flagged')
|
||||
AND channel_id = $1
|
||||
ORDER BY created_at DESC
|
||||
LIMIT $2`,
|
||||
[channelId, limit],
|
||||
);
|
||||
return rows;
|
||||
}
|
||||
|
||||
const { rows } = await pool.query(
|
||||
`SELECT id, guild_id, channel_id, user_id, username, avatar_url,
|
||||
content, type, created_at, ai_status, ai_severity,
|
||||
ai_confidence, ai_analysis
|
||||
FROM messages
|
||||
WHERE ai_status IN ('warn', 'flagged')
|
||||
ORDER BY created_at DESC
|
||||
LIMIT $1`,
|
||||
[limit],
|
||||
);
|
||||
return rows;
|
||||
}
|
||||
|
||||
async delete(id: string): Promise<boolean> {
|
||||
const pool = getPool();
|
||||
const { rowCount } = await pool.query(
|
||||
@@ -308,7 +363,8 @@ export class MessagesRepository {
|
||||
uploaded_at: (r.uploaded_at as number | null) ?? null,
|
||||
}));
|
||||
|
||||
const nextCursor = data.length > limit ? String(data[limit].created_at) : null;
|
||||
const nextCursor =
|
||||
data.length > limit ? String(data[limit].created_at) : null;
|
||||
const trimmed = data.slice(0, limit);
|
||||
|
||||
return { data: trimmed, nextCursor };
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
import type { Request, Response, Router } from "express";
|
||||
import express from "express";
|
||||
import { getPool } from "../../shared/database/index.js";
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
import { asyncHandler } from "../../shared/middlewares/index.js";
|
||||
import {
|
||||
handleGetAttachmentsByChannel,
|
||||
@@ -85,7 +84,6 @@ export function createMessagesRouter(): Router {
|
||||
}),
|
||||
);
|
||||
|
||||
|
||||
// POST /api/messages/:id/reanalyze - Mark single message for re-analysis
|
||||
router.post(
|
||||
"/messages/:id/reanalyze",
|
||||
@@ -104,20 +102,11 @@ export function createMessagesRouter(): Router {
|
||||
|
||||
reanalyzeInFlight.add(id);
|
||||
try {
|
||||
const pool = getPool();
|
||||
await pool.query(
|
||||
// Only revert to pending if the message is not currently being
|
||||
// processed (pending) already — prevents write amplification when
|
||||
// the recovery worker already picked it up between UI clicks.
|
||||
`UPDATE messages SET ai_status = 'pending'
|
||||
WHERE id = $1 AND ai_status != 'pending'`,
|
||||
[id],
|
||||
);
|
||||
await messagesService.markForReanalysis(id);
|
||||
} finally {
|
||||
reanalyzeInFlight.delete(id);
|
||||
}
|
||||
|
||||
logger.debug({ id }, "Message marked for re-analysis");
|
||||
res.status(200).json({ ok: true });
|
||||
}),
|
||||
);
|
||||
@@ -129,36 +118,7 @@ export function createMessagesRouter(): Router {
|
||||
const limit = Number(req.query.limit) || 20;
|
||||
const channelId = (req.query.channelId as string) || undefined;
|
||||
|
||||
const pool = getPool();
|
||||
|
||||
let sqlQuery: string;
|
||||
let params: (string | number)[];
|
||||
if (channelId) {
|
||||
sqlQuery = `
|
||||
SELECT id, guild_id, channel_id, user_id, username, avatar_url,
|
||||
content, type, created_at, ai_status, ai_severity,
|
||||
ai_confidence, ai_analysis
|
||||
FROM messages
|
||||
WHERE ai_status IN ('warn', 'flagged')
|
||||
AND channel_id = $1
|
||||
ORDER BY created_at DESC
|
||||
LIMIT $2
|
||||
`;
|
||||
params = [channelId, limit];
|
||||
} else {
|
||||
sqlQuery = `
|
||||
SELECT id, guild_id, channel_id, user_id, username, avatar_url,
|
||||
content, type, created_at, ai_status, ai_severity,
|
||||
ai_confidence, ai_analysis
|
||||
FROM messages
|
||||
WHERE ai_status IN ('warn', 'flagged')
|
||||
ORDER BY created_at DESC
|
||||
LIMIT $1
|
||||
`;
|
||||
params = [limit];
|
||||
}
|
||||
|
||||
const { rows } = await pool.query(sqlQuery, params);
|
||||
const rows = await messagesService.getReviewMessages(channelId, limit);
|
||||
logger.debug({ limit, channelId }, "Review query executed");
|
||||
res.json({ results: rows, limit, cursor: null });
|
||||
}),
|
||||
@@ -168,7 +128,7 @@ export function createMessagesRouter(): Router {
|
||||
router.post(
|
||||
"/messages/:id/moderate",
|
||||
asyncHandler(async (req: Request, res: Response) => {
|
||||
const id = req.params.id;
|
||||
const id = String(req.params.id ?? "");
|
||||
if (!id) {
|
||||
res.status(400).json({ error: "MISSING_ID" });
|
||||
return;
|
||||
@@ -196,20 +156,12 @@ export function createMessagesRouter(): Router {
|
||||
}
|
||||
|
||||
// Fetch the message to get guild/user context
|
||||
const pool = getPool();
|
||||
const { rows } = await pool.query(
|
||||
`SELECT id, guild_id, channel_id, thread_id, user_id, content
|
||||
FROM messages WHERE id = $1`,
|
||||
[id],
|
||||
);
|
||||
|
||||
if (rows.length === 0) {
|
||||
const msg = await messagesService.getMessageById(id).catch(() => null);
|
||||
if (!msg) {
|
||||
res.status(404).json({ error: "MESSAGE_NOT_FOUND" });
|
||||
return;
|
||||
}
|
||||
|
||||
const msg = rows[0] as Record<string, unknown>;
|
||||
|
||||
// Publish command to DG via Redis
|
||||
const { publishCommand } = await import("../../ws/redis-bridge.js");
|
||||
await publishCommand({
|
||||
@@ -217,9 +169,9 @@ export function createMessagesRouter(): Router {
|
||||
type: "moderation:action",
|
||||
payload: {
|
||||
messageId: id,
|
||||
guildId: String(msg.guild_id ?? ""),
|
||||
channelId: (msg.thread_id as string) || String(msg.channel_id ?? ""),
|
||||
userId: String(msg.user_id ?? ""),
|
||||
guildId: msg.guild_id,
|
||||
channelId: msg.thread_id || msg.channel_id,
|
||||
userId: msg.user_id,
|
||||
actionType,
|
||||
reason: reason ?? "Manual moderation from dashboard",
|
||||
requestedAt: Date.now(),
|
||||
|
||||
@@ -46,12 +46,33 @@ export class MessagesService {
|
||||
return messagesRepository.getAttachmentsByChannel(channelId, query);
|
||||
}
|
||||
|
||||
async markForReanalysis(id: string): Promise<void> {
|
||||
if (!id) {
|
||||
throw new ValidationError("message ID is required");
|
||||
}
|
||||
|
||||
logger.debug({ id }, "Marking message for re-analysis");
|
||||
await messagesRepository.markForReanalysis(id);
|
||||
}
|
||||
|
||||
async getReviewMessages(
|
||||
channelId?: string,
|
||||
limit?: number,
|
||||
): Promise<Record<string, unknown>[]> {
|
||||
logger.debug({ channelId, limit }, "Getting review messages");
|
||||
return messagesRepository.getReviewMessages(channelId, limit);
|
||||
}
|
||||
|
||||
async reanalyzeErrorBatch(opts: {
|
||||
guildId?: string;
|
||||
channelId?: string;
|
||||
messageIds?: string[];
|
||||
}) {
|
||||
if (!opts.guildId && !opts.channelId && (!opts.messageIds || opts.messageIds.length === 0)) {
|
||||
if (
|
||||
!opts.guildId &&
|
||||
!opts.channelId &&
|
||||
(!opts.messageIds || opts.messageIds.length === 0)
|
||||
) {
|
||||
throw new ValidationError(
|
||||
"At least one of guildId, channelId, or messageIds[] is required",
|
||||
);
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
import type { Request, Response, Router } from "express";
|
||||
import express from "express";
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
import { asyncHandler } from "../../shared/middlewares/index.js";
|
||||
import { recordingsService } from "./recordings.service.js";
|
||||
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
import { sql } from "drizzle-orm";
|
||||
import { getDatabase } from "../../shared/database/index.js";
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
|
||||
const logger = createChildLogger("recordings.service");
|
||||
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
import type { Request, Response, Router } from "express";
|
||||
import express from "express";
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
import { asyncHandler } from "../../shared/middlewares/index.js";
|
||||
import { uiStateService } from "./ui-state.service.js";
|
||||
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
import { sql } from "drizzle-orm";
|
||||
import { getDatabase } from "../../shared/database/index.js";
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
|
||||
const logger = createChildLogger("ui-state.service");
|
||||
|
||||
|
||||
@@ -0,0 +1,39 @@
|
||||
import { z } from "zod";
|
||||
|
||||
export const voiceCommandSchema = z.object({
|
||||
command: z.string().min(1, "command is required"),
|
||||
});
|
||||
|
||||
export const connectVoiceSchema = z.object({
|
||||
guildId: z.string().min(1, "guildId is required"),
|
||||
channelId: z.string().min(1, "channelId is required"),
|
||||
});
|
||||
|
||||
export const guildIdParamSchema = z.object({
|
||||
guildId: z.string().min(1),
|
||||
});
|
||||
|
||||
export const guildSchema = z.object({
|
||||
id: z.string(),
|
||||
name: z.string(),
|
||||
icon: z.string().nullable(),
|
||||
});
|
||||
|
||||
export const channelSchema = z.object({
|
||||
id: z.string(),
|
||||
name: z.string(),
|
||||
type: z.enum(["voice", "text"]),
|
||||
});
|
||||
|
||||
export const voiceStatusSchema = z.object({
|
||||
connected: z.boolean(),
|
||||
activeGuildId: z.string().nullable(),
|
||||
activeChannelId: z.string().nullable(),
|
||||
activeChannelName: z.string().nullable(),
|
||||
});
|
||||
|
||||
export type VoiceCommand = z.infer<typeof voiceCommandSchema>;
|
||||
export type ConnectVoice = z.infer<typeof connectVoiceSchema>;
|
||||
export type Guild = z.infer<typeof guildSchema>;
|
||||
export type Channel = z.infer<typeof channelSchema>;
|
||||
export type VoiceStatus = z.infer<typeof voiceStatusSchema>;
|
||||
@@ -1,9 +1,6 @@
|
||||
import { getPool } from "../../shared/database/index.js";
|
||||
import {
|
||||
publishCommand,
|
||||
readRedisStatus,
|
||||
} from "../../shared/redis/index.js";
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
import { getPool } from "../../shared/database/index.js";
|
||||
import { publishCommand, readRedisStatus } from "../../shared/redis/index.js";
|
||||
|
||||
const logger = createChildLogger("voice.service");
|
||||
|
||||
@@ -32,8 +29,7 @@ export interface VoiceStatus {
|
||||
*/
|
||||
export async function getGuilds(): Promise<Guild[]> {
|
||||
const reply = await publishCommand<Guild[]>("guilds:list", {});
|
||||
if (reply?.success && reply.data && reply.data.length > 0)
|
||||
return reply.data;
|
||||
if (reply?.success && reply.data && reply.data.length > 0) return reply.data;
|
||||
|
||||
// Fallback: Postgres with synthetic names
|
||||
logger.warn(
|
||||
@@ -59,8 +55,7 @@ export async function getTextChannels(guildId: string): Promise<Channel[]> {
|
||||
const reply = await publishCommand<Channel[]>("guilds:text-channels", {
|
||||
guildId,
|
||||
});
|
||||
if (reply?.success && reply.data && reply.data.length > 0)
|
||||
return reply.data;
|
||||
if (reply?.success && reply.data && reply.data.length > 0) return reply.data;
|
||||
|
||||
// Fallback: Postgres with synthetic names
|
||||
logger.warn(
|
||||
@@ -117,12 +112,14 @@ export async function connectVoice(
|
||||
|
||||
// Fallback: read from Redis status key
|
||||
const cached = await readRedisStatus("voice:status");
|
||||
return (cached as unknown as VoiceStatus) ?? {
|
||||
connected: false,
|
||||
activeGuildId: null,
|
||||
activeChannelId: null,
|
||||
activeChannelName: null,
|
||||
};
|
||||
return (
|
||||
(cached as unknown as VoiceStatus) ?? {
|
||||
connected: false,
|
||||
activeGuildId: null,
|
||||
activeChannelId: null,
|
||||
activeChannelName: null,
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -133,10 +130,12 @@ export async function disconnectVoice(): Promise<VoiceStatus> {
|
||||
if (reply?.success && reply.data) return reply.data;
|
||||
|
||||
const cached = await readRedisStatus("voice:status");
|
||||
return (cached as unknown as VoiceStatus) ?? {
|
||||
connected: false,
|
||||
activeGuildId: null,
|
||||
activeChannelId: null,
|
||||
activeChannelName: null,
|
||||
};
|
||||
return (
|
||||
(cached as unknown as VoiceStatus) ?? {
|
||||
connected: false,
|
||||
activeGuildId: null,
|
||||
activeChannelId: null,
|
||||
activeChannelName: null,
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
import { drizzle } from "drizzle-orm/node-postgres";
|
||||
import { Pool } from "pg";
|
||||
import { config } from "../config/index.js";
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
|
||||
const logger = createChildLogger("database");
|
||||
|
||||
|
||||
@@ -1,10 +1,10 @@
|
||||
import type { NextFunction, Request, Response } from "express";
|
||||
import {
|
||||
AppError,
|
||||
UnauthorizedError,
|
||||
ValidationError,
|
||||
} from "@bete/shared/errors";
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
import type { NextFunction, Request, Response } from "express";
|
||||
|
||||
const logger = createChildLogger("middleware");
|
||||
|
||||
@@ -54,7 +54,11 @@ export function asyncHandler(
|
||||
* Validate that a value is a non-empty string, or throw a descriptive error.
|
||||
* Use for both route params and query string values.
|
||||
*/
|
||||
export function requireParam(value: unknown, kind: string, name: string): string {
|
||||
export function requireParam(
|
||||
value: unknown,
|
||||
kind: string,
|
||||
name: string,
|
||||
): string {
|
||||
if (typeof value !== "string" || value.length === 0) {
|
||||
throw new Error(`Missing ${kind}: ${name}`);
|
||||
}
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
import Redis from "ioredis";
|
||||
import { config } from "../config/index.js";
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
|
||||
const logger = createChildLogger("redis.command-channel");
|
||||
|
||||
@@ -73,13 +73,21 @@ export async function publishCommand<T = unknown>(
|
||||
timeoutMs = 5000,
|
||||
): Promise<CommandReply<T> | null> {
|
||||
if (!ensureRedisConfig()) {
|
||||
logger.warn({ commandType }, "Redis not configured, skipping command publish");
|
||||
logger.warn(
|
||||
{ commandType },
|
||||
"Redis not configured, skipping command publish",
|
||||
);
|
||||
return null;
|
||||
}
|
||||
|
||||
const id = randomUUID();
|
||||
const replyChannel = `backend:command:reply:${id}`;
|
||||
const command: CommandMessage = { id, type: commandType, payload, replyChannel };
|
||||
const command: CommandMessage = {
|
||||
id,
|
||||
type: commandType,
|
||||
payload,
|
||||
replyChannel,
|
||||
};
|
||||
|
||||
return new Promise<CommandReply<T> | null>((resolve) => {
|
||||
const pub = getPublisher();
|
||||
@@ -88,7 +96,9 @@ export async function publishCommand<T = unknown>(
|
||||
const timer = setTimeout(() => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
sub.unsubscribe(replyChannel).catch(() => {/* ignore */});
|
||||
sub.unsubscribe(replyChannel).catch(() => {
|
||||
/* ignore */
|
||||
});
|
||||
logger.warn({ id, commandType }, "Command timed out waiting for reply");
|
||||
resolve(null);
|
||||
}, timeoutMs);
|
||||
@@ -99,11 +109,16 @@ export async function publishCommand<T = unknown>(
|
||||
if (channel !== replyChannel || settled) return;
|
||||
settled = true;
|
||||
clearTimeout(timer);
|
||||
sub.unsubscribe(replyChannel).catch(() => {/* ignore */});
|
||||
sub.unsubscribe(replyChannel).catch(() => {
|
||||
/* ignore */
|
||||
});
|
||||
|
||||
try {
|
||||
const reply: CommandReply<T> = JSON.parse(message);
|
||||
logger.debug({ id, commandType, success: reply.success }, "Command reply received");
|
||||
logger.debug(
|
||||
{ id, commandType, success: reply.success },
|
||||
"Command reply received",
|
||||
);
|
||||
resolve(reply);
|
||||
} catch (err) {
|
||||
logger.error({ id, err }, "Failed to parse command reply");
|
||||
@@ -113,29 +128,34 @@ export async function publishCommand<T = unknown>(
|
||||
|
||||
sub.on("message", onMessage);
|
||||
|
||||
sub.subscribe(replyChannel).then(() => {
|
||||
pub
|
||||
.publish("backend:command", JSON.stringify(command))
|
||||
.then(() => {
|
||||
logger.debug({ id, commandType }, "Command published");
|
||||
})
|
||||
.catch((err: Error) => {
|
||||
if (!settled) {
|
||||
settled = true;
|
||||
clearTimeout(timer);
|
||||
sub.unsubscribe(replyChannel).catch(() => {/* ignore */});
|
||||
logger.error({ err }, "Failed to publish command");
|
||||
resolve(null);
|
||||
}
|
||||
});
|
||||
}).catch((err: Error) => {
|
||||
if (!settled) {
|
||||
settled = true;
|
||||
clearTimeout(timer);
|
||||
logger.error({ err }, "Failed to subscribe to reply channel");
|
||||
resolve(null);
|
||||
}
|
||||
});
|
||||
sub
|
||||
.subscribe(replyChannel)
|
||||
.then(() => {
|
||||
pub
|
||||
.publish("backend:command", JSON.stringify(command))
|
||||
.then(() => {
|
||||
logger.debug({ id, commandType }, "Command published");
|
||||
})
|
||||
.catch((err: Error) => {
|
||||
if (!settled) {
|
||||
settled = true;
|
||||
clearTimeout(timer);
|
||||
sub.unsubscribe(replyChannel).catch(() => {
|
||||
/* ignore */
|
||||
});
|
||||
logger.error({ err }, "Failed to publish command");
|
||||
resolve(null);
|
||||
}
|
||||
});
|
||||
})
|
||||
.catch((err: Error) => {
|
||||
if (!settled) {
|
||||
settled = true;
|
||||
clearTimeout(timer);
|
||||
logger.error({ err }, "Failed to subscribe to reply channel");
|
||||
resolve(null);
|
||||
}
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
@@ -147,7 +167,10 @@ export async function publishCommandNoReply(
|
||||
payload: Record<string, unknown> = {},
|
||||
): Promise<void> {
|
||||
if (!ensureRedisConfig()) {
|
||||
logger.warn({ commandType }, "Redis not configured, skipping command publish");
|
||||
logger.warn(
|
||||
{ commandType },
|
||||
"Redis not configured, skipping command publish",
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -213,7 +236,9 @@ export function subscribe(
|
||||
// Status helpers — read keys set by discord-gateway
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export async function readRedisStatus(key: string): Promise<Record<string, unknown> | null> {
|
||||
export async function readRedisStatus(
|
||||
key: string,
|
||||
): Promise<Record<string, unknown> | null> {
|
||||
if (!ensureRedisConfig()) {
|
||||
return null;
|
||||
}
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
import Redis from "ioredis";
|
||||
import { config } from "../shared/config/index.js";
|
||||
import { getCommandPublisher } from "../shared/redis/index.js";
|
||||
import { createChildLogger } from "@bete/shared/logger";
|
||||
import { broadcastRaw } from "./broadcast.js";
|
||||
|
||||
const logger = createChildLogger("ws.redis-bridge");
|
||||
|
||||
@@ -66,38 +66,61 @@ export function createWebSocketServer(server: Server): WebSocketServer {
|
||||
|
||||
ws.on("message", (data: Buffer) => {
|
||||
// Handle JSON messages from browser
|
||||
if (typeof data === 'string' || (Buffer.isBuffer(data) && data.length > 0 && data[0] === 0x7B)) {
|
||||
if (
|
||||
typeof data === "string" ||
|
||||
(Buffer.isBuffer(data) && data.length > 0 && data[0] === 0x7b)
|
||||
) {
|
||||
try {
|
||||
const message = JSON.parse(data.toString());
|
||||
|
||||
if (message.type === 'voice_transmit' && message.buffer) {
|
||||
if (message.type === "voice_transmit" && message.buffer) {
|
||||
// Forward PCM data to Redis for discord-gateway
|
||||
import('../shared/redis/index.js').then(({ getCommandPublisher }) => {
|
||||
const publisher = getCommandPublisher();
|
||||
publisher.publish('backend:voice:transmit', JSON.stringify({
|
||||
type: 'pcm',
|
||||
buffer: message.buffer
|
||||
})).catch((err: Error) => {
|
||||
logger.error({ err }, 'Failed to publish voice transmit to Redis');
|
||||
});
|
||||
});
|
||||
} else if (message.type === 'voice_command' && message.command) {
|
||||
import("../shared/redis/index.js").then(
|
||||
({ getCommandPublisher }) => {
|
||||
const publisher = getCommandPublisher();
|
||||
publisher
|
||||
.publish(
|
||||
"backend:voice:transmit",
|
||||
JSON.stringify({
|
||||
type: "pcm",
|
||||
buffer: message.buffer,
|
||||
}),
|
||||
)
|
||||
.catch((err: Error) => {
|
||||
logger.error(
|
||||
{ err },
|
||||
"Failed to publish voice transmit to Redis",
|
||||
);
|
||||
});
|
||||
},
|
||||
);
|
||||
} else if (message.type === "voice_command" && message.command) {
|
||||
// Forward voice commands to discord-gateway
|
||||
import('../shared/redis/index.js').then(({ getCommandPublisher }) => {
|
||||
const publisher = getCommandPublisher();
|
||||
const commandId = `cmd-${Date.now()}-${Math.random().toString(36).slice(2, 9)}`;
|
||||
publisher.publish('backend:command', JSON.stringify({
|
||||
id: commandId,
|
||||
type: message.command,
|
||||
payload: {},
|
||||
replyChannel: `reply:${commandId}`
|
||||
})).catch((err: Error) => {
|
||||
logger.error({ err }, 'Failed to publish voice command to Redis');
|
||||
});
|
||||
});
|
||||
import("../shared/redis/index.js").then(
|
||||
({ getCommandPublisher }) => {
|
||||
const publisher = getCommandPublisher();
|
||||
const commandId = `cmd-${Date.now()}-${Math.random().toString(36).slice(2, 9)}`;
|
||||
publisher
|
||||
.publish(
|
||||
"backend:command",
|
||||
JSON.stringify({
|
||||
id: commandId,
|
||||
type: message.command,
|
||||
payload: {},
|
||||
replyChannel: `reply:${commandId}`,
|
||||
}),
|
||||
)
|
||||
.catch((err: Error) => {
|
||||
logger.error(
|
||||
{ err },
|
||||
"Failed to publish voice command to Redis",
|
||||
);
|
||||
});
|
||||
},
|
||||
);
|
||||
}
|
||||
} catch (err) {
|
||||
logger.debug({ err }, 'Failed to parse WebSocket message as JSON');
|
||||
logger.debug({ err }, "Failed to parse WebSocket message as JSON");
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user