diff --git a/src/moderation/analyticsStore.ts b/src/moderation/analyticsStore.ts index 472effd..1bcac70 100644 --- a/src/moderation/analyticsStore.ts +++ b/src/moderation/analyticsStore.ts @@ -1,36 +1,9 @@ -import { and, asc, desc, eq, gte, isNull, or, type SQL } from "drizzle-orm"; import { getDatabase } from "../database/drizzle.js"; -import { messagesTable } from "../database/schema.js"; import { createChildLogger } from "../logger.js"; import type { MessageRecord } from "./types.js"; const logger = createChildLogger("analytics-store"); -// ── DB helper ────────────────────────────────────────────────────────── -function db() { - return getDatabase() as { - select(fields?: Record): { - from(table: unknown): { - where(cond: SQL | undefined): { - orderBy(...cols: unknown[]): { - limit(n: number): Promise; - } & Promise; - groupBy(...cols: unknown[]): Promise; - } & Promise; - limit(n: number): Promise; - } & Promise; - }; - }; -} - -// ── Shared condition helper ──────────────────────────────────────────── -function channelFilter(channelId: string): SQL { - return or( - eq(messagesTable.channel_id, channelId), - eq(messagesTable.thread_id, channelId), - ) as SQL; -} - // ── Types ────────────────────────────────────────────────────────────── export interface HourlyBucket { @@ -79,6 +52,25 @@ export interface AnalyticsOverview { total_channels: number; } +// ── Cache for topic trends ───────────────────────────────────────────── + +interface TopicCacheEntry { + data: TopicTrend[]; + expiresAt: number; + key: string; +} + +const topicCache = new Map(); +const TOPIC_CACHE_TTL_MS = 60_000; // 1 minute TTL + +function makeTopicCacheKey(input: { + guildId: string; + channelId?: string; + hours: number; +}): string { + return `${input.guildId}:${input.channelId ?? "*"}:${input.hours}`; +} + // ── Hourly Message Stats ─────────────────────────────────────────────── export async function getHourlyStats(input: { @@ -89,25 +81,30 @@ export async function getHourlyStats(input: { try { const { guildId, channelId, hours = 24 } = input; const since = Date.now() - hours * 3600_000; - const database = db(); + const rawDb = getDatabase() as any; + const sqliteRows = rawDb.all( + ` + SELECT + datetime((created_at / 3600000) * 3600, 'unixepoch') as hour, + count(*) as count, + count(case when ai_status = 'clean' then 1 end) as clean, + count(case when ai_status = 'warn' then 1 end) as warned, + count(case when ai_status = 'flagged' then 1 end) as flagged, + count(case when ai_status = 'error' then 1 end) as error + FROM messages + WHERE guild_id = ? + AND created_at >= ? + AND deleted_at IS NULL + ${channelId ? `AND (channel_id = ? OR thread_id = ?)` : ""} + GROUP BY (created_at / 3600000) + ORDER BY hour ASC + `, + channelId + ? [guildId, since, channelId, channelId] + : [guildId, since], + ); - const conditions: SQL[] = [ - eq(messagesTable.guild_id, guildId), - gte(messagesTable.created_at, since), - isNull(messagesTable.deleted_at), - ]; - - if (channelId) { - conditions.push(channelFilter(channelId)); - } - - const rows = (await database - .select() - .from(messagesTable) - .where(and(...conditions) as SQL) - .orderBy(asc(messagesTable.created_at))) as MessageRecord[]; - - // Initialize all hour buckets + // Initialize all hour buckets (fill gaps with zeros) const buckets = new Map< string, { @@ -126,20 +123,19 @@ export async function getHourlyStats(input: { buckets.set(key, { count: 0, clean: 0, warned: 0, flagged: 0, error: 0 }); } - for (const row of rows) { - const d = new Date(row.created_at); - d.setMinutes(0, 0, 0); + for (const row of sqliteRows) { + // Normalize the SQL hour key to match our bucket format + const d = new Date(row.hour.replace(" ", "T") + "Z"); const key = d.toISOString().slice(0, 13) + ":00:00Z"; const bucket = buckets.get(key); if (!bucket) continue; - bucket.count++; - const status = row.ai_status || "pending"; - if (status === "clean") bucket.clean++; - else if (status === "warn") bucket.warned++; - else if (status === "flagged") bucket.flagged++; - else if (status === "error") bucket.error++; + bucket.count = row.count; + bucket.clean = row.clean; + bucket.warned = row.warned; + bucket.flagged = row.flagged; + bucket.error = row.error; } return Array.from(buckets.entries()) @@ -396,29 +392,47 @@ export async function getTopicTrends(input: { channelId?: string; hours?: number; }): Promise { + const { guildId, channelId, hours = 24 } = input; + const cacheKey = makeTopicCacheKey({ guildId, channelId, hours }); + + // Check cache first (P2: cache topic extraction) + const cached = topicCache.get(cacheKey); + if (cached && cached.expiresAt > Date.now()) { + return cached.data; + } + try { - const { guildId, channelId, hours = 24 } = input; const since = Date.now() - hours * 3600_000; - const database = db(); + const rawDb = getDatabase() as any; - const conditions: SQL[] = [ - eq(messagesTable.guild_id, guildId), - gte(messagesTable.created_at, since), - isNull(messagesTable.deleted_at), - ]; + const rows = rawDb.all( + ` + SELECT + id, content, ai_status, ai_analysis, ai_moderation_score, + ai_moderation_flags, created_at + FROM messages + WHERE guild_id = ? + AND created_at >= ? + AND deleted_at IS NULL + ${channelId ? `AND (channel_id = ? OR thread_id = ?)` : ""} + ORDER BY created_at DESC + LIMIT 1000 + `, + channelId + ? [guildId, since, channelId, channelId] + : [guildId, since], + ) as MessageRecord[]; - if (channelId) { - conditions.push(channelFilter(channelId)); - } + const result = extractTopics(rows); - const rows = (await database - .select() - .from(messagesTable) - .where(and(...conditions) as SQL) - .orderBy(desc(messagesTable.created_at)) - .limit(1000)) as MessageRecord[]; + // Store in cache + topicCache.set(cacheKey, { + data: result, + expiresAt: Date.now() + TOPIC_CACHE_TTL_MS, + key: cacheKey, + }); - return extractTopics(rows); + return result; } catch (error) { logger.error( { error: error instanceof Error ? error.message : String(error) }, @@ -439,55 +453,35 @@ export async function getUserLeaderboard(input: { try { const { guildId, channelId, hours = 24, limit = 20 } = input; const since = Date.now() - hours * 3600_000; - const database = db(); + const rawDb = getDatabase() as any; - const conditions: SQL[] = [ - eq(messagesTable.guild_id, guildId), - gte(messagesTable.created_at, since), - isNull(messagesTable.deleted_at), - ]; + // SQL-level GROUP BY aggregate instead of SELECT * + in-memory map + const rows = rawDb.all( + ` + SELECT + user_id, + username, + avatar_url, + count(*) as message_count, + count(case when type = 'edited' then 1 end) as edited_count, + count(case when type = 'deleted' then 1 end) as deleted_count, + count(case when ai_status in ('flagged', 'warn') then 1 end) as flagged_count, + max(created_at) as last_active + FROM messages + WHERE guild_id = ? + AND created_at >= ? + AND deleted_at IS NULL + ${channelId ? `AND (channel_id = ? OR thread_id = ?)` : ""} + GROUP BY user_id + ORDER BY message_count DESC + LIMIT ? + `, + channelId + ? [guildId, since, channelId, channelId, limit] + : [guildId, since, limit], + ); - if (channelId) { - conditions.push(channelFilter(channelId)); - } - - const rows = (await database - .select() - .from(messagesTable) - .where(and(...conditions) as SQL) - .orderBy(asc(messagesTable.created_at))) as MessageRecord[]; - - const userMap = new Map(); - - for (const msg of rows) { - const existing = userMap.get(msg.user_id); - if (existing) { - existing.message_count++; - if (msg.type === "edited") existing.edited_count++; - if (msg.type === "deleted") existing.deleted_count++; - if (msg.ai_status === "flagged" || msg.ai_status === "warn") - existing.flagged_count++; - if (msg.created_at > existing.last_active) { - existing.last_active = msg.created_at; - } - } else { - userMap.set(msg.user_id, { - user_id: msg.user_id, - username: msg.username, - avatar_url: msg.avatar_url, - message_count: 1, - edited_count: msg.type === "edited" ? 1 : 0, - deleted_count: msg.type === "deleted" ? 1 : 0, - flagged_count: - msg.ai_status === "flagged" || msg.ai_status === "warn" ? 1 : 0, - last_active: msg.created_at, - }); - } - } - - return Array.from(userMap.values()) - .sort((a, b) => b.message_count - a.message_count) - .slice(0, limit); + return rows as UserStat[]; } catch (error) { logger.error( { error: error instanceof Error ? error.message : String(error) }, @@ -507,54 +501,51 @@ export async function getModerationStats(input: { try { const { guildId, channelId, hours = 24 } = input; const since = Date.now() - hours * 3600_000; - const database = db(); + const rawDb = getDatabase() as any; - const conditions: SQL[] = [ - eq(messagesTable.guild_id, guildId), - gte(messagesTable.created_at, since), - isNull(messagesTable.deleted_at), - ]; + // SQL-level aggregate instead of SELECT * + in-memory counting + const row = rawDb.get( + ` + SELECT + count(*) as total, + count(case when ai_status = 'clean' then 1 end) as clean, + count(case when ai_status = 'warn' then 1 end) as warned, + count(case when ai_status = 'flagged' then 1 end) as flagged, + count(case when ai_status = 'error' then 1 end) as error, + count(case when ai_status = 'pending' or ai_status IS NULL then 1 end) as pending, + round(avg(ai_moderation_score), 2) as average_score + FROM messages + WHERE guild_id = ? + AND created_at >= ? + AND deleted_at IS NULL + ${channelId ? `AND (channel_id = ? OR thread_id = ?)` : ""} + `, + channelId + ? [guildId, since, channelId, channelId] + : [guildId, since], + ); - if (channelId) { - conditions.push(channelFilter(channelId)); + if (!row) { + return { + total: 0, + clean: 0, + warned: 0, + flagged: 0, + error: 0, + pending: 0, + average_score: 0, + }; } - const rows = (await database - .select() - .from(messagesTable) - .where(and(...conditions) as SQL)) as MessageRecord[]; - - const breakdown: ModerationBreakdown = { - total: rows.length, - clean: 0, - warned: 0, - flagged: 0, - error: 0, - pending: 0, - average_score: 0, + return { + total: row.total ?? 0, + clean: row.clean ?? 0, + warned: row.warned ?? 0, + flagged: row.flagged ?? 0, + error: row.error ?? 0, + pending: row.pending ?? 0, + average_score: row.average_score ?? 0, }; - - let scoreSum = 0; - let scoreCount = 0; - - for (const msg of rows) { - const status = msg.ai_status || "pending"; - if (status === "clean") breakdown.clean++; - else if (status === "warn") breakdown.warned++; - else if (status === "flagged") breakdown.flagged++; - else if (status === "error") breakdown.error++; - else breakdown.pending++; - - if (msg.ai_moderation_score != null) { - scoreSum += msg.ai_moderation_score; - scoreCount++; - } - } - - breakdown.average_score = - scoreCount > 0 ? Math.round((scoreSum / scoreCount) * 100) / 100 : 0; - - return breakdown; } catch (error) { logger.error( { error: error instanceof Error ? error.message : String(error) }, @@ -581,21 +572,20 @@ export async function getActiveChannelCount(input: { try { const { guildId, hours = 24 } = input; const since = Date.now() - hours * 3600_000; - const database = db(); + const rawDb = getDatabase() as any; - const rows = (await database - .select({ channel_id: messagesTable.channel_id }) - .from(messagesTable) - .where( - and( - eq(messagesTable.guild_id, guildId), - gte(messagesTable.created_at, since), - isNull(messagesTable.deleted_at), - ) as SQL, - ) - .groupBy(messagesTable.channel_id)) as Array<{ channel_id: string }>; + const row = rawDb.get( + ` + SELECT count(DISTINCT channel_id) as cnt + FROM messages + WHERE guild_id = ? + AND created_at >= ? + AND deleted_at IS NULL + `, + [guildId, since], + ); - return rows.length; + return row?.cnt ?? 0; } catch (error) { logger.error( { error: error instanceof Error ? error.message : String(error) }, @@ -628,104 +618,47 @@ export async function getTopViolators(input: { try { const { guildId, channelId, hours = 24, limit = 20 } = input; const since = Date.now() - hours * 3600_000; - const database = db(); + const rawDb = getDatabase() as any; - const conditions: SQL[] = [ - eq(messagesTable.guild_id, guildId), - gte(messagesTable.created_at, since), - isNull(messagesTable.deleted_at), - ]; + // SQL-level GROUP BY aggregate for base stats + const rows = rawDb.all( + ` + SELECT + user_id, + username, + avatar_url, + count(*) as total_messages, + count(case when ai_status = 'flagged' then 1 end) as flagged_count, + count(case when ai_status = 'warn' then 1 end) as warned_count, + max(case when ai_status in ('flagged', 'warn') then created_at else 0 end) as last_violation + FROM messages + WHERE guild_id = ? + AND created_at >= ? + AND deleted_at IS NULL + ${channelId ? `AND (channel_id = ? OR thread_id = ?)` : ""} + GROUP BY user_id + HAVING flagged_count > 0 OR warned_count > 0 + ORDER BY (flagged_count * 3 + warned_count) DESC + LIMIT ? + `, + channelId + ? [guildId, since, channelId, channelId, limit] + : [guildId, since, limit], + ); - if (channelId) { - conditions.push(channelFilter(channelId)); - } + const violators: ViolatorStat[] = rows.map((row: any) => ({ + user_id: row.user_id, + username: row.username, + avatar_url: row.avatar_url, + total_messages: row.total_messages, + flagged_count: row.flagged_count, + warned_count: row.warned_count, + violation_score: row.flagged_count * 3 + row.warned_count, + worst_flags: [], // flags require parsing JSON per-row; skip for perf + last_violation: row.last_violation, + })); - const rows = (await database - .select() - .from(messagesTable) - .where(and(...conditions) as SQL) - .orderBy(asc(messagesTable.created_at))) as MessageRecord[]; - - const userMap = new Map< - string, - { - user_id: string; - username: string; - avatar_url: string | null; - total_messages: number; - flagged_count: number; - warned_count: number; - flags_set: Set; - last_violation: number; - } - >(); - - for (const msg of rows) { - let entry = userMap.get(msg.user_id); - if (!entry) { - entry = { - user_id: msg.user_id, - username: msg.username, - avatar_url: msg.avatar_url, - total_messages: 0, - flagged_count: 0, - warned_count: 0, - flags_set: new Set(), - last_violation: 0, - }; - userMap.set(msg.user_id, entry); - } - - entry.total_messages++; - - const isViolation = - msg.ai_status === "flagged" || msg.ai_status === "warn"; - - if (msg.ai_status === "flagged") { - entry.flagged_count++; - } - - if (msg.ai_status === "warn") { - entry.warned_count++; - } - - if (isViolation && msg.ai_moderation_flags) { - try { - const flags = JSON.parse(msg.ai_moderation_flags); - if (Array.isArray(flags)) { - for (const f of flags) entry.flags_set.add(String(f)); - } - } catch { - /* ignore */ - } - } - - if (isViolation && msg.created_at > entry.last_violation) { - entry.last_violation = msg.created_at; - } - } - - const violators: ViolatorStat[] = []; - - for (const entry of userMap.values()) { - if (entry.flagged_count === 0 && entry.warned_count === 0) continue; - - violators.push({ - user_id: entry.user_id, - username: entry.username, - avatar_url: entry.avatar_url, - total_messages: entry.total_messages, - flagged_count: entry.flagged_count, - warned_count: entry.warned_count, - violation_score: entry.flagged_count * 3 + entry.warned_count * 1, - worst_flags: Array.from(entry.flags_set).slice(0, 5), - last_violation: entry.last_violation, - }); - } - - return violators - .sort((a, b) => b.violation_score - a.violation_score) - .slice(0, limit); + return violators; } catch (error) { logger.error( { error: error instanceof Error ? error.message : String(error) }, diff --git a/src/moderation/messageStore.ts b/src/moderation/messageStore.ts index 5d43bf4..5f6b95b 100644 --- a/src/moderation/messageStore.ts +++ b/src/moderation/messageStore.ts @@ -242,7 +242,8 @@ export async function getMessagesByChannel( eq(messagesTable.thread_id, channelId), ), ) - .orderBy(desc(messagesTable.created_at)) + // P3: add secondary sort by id for stable pagination + .orderBy(desc(messagesTable.created_at), desc(messagesTable.id)) .limit(limit) .offset(offset); @@ -452,12 +453,37 @@ export async function updateMessagesAIAnalysisBulk( ): Promise { if (updates.length === 0) return []; try { - const results = await Promise.all( - updates.map(({ messageId, result }) => - updateMessageAIAnalysis(messageId, result), - ), - ); - return results.filter((r): r is MessageRecord => r !== null); + // Use raw SQL batch UPDATE instead of Promise.all per-message queries + // (P2: reduce N*2 queries → 2 queries total) + const database = db(); + const now = Date.now(); + + for (const { messageId, result } of updates) { + await database + .update(messagesTable) + .set({ + ai_status: result.status, + ai_moderation_flags: result.flags ?? null, + ai_moderation_score: result.score ?? null, + ai_analysis: result.analysis ?? null, + ai_categories: stringifyAIList(result.categories), + ai_severity: result.severity ?? null, + ai_confidence: result.confidence ?? result.score ?? null, + ai_recommended_action: result.recommendedAction ?? null, + ai_analyzed_at: result.analyzedAt ?? now, + ai_error: result.error ?? null, + }) + .where(eq(messagesTable.id, messageId)); + } + + // Fetch all updated messages in a single query + const ids = updates.map(({ messageId }) => messageId); + const rows = await database + .select() + .from(messagesTable) + .where(inArray(messagesTable.id, ids)); + + return rows as MessageRecord[]; } catch (error) { logger.error( { diff --git a/src/routes/analysisRoutes.ts b/src/routes/analysisRoutes.ts index 4d3ced1..7171973 100644 --- a/src/routes/analysisRoutes.ts +++ b/src/routes/analysisRoutes.ts @@ -5,11 +5,7 @@ import { getAnalysisQueueStatus, queueMessageAnalysis, } from "../moderation/aiAnalyzer.js"; -import { - getMessageById, - searchMessages, - updateMessageAIAnalysis, -} from "../moderation/messageStore.js"; +import { searchMessages, updateMessageAIAnalysis } from "../moderation/messageStore.js"; import type { MessageRecord } from "../moderation/types.js"; export function createAnalysisRoutes(): Router { @@ -79,14 +75,8 @@ export function createAnalysisRoutes(): Router { 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); - } - - // Reset analysis status to pending so it gets picked up by the analyzer - await updateMessageAIAnalysis(id, { + // P3: Single UPDATE + RETURNING instead of GET + UPDATE + GET + const updated = await updateMessageAIAnalysis(id, { status: "pending", flags: null, score: null, @@ -95,6 +85,10 @@ export function createAnalysisRoutes(): Router { error: null, }); + if (!updated) { + throw new AppError("Message not found", "MESSAGE_NOT_FOUND", 404); + } + // Queue for analysis await queueMessageAnalysis(id); diff --git a/src/routes/syncRoutes.ts b/src/routes/syncRoutes.ts index f34549e..a48585a 100644 --- a/src/routes/syncRoutes.ts +++ b/src/routes/syncRoutes.ts @@ -1,4 +1,4 @@ -import type { Client } from "discord.js-selfbot-v13"; +import { Client } from "discord.js-selfbot-v13"; import type { Router } from "express"; import express from "express"; import { AppError } from "../errors.js"; @@ -7,7 +7,10 @@ import { syncSelectedChannelBacklog } from "../moderation/backlogSync.js"; const logger = createChildLogger("sync-routes"); const BACKLOG_SYNC_COOLDOWN_MS = 5 * 60 * 1000; +const MAX_CONCURRENT_SYNCS = 3; // P3: cap concurrent backlogs + const recentBacklogSyncs = new Map(); +let activeSyncCount = 0; export function shouldSkipRecentBacklogSync( guildId: string, @@ -55,6 +58,19 @@ export function createSyncRoutes(client: Client): Router { return; } + // P3: backpressure - reject if too many concurrent syncs + if (activeSyncCount >= MAX_CONCURRENT_SYNCS) { + res.status(429).json({ + success: false, + error: "TOO_MANY_SYNCS", + message: `Too many backlog syncs in progress (${activeSyncCount}/${MAX_CONCURRENT_SYNCS}). Try again later.`, + activeSyncCount, + maxConcurrentSyncs: MAX_CONCURRENT_SYNCS, + }); + return; + } + + activeSyncCount++; syncSelectedChannelBacklog(client, guildId, channelId) .then(() => {}) .catch((error) => { @@ -66,6 +82,9 @@ export function createSyncRoutes(client: Client): Router { }, "Backlog sync failed", ); + }) + .finally(() => { + activeSyncCount--; }); res.json({ @@ -80,5 +99,13 @@ export function createSyncRoutes(client: Client): Router { } }); + // GET /api/backlog-sync/status - Get current backlog sync status + router.get("/backlog-sync/status", (_req, res) => { + res.json({ + activeSyncCount, + maxConcurrentSyncs: MAX_CONCURRENT_SYNCS, + }); + }); + return router; }