diff --git a/.env.example b/.env.example index d46cff5..bd7b864 100644 --- a/.env.example +++ b/.env.example @@ -97,6 +97,7 @@ AI_LLM_MAX_CONCURRENT=5 # Max concurrent LLM API calls (default: AI_LLM_IMAGE_MAX_DIMENSION=1024 # Max image dimension in pixels before resize (default: 1024) AI_LLM_TEXT_BATCH_SIZE=20 # Max messages per text-only moderation batch (default: 20) AI_LLM_MEDIA_ANALYSIS_TIMEOUT_MS=60000 # Timeout in ms for media analysis calls (default: 60000) +AI_LLM_TEXT_ANALYSIS_TIMEOUT_MS=30000 # Timeout in ms for text-only analysis calls (default: 30000) # === AI Analysis Tuning === AI_ANALYSIS_DEBOUNCE_MS=500 # Debounce window for batching messages in ms (default: 500) @@ -110,11 +111,6 @@ AI_ANALYSIS_PROCESSING_TIMEOUT_MS=120000 # Conversation lock timeout in ms (defa AI_ANALYSIS_INDIVIDUAL_MAX_CONCURRENT=50 # Max concurrent individual-fallback jobs (default: 50) AI_ANALYSIS_INDIVIDUAL_CB_THRESHOLD=50 # Consecutive errors before circuit breaker trips (default: 50) -# === OpenAI Moderation (optional separate provider) === -# OPENAI_MODERATION_API_KEY= # OpenAI API key for moderation endpoint -# OPENAI_MODERATION_BASE_URL=https://api.openai.com/v1 # OpenAI moderation base URL (default) -# OPENAI_MODERATION_MODEL=omni-moderation-latest # OpenAI moderation model (default) - # === Auto-Delete === AUTO_DELETE_FLAGGED_ENABLED=true # Enable auto-deletion of flagged messages (default: true) AUTO_DELETE_FLAGGED_DRY_RUN=true # Dry-run mode: log but do not delete (default: false) diff --git a/services/discord-gateway/src/modules/ai-moderation/ai-analysis-worker.ts b/services/discord-gateway/src/modules/ai-moderation/ai-analysis-worker.ts index faf4c7f..878ba69 100644 --- a/services/discord-gateway/src/modules/ai-moderation/ai-analysis-worker.ts +++ b/services/discord-gateway/src/modules/ai-moderation/ai-analysis-worker.ts @@ -5,11 +5,15 @@ * * ## Pipeline * - * Message → LLM evaluator (with conversation context + media evidence) + * Message batch → runModerationAnalysis (orchestrator) → LLM evaluator + * (with conversation context + media evidence) → per-message verdicts * - * Every message is judged by the LLM — there is no regex/heuristic - * pre-classification. A failed LLM call yields an explicit "error" status - * (never a heuristic verdict), and the recovery worker retries it later. + * The orchestrator splits text-only vs media internally and runs both + * paths in parallel with ONE LLM call per sub-batch — a 20-message text + * batch costs 1 LLM call, not 20. Every message is judged by the LLM — + * there is no regex/heuristic pre-classification. A failed LLM call yields + * an explicit "error" status (never a heuristic verdict), and the recovery + * worker retries it later. */ import { createChildLogger } from "@/shared/logger/index"; @@ -115,7 +119,10 @@ export default async function workerRouter( if (!config.AI_LLM_API_KEY) { const errorMsg = "AI_LLM_API_KEY is missing from environment. Worker cannot process moderation requests without credentials."; - logger.error({ error: errorMsg }, "AI_LLM_API_KEY is missing from environment"); + logger.error( + { error: errorMsg }, + "AI_LLM_API_KEY is missing from environment", + ); if (job.type === "batch") { return { @@ -172,60 +179,9 @@ export default async function workerRouter( } // --------------------------------------------------------------------------- -// Single-pass LLM pipeline +// Result normalization // --------------------------------------------------------------------------- -/** - * Runs the LLM moderation analysis on a single message. - * - * The LLM verdict IS the result — confidence, severity, flags and - * analysis all come from the model. On failure the message is marked - * "error" (explicit, retryable) instead of receiving a heuristic verdict. - */ -async function runLLMAnalysis( - message: MessageRecord, - contextText: string, - attachments: Awaited>, -): Promise { - try { - const moderationResult = await runModerationAnalysis({ - targets: [message], - contextText, - attachments, - }); - - if (moderationResult.results.length === 0) { - return buildFallbackResult(message.id, "No LLM result returned"); - } - - const llmResult = moderationResult.results[0] as unknown as AnalysisResult; - if (llmResult.status === "error") { - return llmResult; - } - - return { - messageId: llmResult.messageId, - status: llmResult.status ?? "clean", - flags: llmResult.flags ?? [], - categories: llmResult.categories ?? [], - severity: llmResult.severity ?? "none", - confidence: normalizeConfidence(llmResult.confidence), - recommendedAction: llmResult.recommendedAction ?? "none", - score: llmResult.score ?? 0, - analysis: - llmResult.analysis?.trim() || - buildFallbackAnalysis(message, llmResult.status ?? "clean"), - }; - } catch (error) { - const errorMsg = error instanceof Error ? error.message : String(error); - logger.warn( - { messageId: message.id, error: errorMsg }, - "LLM analysis failed for message", - ); - return buildFallbackResult(message.id, errorMsg); - } -} - /** Clamp LLM-provided confidence to [0, 1]; default by status when missing. */ function normalizeConfidence(raw: number | undefined | null): number { if (typeof raw === "number" && Number.isFinite(raw)) { @@ -239,10 +195,7 @@ function normalizeConfidence(raw: number | undefined | null): number { * Quotes the actual message content so the log still explains WHAT was said * instead of a bare template like "Tidak ada indikasi pelanggaran." */ -function buildFallbackAnalysis( - message: MessageRecord, - status: string, -): string { +function buildFallbackAnalysis(message: MessageRecord, status: string): string { const raw = (message.edited_content ?? message.content ?? "").trim(); const snippet = raw.length > 120 ? `${raw.slice(0, 120).trimEnd()}…` : raw; @@ -273,8 +226,31 @@ function buildFallbackResult( }; } +/** Normalize one orchestrator result into the worker's AnalysisResult shape. */ +function normalizeResult( + result: AnalysisResult, + message: MessageRecord | undefined, +): AnalysisResult { + const status = result.status ?? "clean"; + return { + messageId: result.messageId, + status, + flags: result.flags ?? [], + categories: result.categories ?? [], + severity: result.severity ?? "none", + confidence: normalizeConfidence(result.confidence), + recommendedAction: result.recommendedAction ?? "none", + score: result.score ?? 0, + analysis: + result.analysis?.trim() || + (message + ? buildFallbackAnalysis(message, status) + : buildFallbackResult(result.messageId, "Missing message").analysis), + }; +} + // --------------------------------------------------------------------------- -// Batch handler +// Batch handler — ONE orchestrator call for the whole batch // --------------------------------------------------------------------------- async function processBatch(job: { @@ -286,7 +262,7 @@ async function processBatch(job: { const firstMessage = messages[0]; if (!firstMessage) return { ok: true, conversationKey, rows: [] }; - // Fetch context + // Fetch context + attachments ONCE for the whole batch. const contextBefore = await messageStore.getConversationContextBefore({ channelId: firstMessage.channel_id, threadId: firstMessage.thread_id, @@ -301,31 +277,31 @@ async function processBatch(job: { }); const contextText = contextLines.join("\n"); - // Fetch attachments const targetIds = messages.map((m) => m.id); const contextIds = contextBefore.map((m) => m.id); - const allMessageIds = [...targetIds, ...contextIds]; - const attachments = - await messageStore.getAttachmentsForMessages(allMessageIds); + const attachments = await messageStore.getAttachmentsForMessages([ + ...targetIds, + ...contextIds, + ]); - // Run LLM analysis for each message - const analysisResults = await Promise.all( - messages.map(async (msg) => { - try { - return await runLLMAnalysis(msg, contextText, attachments); - } catch (error) { - const errorMsg = error instanceof Error ? error.message : String(error); - logger.error( - { messageId: msg.id, error: errorMsg }, - "LLM analysis failed for message", - ); - return buildFallbackResult(msg.id, errorMsg); - } - }), + // The orchestrator handles text/media split + caching + parallel paths + // internally, so a 20-message batch = 1 text LLM call (+1 media call + // when media is present), not N per-message calls. + const moderationResult = await runModerationAnalysis({ + targets: messages, + contextText, + attachments, + }); + + const results = moderationResult.results.map((r) => + normalizeResult( + r as unknown as AnalysisResult, + messages.find((m) => m.id === r.messageId), + ), ); // Save results to DB - const updates = analysisResults.map((result) => ({ + const updates = results.map((result) => ({ messageId: result.messageId, result: { status: result.status, @@ -393,8 +369,28 @@ async function processIndividual(job: { ]); try { - const result = await runLLMAnalysis(message, contextText, attachments); - return { ok: true, results: [result] }; + const moderationResult = await runModerationAnalysis({ + targets: [message], + contextText, + attachments, + }); + + if (moderationResult.results.length === 0) { + return { + ok: true, + results: [buildFallbackResult(message.id, "No LLM result returned")], + }; + } + + const llmResult = moderationResult.results[0] as unknown as AnalysisResult; + if (llmResult.status === "error") { + return { ok: true, results: [llmResult] }; + } + + return { + ok: true, + results: [normalizeResult(llmResult, message)], + }; } catch (error) { const errorMsg = error instanceof Error ? error.message : String(error); logger.error( diff --git a/services/discord-gateway/src/modules/ai-moderation/aiAnalyzer.ts b/services/discord-gateway/src/modules/ai-moderation/aiAnalyzer.ts index 93fd969..0501177 100644 --- a/services/discord-gateway/src/modules/ai-moderation/aiAnalyzer.ts +++ b/services/discord-gateway/src/modules/ai-moderation/aiAnalyzer.ts @@ -1,5 +1,5 @@ -import { createChildLogger } from "@/shared/logger/index"; import type { Client } from "discord.js-selfbot-v13"; +import { createChildLogger } from "@/shared/logger/index"; import { config } from "../../shared/config/config.js"; import type { EventBroadcaster } from "../event-broadcaster/index.js"; import { messageStore } from "../message-capture/messageStore.js"; @@ -33,9 +33,17 @@ import { setModerationClient, setSharedEventBroadcaster, } from "./moderationState.js"; +import { deleteExpiredQdrantPoints } from "./qdrantClient.js"; +import { pruneExpiredTexts } from "./textCacheStore.js"; const logger = createChildLogger("ai-analyzer"); +// --------------------------------------------------------------------------- +// Cache hygiene (expired verdict sweep) +// --------------------------------------------------------------------------- +const CACHE_PRUNE_INTERVAL_MS = 6 * 60 * 60 * 1000; // every 6 hours +let lastCachePruneAt = 0; + // --------------------------------------------------------------------------- // Re-exports from sub-modules (preserving original public API) // --------------------------------------------------------------------------- @@ -133,6 +141,26 @@ export function startPendingAIAnalysisWorker( .catch(console.error); setInterval(() => { + // [D] Periodic cache hygiene: purge expired moderation verdicts from + // Postgres and Qdrant. Expired entries are never reused (filters check + // expires_at) but accumulate forever without this sweep. + const now = Date.now(); + if (now - lastCachePruneAt >= CACHE_PRUNE_INTERVAL_MS) { + lastCachePruneAt = now; + Promise.all([pruneExpiredTexts(), deleteExpiredQdrantPoints()]) + .then(([pgDeleted, qdDeleted]) => { + if (pgDeleted > 0 || qdDeleted > 0) { + logger.info( + { pgDeleted, qdDeleted }, + "Expired moderation cache pruned", + ); + } + }) + .catch((err: unknown) => { + logger.warn({ error: String(err) }, "Moderation cache prune failed"); + }); + } + messageStore.revertStuckProcessingMessages(300000).catch((err: unknown) => { logger.error( { error: String(err) }, diff --git a/services/discord-gateway/src/modules/ai-moderation/archive/aiAnalysisWorker.ts b/services/discord-gateway/src/modules/ai-moderation/archive/aiAnalysisWorker.ts deleted file mode 100644 index 0fe7f4d..0000000 --- a/services/discord-gateway/src/modules/ai-moderation/archive/aiAnalysisWorker.ts +++ /dev/null @@ -1,327 +0,0 @@ -import { createChildLogger } from "@/shared/logger/index"; -import { config } from "../../shared/config/config.js"; -import { initializeDatabase } from "../../shared/database/drizzle.js"; -import { extractMessageMediaEvidence } from "../message-capture/messageMetadata.js"; -import { messageStore } from "../message-capture/messageStore.js"; -import type { - AnalysisResult, - MessageRecord, -} from "../message-capture/types.js"; -import { buildConversationContext } from "./conversationContext.js"; -import { runModerationAnalysis } from "./moderationOrchestrator.js"; -import { runSimpleTextFallback } from "./simpleFallback.js"; - -const logger = createChildLogger("aiAnalysisWorker"); - -let dbInitialized = false; -let dbInitPromise: Promise | null = null; - -async function ensureDb() { - if (dbInitialized) return; - if (!dbInitPromise) { - dbInitPromise = initializeDatabase().then(() => { - dbInitialized = true; - }); - } - await dbInitPromise; -} - -// --------------------------------------------------------------------------- -// Job types — the default export routes on `type` -// --------------------------------------------------------------------------- - -type WorkerJob = - | { type: "batch"; conversationKey: string; messages: MessageRecord[] } - | { type: "individual"; message: MessageRecord; skipNormalAnalysis: boolean }; - -type BatchOkResponse = { - ok: true; - conversationKey: string; - rows: MessageRecord[]; -}; -type BatchErrorResponse = { - ok: false; - conversationKey: string; - rows: MessageRecord[]; - error: string; -}; -type IndividualOkResponse = { ok: true; results: AnalysisResult[] }; -type IndividualErrorResponse = { - ok: false; - results: AnalysisResult[]; - error: string; -}; - -type WorkerResponse = - | BatchOkResponse - | BatchErrorResponse - | IndividualOkResponse - | IndividualErrorResponse; - -/** - * Default export — Piscina worker entry point. - * Routes to the correct handler based on `type` field. - */ -export default async function workerRouter( - job: WorkerJob, -): Promise { - if (!config.AI_LLM_API_KEY) { - const errorMsg = - "AI_LLM_API_KEY is missing from environment. Worker cannot process moderation requests without credentials."; - logger.error( - { error: errorMsg }, - "AI_LLM_API_KEY is missing from environment", - ); - - if (job.type === "batch") { - return { - ok: false, - conversationKey: job.conversationKey, - rows: [], - error: errorMsg, - }; - } - return { ok: false, results: [], error: errorMsg }; - } - - try { - await ensureDb(); - } catch (dbError) { - const msg = dbError instanceof Error ? dbError.message : String(dbError); - if (job.type === "batch") { - return { - ok: false, - conversationKey: job.conversationKey, - rows: [], - error: `Database init failed: ${msg}`, - }; - } - return { ok: false, results: [], error: `Database init failed: ${msg}` }; - } - - try { - if (job.type === "batch") { - return await processBatch(job); - } - return await processIndividual(job); - } catch (error) { - const errorMessage = error instanceof Error ? error.message : String(error); - const errorStack = error instanceof Error ? error.stack : undefined; - logger.error( - { type: job.type, error: errorMessage, stack: errorStack }, - "Worker job failed", - ); - if (job.type === "batch") { - return { - ok: false, - conversationKey: job.conversationKey, - rows: [], - error: errorMessage, - }; - } - return { ok: false, results: [], error: errorMessage }; - } -} - -// --------------------------------------------------------------------------- -// Batch handler -// --------------------------------------------------------------------------- - -async function processBatch(job: { - type: "batch"; - conversationKey: string; - messages: MessageRecord[]; -}): Promise { - const { conversationKey, messages } = job; - const firstMessage = messages[0]; - if (!firstMessage) return { ok: true, conversationKey, rows: [] }; - - const contextBefore = await messageStore.getConversationContextBefore({ - channelId: firstMessage.channel_id, - threadId: firstMessage.thread_id, - beforeCreatedAt: firstMessage.created_at, - limit: config.AI_ANALYSIS_CONTEXT_MESSAGE_LIMIT, - }); - - const contextLines = buildConversationContext({ - contextBefore, - targets: messages, - maxTokens: config.AI_ANALYSIS_MAX_CONTEXT_TOKENS, - }); - - const targetIds = messages.map((m) => m.id); - const contextIds = contextBefore.map((m) => m.id); - const allMessageIds = [...targetIds, ...contextIds]; - const attachments = - await messageStore.getAttachmentsForMessages(allMessageIds); - - // ── Split: text-only vs media ────────────────────────────────────── - // Text-only analysis runs fast (single LLM call, no vision). - // Media analysis is slow (download + vision → LLM). - // Messages with BOTH text and media go into both arrays: - // - text batch → analyzes the text content immediately - // - media batch → analyzes images/video when ready - // By splitting here, text results are saved to DB immediately - // instead of waiting for media downloads to finish. - // ──────────────────────────────────────────────────────────────────── - const textOnly: MessageRecord[] = []; - const media: MessageRecord[] = []; - - for (const msg of messages) { - const meta = msg.metadata - ? extractMessageMediaEvidence(msg.metadata) - : null; - if ( - meta && - (meta.attachments.length > 0 || - meta.stickers.length > 0 || - meta.embeds.length > 0) - ) { - media.push(msg); - // If the message also has text content, analyze it in the text batch too - const rawContent = msg.edited_content ?? msg.content; - if (rawContent.trim().length > 0) { - textOnly.push(msg); - } - } else { - textOnly.push(msg); - } - } - - const allRows: MessageRecord[] = []; - - // ── Parallel: text-only + media analysis run concurrently ────────── - // Text-only → fast LLM call. Media → download + vision + LLM. - // Running both in parallel means media downloads overlap with text LLM call. - // Each path saves to DB as soon as its own results are ready. - // ──────────────────────────────────────────────────────────────────── - const textPromise = - textOnly.length > 0 - ? runModerationAnalysis({ - targets: textOnly, - contextText: contextLines.join("\n"), - attachments, - }).then((result) => { - const updates = result.results.map((analysisResult) => ({ - messageId: analysisResult.messageId, - result: { - status: analysisResult.status, - flags: JSON.stringify(analysisResult.flags), - score: analysisResult.score, - analysis: analysisResult.analysis, - categories: analysisResult.categories, - severity: analysisResult.severity, - confidence: analysisResult.confidence, - recommendedAction: analysisResult.recommendedAction, - analyzedAt: Date.now(), - error: null, - }, - })); - if (updates.length > 0) { - return messageStore - .updateMessagesAIAnalysisBulk(updates) - .then((rows) => { - allRows.push(...rows); - logger.info( - { count: updates.length, conversationKey }, - "Text-only batch saved — media analysis still in progress", - ); - }); - } - }) - : Promise.resolve(); - - const mediaPromise = - media.length > 0 - ? runModerationAnalysis({ - targets: media, - contextText: contextLines.join("\n"), - attachments, - }).then((result) => { - const updates = result.results.map((analysisResult) => ({ - messageId: analysisResult.messageId, - result: { - status: analysisResult.status, - flags: JSON.stringify(analysisResult.flags), - score: analysisResult.score, - analysis: analysisResult.analysis, - categories: analysisResult.categories, - severity: analysisResult.severity, - confidence: analysisResult.confidence, - recommendedAction: analysisResult.recommendedAction, - analyzedAt: Date.now(), - error: null, - }, - })); - if (updates.length > 0) { - return messageStore - .updateMessagesAIAnalysisBulk(updates) - .then((rows) => { - allRows.push(...rows); - }); - } - }) - : Promise.resolve(); - - // Wait for both to complete - await Promise.all([textPromise, mediaPromise]); - - logger.info( - { - total: messages.length, - textOnly: textOnly.length, - media: media.length, - saved: allRows.length, - }, - "Batch analysis complete", - ); - - return { ok: true, conversationKey, rows: allRows }; -} - -// --------------------------------------------------------------------------- -// Individual fallback handler (offloaded from main thread) -// --------------------------------------------------------------------------- - -async function processIndividual(job: { - type: "individual"; - message: MessageRecord; - skipNormalAnalysis: boolean; -}): Promise { - const { message, skipNormalAnalysis } = job; - - const contextBefore = await messageStore.getConversationContextBefore({ - channelId: message.channel_id, - threadId: message.thread_id, - beforeCreatedAt: message.created_at, - limit: config.AI_ANALYSIS_CONTEXT_MESSAGE_LIMIT, - }); - - const contextLines = buildConversationContext({ - contextBefore, - targets: [message], - maxTokens: config.AI_ANALYSIS_MAX_CONTEXT_TOKENS, - }); - - const contextIds = contextBefore.map((m) => m.id); - const attachments = await messageStore.getAttachmentsForMessages([ - message.id, - ...contextIds, - ]); - - let results: AnalysisResult[]; - - if (skipNormalAnalysis) { - const simpleResult = await runSimpleTextFallback(message); - results = [simpleResult]; - } else { - const moderationResult = await runModerationAnalysis({ - targets: [message], - contextText: contextLines.join("\n"), - attachments, - }); - results = moderationResult.results; - } - - return { ok: true, results }; -} diff --git a/services/discord-gateway/src/modules/ai-moderation/llmCaller.ts b/services/discord-gateway/src/modules/ai-moderation/llmCaller.ts index 719a1c1..4ed3eb2 100644 --- a/services/discord-gateway/src/modules/ai-moderation/llmCaller.ts +++ b/services/discord-gateway/src/modules/ai-moderation/llmCaller.ts @@ -9,9 +9,10 @@ * * Both sides now import from this module instead. */ + +import type { ChatCompletion } from "openai/resources/chat/completions"; import { createChildLogger } from "@/shared/logger/index"; import { delay, retryWithBackoff } from "@/shared/utils/index"; -import type { ChatCompletion } from "openai/resources/chat/completions"; import { config } from "../../shared/config/config.js"; import type { AnalysisResult } from "../message-capture/types.js"; import { llmChat } from "./llmClient.js"; @@ -27,11 +28,24 @@ export interface RetryState { lastInvalidContent: string | null; } +/** + * Content builder contract: produces the moderation prompt split into + * SYSTEM (rules / output schema / context — stable) and USER (the actual + * `` payload). Kept as two roles so routers and + * providers that treat system messages differently get the correct framing. + */ +export interface ModerationPromptContent { + system: string; + user: string; +} + // --------------------------------------------------------------------------- // Shared LLM call + parse + fallback helper // --------------------------------------------------------------------------- export async function callModerationLLM( - buildContent: (state: RetryState) => Promise, + buildContent: ( + state: RetryState, + ) => Promise, targetIds: string[], label: string, signal?: AbortSignal, @@ -52,8 +66,15 @@ export async function callModerationLLM( async () => { try { const content = await buildContent(state); + const messages = + typeof content === "string" + ? [{ role: "user" as const, content }] + : [ + { role: "system" as const, content: content.system }, + { role: "user" as const, content: content.user }, + ]; const completion = await llmChat({ - messages: [{ role: "user", content }], + messages, max_tokens: 16384, jsonResponse: { type: "json_object" }, retries: 0, @@ -133,6 +154,23 @@ export async function callModerationLLM( ); parsed = analysis.parsed; result = analysis.result; + + // [I] Token usage accounting — surface provider-reported usage per batch + // so cost per channel/guild can be tracked (routers bill per token). + const usage = result?.usage; + if (usage && (usage.prompt_tokens || usage.completion_tokens)) { + log.info( + { + label, + targetIds, + model: config.AI_LLM_MODEL, + prompt_tokens: usage.prompt_tokens, + completion_tokens: usage.completion_tokens, + total_tokens: usage.total_tokens, + }, + `LLM usage (${label})`, + ); + } } catch (err) { if (err instanceof Error && err.name === "AbortError") throw err; diff --git a/services/discord-gateway/src/modules/ai-moderation/llmClient.ts b/services/discord-gateway/src/modules/ai-moderation/llmClient.ts index c7af90e..3969423 100644 --- a/services/discord-gateway/src/modules/ai-moderation/llmClient.ts +++ b/services/discord-gateway/src/modules/ai-moderation/llmClient.ts @@ -6,10 +6,10 @@ * defaults are maintained in one place. */ -import { createChildLogger } from "@/shared/logger/index"; -import { retryWithBackoff } from "@/shared/utils/index"; import OpenAI from "openai"; import pLimit from "p-limit"; +import { createChildLogger } from "@/shared/logger/index"; +import { retryWithBackoff } from "@/shared/utils/index"; import { config } from "../../shared/config/config.js"; const log = createChildLogger("llm-client"); @@ -245,39 +245,13 @@ export async function llmChat( ); } -/** - * Convenience for the legacy text-only badword detection call in - * `indonesianTextNormalizer`. Returns parsed flags or []. - */ -export async function llmDetectBadwords(text: string): Promise { - const completion = await llmChat({ - messages: [ - { - role: "user", - content: - "Deteksi kata kasar / pelanggaran ringan dari teks Indonesia berikut. " + - 'Balas hanya JSON object dengan format {"flags":[...]} dan gunakan hanya flag valid ini: ' + - Array.from(VALID_PRIMARY_AI_FLAGS).join(", ") + - ". Jika tidak ada pelanggaran, flags harus array kosong. Teks: " + - text, - }, - ], - max_tokens: 200, - temperature: 0.1, - top_p: 0.9, - jsonResponse: { type: "json_object" }, - retries: 2, - }); - - if (!completion) return []; - const content = completion.choices[0]?.message?.content?.trim(); - if (!content) return []; - return extractFlagsFromContent(content); -} - /** * Convenience for vision (image/sticker/emoji) analysis. * Returns the raw completion content (trimmed) or null. + * + * NOTE: retries are disabled here on purpose — visionAnalyzer.ts already + * wraps this call in its own 3-attempt loop with exponential backoff. + * A second retry layer would multiply worst-case API calls (3×3=9/image). */ export async function llmVision( promptText: string, @@ -297,91 +271,9 @@ export async function llmVision( max_tokens: 500, temperature: 0.1, top_p: 0.9, - retries: 2, + retries: 0, }); if (!completion) return null; return completion.choices[0]?.message?.content?.trim() ?? null; } - -// --------------------------------------------------------------------------- -// Flag extraction (reused from indonesianTextNormalizer) -// --------------------------------------------------------------------------- - -const VALID_PRIMARY_AI_FLAGS = new Set([ - "spam", - "hate_speech", - "sara", - "hoaks", - "harassment", - "vulgar_language", - "sexual_content", - "sexual_deviation", - "violence", - "self_harm", - "doxxing", - "scam", - "misinformation", - "nsfw_image", - "gore_image", - "illegal_content", - "gambling", - "drugs", - "child_safety", - "financial_scam", - "religious_insult", - "self_promo", - "conflict_instigation", - "offensive_username", - "potential_evasion", - "unclear_context", -]); - -function normalizeFlag(value: string): string | null { - const lower = value - .trim() - .toLowerCase() - .replace(/[\s-]+/g, "_"); - if (!lower) return null; - if (VALID_PRIMARY_AI_FLAGS.has(lower)) return lower; - return null; -} - -function extractFlagsFromContent(content: string): string[] { - const flags = new Set(); - let parsed: unknown; - try { - parsed = JSON.parse(content); - } catch { - parsed = null; - } - - const addValue = (v: unknown) => { - if (typeof v !== "string") return; - const n = normalizeFlag(v); - if (n) flags.add(n); - }; - - if (Array.isArray(parsed)) { - for (const item of parsed) addValue(item); - } else if (parsed && typeof parsed === "object") { - const obj = parsed as Record; - for (const key of ["flags", "categories", "badwords"]) { - const val = obj[key]; - if (Array.isArray(val)) { - for (const item of val) addValue(item); - } else { - addValue(val); - } - } - } - - if (flags.size > 0) return Array.from(flags); - - const lower = content.toLowerCase(); - for (const flag of VALID_PRIMARY_AI_FLAGS) { - if (lower.includes(flag)) flags.add(flag); - } - - return Array.from(flags); -} diff --git a/services/discord-gateway/src/modules/ai-moderation/mediaBatchProcessor.ts b/services/discord-gateway/src/modules/ai-moderation/mediaBatchProcessor.ts index 1a57005..01043de 100644 --- a/services/discord-gateway/src/modules/ai-moderation/mediaBatchProcessor.ts +++ b/services/discord-gateway/src/modules/ai-moderation/mediaBatchProcessor.ts @@ -13,9 +13,9 @@ import type { MessageRecord, } from "../message-capture/types.js"; import { getChannelCulture } from "./channelCultureStore.js"; -import { prepareMediaMessage } from "./mediaAnalysisClient.js"; import type { RetryState } from "./llmCaller.js"; import { callModerationLLM } from "./llmCaller.js"; +import { prepareMediaMessage } from "./mediaAnalysisClient.js"; import { buildSystemPrompt as buildSystemPromptModular } from "./moderationPrompt.js"; import { buildCorrectedFewShotExamples } from "./textBatchProcessor.js"; @@ -65,7 +65,7 @@ export async function runMediaBatch( }); const messagesBlock = prepared.map((p) => p.messageBlock).join("\n"); - const userContent = `${systemText}\n\n\n${messagesBlock}\n`; + const userContent = `\n${messagesBlock}\n`; const perMsgTimeout = config.AI_LLM_MEDIA_ANALYSIS_TIMEOUT_MS ?? 60000; const batchTimeout = Math.min( @@ -79,7 +79,7 @@ export async function runMediaBatch( try { const result = await callModerationLLM( - async (_state: RetryState) => userContent, + async (_state: RetryState) => ({ system: systemText, user: userContent }), targetIds, `media-batch:${targetIds.length}msgs`, abortController.signal, diff --git a/services/discord-gateway/src/modules/ai-moderation/moderationOrchestrator.ts b/services/discord-gateway/src/modules/ai-moderation/moderationOrchestrator.ts index 58cd625..a8ffe5c 100644 --- a/services/discord-gateway/src/modules/ai-moderation/moderationOrchestrator.ts +++ b/services/discord-gateway/src/modules/ai-moderation/moderationOrchestrator.ts @@ -12,15 +12,19 @@ import type { AttachmentRecord, MessageRecord, } from "../message-capture/types.js"; +import { embedTexts, isEmbeddingEnabled } from "./embeddingClient.js"; import { hasMediaContent } from "./mediaAnalysisClient.js"; import { runMediaBatch } from "./mediaBatchProcessor.js"; +import { isQdrantConfigured, searchQdrantBatch } from "./qdrantClient.js"; +import { logCacheEvent } from "./responseLogger.js"; import { initSearxngCache } from "./searxngSearch.js"; import { runTextOnlyBatch } from "./textBatchProcessor.js"; -import { embedText, isEmbeddingEnabled } from "./embeddingClient.js"; import { findSimilarTextModeration, getCachedTextModeration, + makeModerationContextKey, makeTextModerationCacheKey, + parseQdrantVerdict, setCachedTextModeration, } from "./textCacheStore.js"; @@ -47,6 +51,13 @@ export interface ModerationOutput { /** * Runs LLM-based moderation analysis on messages. * Splits text-only vs media, runs both paths in parallel, applies caching. + * + * Cache strategy (two-phase, batched): + * 1. Exact-hash lookups (no API) — key is content + conversation context + * (channel/thread) because LLM verdicts depend on context. + * 2. Semantic near-duplicate lookup — ONE embeddings call for all uncached + * text targets, then ONE Qdrant batch search (index-aligned), instead of + * N sequential embed→search round-trips. */ export async function runModerationAnalysis( input: ModerationInput, @@ -56,10 +67,11 @@ export async function runModerationAnalysis( initSearxngCache(config.REDIS_URL); if (!targets.length) throw new Error("No targets provided for analysis"); - // Per-user moderation cache check (text-only) + // ── Phase 1: exact-hash cache (per conversation context) ──────────────── const cacheHits: AnalysisResult[] = []; const uncachedTargets: MessageRecord[] = []; - const seenCacheKeys = new Set(); + // cacheKey → result for identical-content dedupe within one batch + const hitByKey = new Map(); // Embedding per exact cache key — computed once during lookup, reused // when the fresh LLM verdict is written back to the semantic cache. const embeddingsByKey = new Map(); @@ -77,17 +89,16 @@ export async function runModerationAnalysis( continue; } - const cacheKey = makeTextModerationCacheKey(rawContent); - if (seenCacheKeys.has(cacheKey)) { - const previousHit = cacheHits.find((h) => h.messageId !== target.id); - if (previousHit) { - cacheHits.push({ ...previousHit, messageId: target.id }); - } else { - uncachedTargets.push(target); - } + const cacheKey = makeTextModerationCacheKey( + rawContent, + makeModerationContextKey(target), + ); + const seen = hitByKey.get(cacheKey); + if (seen) { + // Same content already resolved this batch — reuse the verdict. + cacheHits.push({ ...seen, messageId: target.id }); continue; } - seenCacheKeys.add(cacheKey); try { const cached = await getCachedTextModeration(cacheKey); @@ -122,7 +133,7 @@ export async function runModerationAnalysis( "Cache entry contains error artifact — treating as miss", ); } else { - cacheHits.push({ + const hit: AnalysisResult = { messageId: target.id, status: cached.status, flags: cached.flags, @@ -135,7 +146,10 @@ export async function runModerationAnalysis( cached.recommendedAction as AnalysisResult["recommendedAction"], policyVersion: "cached-user-moderation-2026-06", evidence: [], - } as AnalysisResult); + }; + cacheHits.push(hit); + hitByKey.set(cacheKey, hit); + logCacheEvent("hit", cacheKey, "text"); continue; } } @@ -143,47 +157,130 @@ export async function runModerationAnalysis( /* proceed */ } - // Semantic cache: reuse verdicts for near-duplicate text (requires the - // configured embedding model). Only non-trivial text-only messages - // qualify — media and empty text never take this path. - if (isEmbeddingEnabled() && rawContent.trim().length >= 5) { - const embedding = await embedText(rawContent); - if (embedding) { - embeddingsByKey.set(cacheKey, embedding); - const semantic = await findSimilarTextModeration( - embedding, - config.AI_LLM_EMBEDDING_MIN_SIMILARITY, - config.AI_LLM_EMBEDDING_MAX_CANDIDATES, + uncachedTargets.push(target); + } + + // ── Phase 2: semantic cache — batched (one embed call + one Qdrant + // batch search for ALL uncached text targets) ───────────────────────── + if (isEmbeddingEnabled()) { + const semanticCandidates = uncachedTargets + .map((t) => ({ + target: t, + cacheKey: makeTextModerationCacheKey( + t.edited_content ?? t.content, + makeModerationContextKey(t), + ), + })) + .filter(({ target }) => { + const raw = (target.edited_content ?? target.content).trim(); + if (raw.length < 5) return false; + if (hasMediaContent(target, attachments)) return false; + return !hitByKey.has( + makeTextModerationCacheKey(raw, makeModerationContextKey(target)), ); - if (semantic) { - log.debug( - { - messageId: target.id, - similarity: Number(semantic.similarity.toFixed(4)), - status: semantic.status, - }, - "Semantic moderation cache hit — reusing stored verdict", + }); + + if (semanticCandidates.length > 0) { + const texts = semanticCandidates.map( + ({ target }) => target.edited_content ?? target.content, + ); + const embeddings = await embedTexts(texts); + if (embeddings && embeddings.length === texts.length) { + // index-aligned with semanticCandidates + for (let i = 0; i < semanticCandidates.length; i++) { + const { target, cacheKey } = semanticCandidates[i]; + embeddingsByKey.set(cacheKey, embeddings[i]); + } + + if (isQdrantConfigured()) { + const batchHits = await searchQdrantBatch( + embeddings, + config.AI_LLM_EMBEDDING_MAX_CANDIDATES, + config.AI_LLM_EMBEDDING_MIN_SIMILARITY, ); - cacheHits.push({ - messageId: target.id, - status: semantic.status, - flags: semantic.flags, - score: semantic.score, - analysis: semantic.analysis, - categories: semantic.categories, - severity: semantic.severity as AnalysisResult["severity"], - confidence: semantic.confidence, - recommendedAction: - semantic.recommendedAction as AnalysisResult["recommendedAction"], - policyVersion: "semantic-cache-2026-07", - evidence: [], - } as AnalysisResult); - continue; + for (let i = 0; i < semanticCandidates.length; i++) { + const { target, cacheKey } = semanticCandidates[i]; + const hits = batchHits[i] ?? []; + if (hits.length === 0) continue; + const verdict = parseQdrantVerdict(hits[0].payload, hits[0].score); + if (!verdict) continue; + log.debug( + { + messageId: target.id, + similarity: Number(verdict.similarity.toFixed(4)), + status: verdict.status, + }, + "Semantic moderation cache hit — reusing stored verdict", + ); + const hit: AnalysisResult = { + messageId: target.id, + status: verdict.status, + flags: verdict.flags, + score: verdict.score, + analysis: verdict.analysis, + categories: verdict.categories, + severity: verdict.severity as AnalysisResult["severity"], + confidence: verdict.confidence, + recommendedAction: + verdict.recommendedAction as AnalysisResult["recommendedAction"], + policyVersion: "semantic-cache-2026-07", + evidence: [], + }; + cacheHits.push(hit); + hitByKey.set(cacheKey, hit); + logCacheEvent("hit", cacheKey, "text"); + } + } else { + // Legacy Postgres fallback path (no Qdrant): per-candidate scan. + for (let i = 0; i < semanticCandidates.length; i++) { + const { target, cacheKey } = semanticCandidates[i]; + const semantic = await findSimilarTextModeration( + embeddings[i], + config.AI_LLM_EMBEDDING_MIN_SIMILARITY, + config.AI_LLM_EMBEDDING_MAX_CANDIDATES, + ); + if (!semantic) continue; + log.debug( + { + messageId: target.id, + similarity: Number(semantic.similarity.toFixed(4)), + status: semantic.status, + }, + "Semantic moderation cache hit (PG fallback) — reusing stored verdict", + ); + const hit: AnalysisResult = { + messageId: target.id, + status: semantic.status, + flags: semantic.flags, + score: semantic.score, + analysis: semantic.analysis, + categories: semantic.categories, + severity: semantic.severity as AnalysisResult["severity"], + confidence: semantic.confidence, + recommendedAction: + semantic.recommendedAction as AnalysisResult["recommendedAction"], + policyVersion: "semantic-cache-2026-07", + evidence: [], + }; + cacheHits.push(hit); + hitByKey.set(cacheKey, hit); + logCacheEvent("hit", cacheKey, "text"); + } + } + + // Drop semantic hits from the LLM work queue. + for (let i = uncachedTargets.length - 1; i >= 0; i--) { + const t = uncachedTargets[i]; + const key = makeTextModerationCacheKey( + t.edited_content ?? t.content, + makeModerationContextKey(t), + ); + if (hitByKey.has(key)) { + uncachedTargets.splice(i, 1); + } } } } - - uncachedTargets.push(target); } if (cacheHits.length > 0) { @@ -248,7 +345,10 @@ export async function runModerationAnalysis( continue; } - const cacheKey = makeTextModerationCacheKey(rawContent); + const cacheKey = makeTextModerationCacheKey( + rawContent, + makeModerationContextKey(target), + ); setCachedTextModeration( cacheKey, { diff --git a/services/discord-gateway/src/modules/ai-moderation/prompts/text-analysis.ts b/services/discord-gateway/src/modules/ai-moderation/prompts/text-analysis.ts deleted file mode 100644 index 961225b..0000000 --- a/services/discord-gateway/src/modules/ai-moderation/prompts/text-analysis.ts +++ /dev/null @@ -1,11 +0,0 @@ -/** - * Text analysis prompt constants and helpers for LLM moderation. - * - * Contains shared types and utilities for text-based analysis scenarios. - */ - -export type { - BuildSystemPromptOptions, - PromptMode, -} from "./system.js"; -export { buildSystemPrompt, sanitizeAiContent } from "./system.js"; diff --git a/services/discord-gateway/src/modules/ai-moderation/qdrantClient.ts b/services/discord-gateway/src/modules/ai-moderation/qdrantClient.ts index 02559e9..07ff582 100644 --- a/services/discord-gateway/src/modules/ai-moderation/qdrantClient.ts +++ b/services/discord-gateway/src/modules/ai-moderation/qdrantClient.ts @@ -22,6 +22,9 @@ export interface QdrantVerdictPayload { flags: string; // JSON string of the full moderation result analyzed_at: number; expires_at: number; + /** Bare content hash (16 hex chars) — enables content-based invalidation + * regardless of the (context-scoped) point id. */ + content_hash?: string; } function baseUrl(): string { @@ -215,6 +218,150 @@ export async function searchQdrant( } } +/** + * Batch search: one HTTP round-trip for N vectors (Qdrant + * `/points/search/batch`). Result is index-aligned with `vectors` — each + * entry is the top hits for that vector (or [] on per-vector failure). + * Used by the orchestrator to avoid N sequential embed→search round-trips. + */ +export async function searchQdrantBatch( + vectors: number[][], + limit: number, + scoreThreshold: number, +): Promise { + if (vectors.length === 0) return []; + try { + const json = (await request( + "POST", + `/collections/${collectionName()}/points/search/batch`, + { + searches: vectors.map((vector) => ({ + vector, + limit, + score_threshold: scoreThreshold, + with_payload: true, + filter: { + must: [ + { + key: "expires_at", + range: { gte: Date.now() }, + }, + ], + }, + })), + }, + )) as { + result?: Array<{ + result?: Array<{ + id?: number; + score?: number; + payload?: QdrantVerdictPayload; + }>; + }>; + }; + + return (json.result ?? []).map((entry) => + (entry.result ?? []) + .filter((hit) => hit.payload?.flags) + .map((hit) => ({ + cacheKey: `qdrant:${hit.id ?? "?"}`, + score: hit.score ?? 0, + payload: hit.payload as QdrantVerdictPayload, + })), + ); + } catch (error) { + log.warn( + { error: error instanceof Error ? error.message : String(error) }, + "Qdrant batch search failed — semantic cache skipped", + ); + return vectors.map(() => []); + } +} + +/** + * Delete expired verdict points from the collection. Best-effort: 404 + * (collection missing) and failures are swallowed — the periodic pruner + * just retries next sweep. + */ +export async function deleteExpiredQdrantPoints(): Promise { + try { + const json = (await request( + "POST", + `/collections/${collectionName()}/points/delete`, + { + filter: { + must: [ + { + key: "expires_at", + range: { lt: Date.now() }, + }, + ], + }, + }, + )) as { result?: { deleted?: number } | null }; + + return json.result?.deleted ?? 0; + } catch (error) { + if (error instanceof Error && error.message.includes("-> 404")) { + log.debug({}, "Qdrant collection absent — nothing to prune"); + } else { + log.warn( + { error: error instanceof Error ? error.message : String(error) }, + "Qdrant expired-point prune failed", + ); + } + return 0; + } +} + +/** + * Delete the verdict point for an exact cache key (used by cache + * invalidation when a moderator corrects a verdict). + */ +export async function deleteQdrantPoint(cacheKey: string): Promise { + try { + await request("POST", `/collections/${collectionName()}/points/delete`, { + points: [qdrantPointId(cacheKey)], + }); + return true; + } catch (error) { + log.warn( + { error: error instanceof Error ? error.message : String(error) }, + "Qdrant point delete failed", + ); + return false; + } +} + +/** + * Delete all verdict points whose payload carries a given bare content hash. + * Used by cache invalidation for corrected verdicts — matches context-scoped + * points that share the same content regardless of their point ids. + */ +export async function deleteQdrantPointsByContentHash( + bareHash: string, +): Promise { + try { + await request("POST", `/collections/${collectionName()}/points/delete`, { + filter: { + must: [ + { + key: "content_hash", + match: { value: bareHash }, + }, + ], + }, + }); + return true; + } catch (error) { + log.warn( + { error: error instanceof Error ? error.message : String(error) }, + "Qdrant content-hash point delete failed", + ); + return false; + } +} + /** True when Qdrant is configured (non-empty URL). */ export function isQdrantConfigured(): boolean { return Boolean(config.QDRANT_URL); diff --git a/services/discord-gateway/src/modules/ai-moderation/textBatchProcessor.ts b/services/discord-gateway/src/modules/ai-moderation/textBatchProcessor.ts index 45d227a..5e7adf2 100644 --- a/services/discord-gateway/src/modules/ai-moderation/textBatchProcessor.ts +++ b/services/discord-gateway/src/modules/ai-moderation/textBatchProcessor.ts @@ -12,13 +12,13 @@ import type { MessageRecord, } from "../message-capture/types.js"; import { getChannelCulture } from "./channelCultureStore.js"; +import type { ModerationPromptContent, RetryState } from "./llmCaller.js"; +import { callModerationLLM } from "./llmCaller.js"; import { buildReferenceXml, escapeXml, getAnalysisContent, } from "./moderationBuilders.js"; -import type { RetryState } from "./llmCaller.js"; -import { callModerationLLM } from "./llmCaller.js"; import { buildSystemPrompt as buildSystemPromptModular, sanitizeAiContent, @@ -74,7 +74,7 @@ export async function runTextOnlyBatch( if (!targets.length) return { results: [], raw: null }; const maxBatchSize = config.AI_LLM_TEXT_BATCH_SIZE ?? 20; - const timeoutMs = config.AI_LLM_MEDIA_ANALYSIS_TIMEOUT_MS ?? 60000; + const timeoutMs = config.AI_LLM_TEXT_ANALYSIS_TIMEOUT_MS ?? 30000; // Parallel: URL fetch + SearXNG const urlFetchPromise = (async () => { @@ -193,7 +193,9 @@ export async function runTextOnlyBatch( } } - const buildContent = async (state: RetryState): Promise => { + const buildContent = async ( + state: RetryState, + ): Promise => { const correction = state.lastParseError ? { error: state.lastParseError, @@ -241,7 +243,10 @@ export async function runTextOnlyBatch( ) .join("\n")}\n` : ""; - return `${systemText}${searxngBlock}\n\n\n${messagesBlock}\n`; + return { + system: systemText, + user: `${searxngBlock}\n\n\n${messagesBlock}\n`, + }; }; const abortController = new AbortController(); @@ -286,7 +291,15 @@ export async function runTextOnlyBatch( config.AI_LLM_MODEL, batchResult.results, 0, - undefined, + ( + batchResult.raw as { + usage?: { + prompt_tokens: number; + completion_tokens: number; + total_tokens: number; + }; + } | null + )?.usage ?? undefined, ); } diff --git a/services/discord-gateway/src/modules/ai-moderation/textCacheStore.ts b/services/discord-gateway/src/modules/ai-moderation/textCacheStore.ts index c580bae..6f015bd 100644 --- a/services/discord-gateway/src/modules/ai-moderation/textCacheStore.ts +++ b/services/discord-gateway/src/modules/ai-moderation/textCacheStore.ts @@ -3,7 +3,11 @@ import { createChildLogger } from "@/shared/logger/index"; import { executeAll, executeGet } from "../../shared/database/drizzle.js"; import { findBestEmbeddingMatch } from "./embeddingClient.js"; import { + deleteExpiredQdrantPoints, + deleteQdrantPoint, + deleteQdrantPointsByContentHash, isQdrantConfigured, + type QdrantVerdictPayload, searchQdrant, upsertQdrantPoint, } from "./qdrantClient.js"; @@ -19,84 +23,6 @@ export interface TextCacheEntry { hit_count: number; } -/** - * Lookup cached analysis result for a normalized text string. - * Returns null if not found or expired. - */ -export async function getCachedText( - text: string, -): Promise { - try { - const row = await executeGet( - `SELECT text, flags, source, analyzed_at, expires_at, hit_count - FROM text_analysis_cache - WHERE text = $1 AND expires_at > $2`, - [text, Date.now()], - ); - - if (!row) return null; - - return { - text: row.text, - flags: JSON.parse(row.flags), - source: row.source, - analyzed_at: row.analyzed_at, - expires_at: row.expires_at, - hit_count: row.hit_count, - }; - } catch (error) { - logger.error( - { error: error instanceof Error ? error.message : String(error) }, - "Failed to get cached text", - ); - return null; - } -} - -/** - * Insert or update a text analysis cache entry. - */ -export async function upsertCachedText( - text: string, - flags: string[], - source: "local" | "primary_ai" | "vision_llm", - expiresAt: number, -): Promise { - const now = Date.now(); - - try { - await executeAll( - `INSERT INTO text_analysis_cache (text, flags, source, analyzed_at, expires_at, hit_count) - VALUES ($1, $2, $3, $4, $5, 0) - ON CONFLICT (text) DO UPDATE SET - flags = EXCLUDED.flags, - source = EXCLUDED.source, - analyzed_at = EXCLUDED.analyzed_at, - expires_at = EXCLUDED.expires_at`, - [text, JSON.stringify(flags), source, now, expiresAt], - ); - } catch (error) { - logger.error( - { error: error instanceof Error ? error.message : String(error) }, - "Failed to upsert cached text", - ); - } -} - -/** - * Increment hit count for a cached text entry (called on cache hit). - */ -export async function incrementTextCacheHit(text: string): Promise { - try { - await executeAll( - `UPDATE text_analysis_cache SET hit_count = hit_count + 1 WHERE text = $1`, - [text], - ); - } catch (_error) { - // Silent fail — this is just a counter, not critical - } -} - /** * Delete expired cache entries. Run periodically to keep the table clean. */ @@ -116,47 +42,6 @@ export async function pruneExpiredTexts(): Promise { } } -/** - * Get cache statistics for observability. - */ -export async function getTextCacheStats(): Promise<{ - total: number; - expired: number; - bySource: Record; -}> { - try { - const now = Date.now(); - - const [totalRow, expiredRow, sourceRows] = await Promise.all([ - executeAll(`SELECT count(*) as cnt FROM text_analysis_cache`), - executeAll( - `SELECT count(*) as cnt FROM text_analysis_cache WHERE expires_at < $1`, - [now], - ), - executeAll( - `SELECT source, count(*) as cnt FROM text_analysis_cache GROUP BY source`, - ), - ]); - - const bySource: Record = {}; - for (const row of sourceRows) { - bySource[row.source] = row.cnt; - } - - return { - total: totalRow[0]?.cnt ?? 0, - expired: expiredRow[0]?.cnt ?? 0, - bySource, - }; - } catch (error) { - logger.error( - { error: error instanceof Error ? error.message : String(error) }, - "Failed to get text cache stats", - ); - return { total: 0, expired: 0, bySource: {} }; - } -} - // --------------------------------------------------------------------------- // Media / vision analysis cache helpers (reuses text_analysis_cache table) // --------------------------------------------------------------------------- @@ -295,15 +180,67 @@ export async function deleteCachedMediaAnalysis( // --------------------------------------------------------------------------- /** - * Generate a deterministic cache key for a per-user moderation result. + * Generate a deterministic cache key for a per-conversation moderation result. * - * Format: user_mod:: - * Two users sending the same text get separate cache entries so that - * per-user action history (e.g. repeated spam) can be tracked later. + * Format: text_mod:: + * + * `context` is "channel:thread" — the LLM verdict depends on conversation + * context, so the same text in a different channel/thread is analyzed + * separately instead of silently reusing a context-free verdict. + * (Previously the key was content-only, which made the "per-user" comment + * misleading — the user ID never participates in the key.) */ -export function makeTextModerationCacheKey(content: string): string { +export function makeTextModerationCacheKey( + content: string, + context?: string, +): string { const hash = createHash("sha256").update(content).digest("hex").slice(0, 16); - return `text_mod:${hash}`; + const ctx = context ? `${context}:` : ""; + return `text_mod:${ctx}${hash}`; +} + +/** + * Conversation context signature used in moderation cache keys. + * Thread-scoped messages use the thread id (discussions inside a thread + * share context), everything else uses the channel id. + */ +export function makeModerationContextKey(message: { + channel_id: string; + thread_id?: string | null; +}): string { + return message.thread_id ?? message.channel_id; +} + +/** + * Invalidate cached moderation verdicts for a piece of content: removes + * matching Postgres rows AND Qdrant points. Called when a moderator + * corrects a verdict so a stale/wrong cached decision cannot resurface. + * + * Handles both key formats: + * - legacy `text_mod:` (content-only, pre-context keys) + * - current `text_mod::` (channel/thread-scoped) + */ +export async function invalidateTextModerationCache( + content: string, +): Promise { + const bareHash = createHash("sha256") + .update(content) + .digest("hex") + .slice(0, 16); + const legacyKey = `text_mod:${bareHash}`; + + const queries: Promise[] = [ + executeAll(`DELETE FROM text_analysis_cache WHERE text LIKE $1`, [ + `text_mod:%${bareHash}`, + ]).catch(() => {}), + ]; + if (isQdrantConfigured()) { + queries.push( + deleteQdrantPoint(legacyKey).catch(() => {}), + deleteQdrantPointsByContentHash(bareHash).catch(() => {}), + ); + } + await Promise.all(queries).catch(() => {}); } /** @@ -360,6 +297,54 @@ export async function getCachedTextModeration(cacheKey: string): Promise<{ } } +/** + * Parse a Qdrant verdict payload into the result shape shared by the + * semantic cache lookups. Returns null on malformed payloads (callers then + * fall through to the LLM). + */ +export function parseQdrantVerdict( + payload: QdrantVerdictPayload, + similarity: number, +): { + text: string; + similarity: number; + status: "clean" | "warn" | "flagged"; + flags: string[]; + score: number; + analysis: string; + categories: string[]; + severity: string; + confidence: number; + recommendedAction: string; +} | null { + let parsed: Record; + try { + parsed = JSON.parse(payload.flags) as Record; + } catch { + return null; + } + if (!parsed || typeof parsed !== "object") return null; + + const storedStatus = (parsed.status as string) ?? "clean"; + const status: "clean" | "warn" | "flagged" = + storedStatus === "warn" || storedStatus === "flagged" + ? storedStatus + : "clean"; + + return { + text: payload.text, + similarity, + status, + flags: (parsed.flags as string[]) ?? [], + score: (parsed.score as number) ?? 0, + analysis: (parsed.analysis as string) ?? "", + categories: (parsed.categories as string[]) ?? [], + severity: (parsed.severity as string) ?? "none", + confidence: (parsed.confidence as number) ?? 0, + recommendedAction: (parsed.recommendedAction as string) ?? "none", + }; +} + /** * Semantic moderation cache lookup. * @@ -389,29 +374,7 @@ export async function findSimilarTextModeration( const hits = await searchQdrant(embedding, limit, minSimilarity); if (hits.length > 0) { const hit = hits[0]; - let parsed: Record; - try { - parsed = JSON.parse(hit.payload.flags) as Record; - } catch { - return null; - } - const storedStatus = (parsed.status as string) ?? "clean"; - const status: "clean" | "warn" | "flagged" = - storedStatus === "warn" || storedStatus === "flagged" - ? storedStatus - : "clean"; - return { - text: hit.payload.text, - similarity: hit.score, - status, - flags: (parsed.flags as string[]) ?? [], - score: (parsed.score as number) ?? 0, - analysis: (parsed.analysis as string) ?? "", - categories: (parsed.categories as string[]) ?? [], - severity: (parsed.severity as string) ?? "none", - confidence: (parsed.confidence as number) ?? 0, - recommendedAction: (parsed.recommendedAction as string) ?? "none", - }; + return parseQdrantVerdict(hit.payload, hit.score); } // No Qdrant hit — fall through to Postgres legacy rows. } @@ -521,6 +484,7 @@ export async function setCachedTextModeration( flags: JSON.stringify(result), analyzed_at: now, expires_at: now + USER_MOD_CACHE_TTL_MS, + content_hash: cacheKey.split(":").pop() ?? "", }); } @@ -666,6 +630,12 @@ export async function getRecentCorrectedModerations( /** * Store a corrected moderation entry for future few-shot injection. + * + * Also invalidates any cached verdicts for the corrected content (both + * Postgres rows and Qdrant points) so the corrected decision propagates + * immediately instead of being shadowed by a stale cache entry. Full + * content is looked up by message_id when available — more precise than + * the (possibly truncated) snippet. */ export async function insertCorrectedModeration(entry: { messageId: string; @@ -696,5 +666,20 @@ export async function insertCorrectedModeration(entry: { { error: error instanceof Error ? error.message : String(error) }, "Failed to insert corrected moderation", ); + return; + } + + // Best-effort invalidation: prefer full content from the messages table. + try { + const row = await executeGet( + `SELECT content, edited_content FROM messages WHERE id = $1`, + [entry.messageId], + ); + const fullContent = (row?.edited_content ?? row?.content ?? "").trim(); + await invalidateTextModerationCache( + fullContent || entry.contentSnippet, + ).catch(() => {}); + } catch { + await invalidateTextModerationCache(entry.contentSnippet).catch(() => {}); } } diff --git a/services/discord-gateway/src/modules/ai-moderation/userReputationStore.ts b/services/discord-gateway/src/modules/ai-moderation/userReputationStore.ts index 89263ae..c0b9667 100644 --- a/services/discord-gateway/src/modules/ai-moderation/userReputationStore.ts +++ b/services/discord-gateway/src/modules/ai-moderation/userReputationStore.ts @@ -1,5 +1,5 @@ -import { createChildLogger } from "@/shared/logger/index"; import { and, desc, eq } from "drizzle-orm"; +import { createChildLogger } from "@/shared/logger/index"; import { getDatabase } from "../../shared/database/drizzle.js"; import { messagesTable, @@ -9,6 +9,131 @@ import { const logger = createChildLogger("userReputationStore"); +// --------------------------------------------------------------------------- +// Trust model v2 — fair, recoverable, escalation-aware +// --------------------------------------------------------------------------- +// +// Problems with v1 that this fixes: +// 1. Trust practically could NOT rise: +2 per 100 clean messages meant a +// single -15 "high" penalty required 750 clean messages to repay. +// 2. Flat penalties regardless of history: first-timers and repeat +// offenders were punished identically. +// 3. Minor infractions could zero out a user (low=-2 at score 2 → 0), +// which is disproportionate. +// +// v2 model: +// - GAIN: +1 trust per 15 consecutive clean messages (cap 100). Recovery +// is real but earned — consistent good behavior rebuilds trust. +// - PENALTY: severity table low=3 / medium=6 / high=12 / critical=25. +// - FIRST OFFENSE: penalty halved (leniency for a single slip). +// - REPEAT OFFENDER: infraction within the last 7 days → ×1.5 (escalation). +// - FLOOR: low/medium infractions cannot push trust below 10/5 — minor +// offenses never permanently cripple a user; high/critical can still +// zero out (severe behavior has severe consequences). +// - Streak resets on infraction; time-based recovery still happens through +// the clean-message gain (no arbitrary idle-decay). +// --------------------------------------------------------------------------- + +export const TRUST_DEFAULTS = { + DEFAULT_TRUST: 50, + MAX_TRUST: 100, + MIN_TRUST: 0, + CLEAN_MESSAGES_PER_POINT: 15, + REPEAT_OFFENSE_WINDOW_MS: 7 * 24 * 60 * 60 * 1000, // 7 days + REPEAT_OFFENSE_MULTIPLIER: 1.5, +} as const; + +export const INFRACTION_PENALTIES: Record< + "low" | "medium" | "high" | "critical", + number +> = { + low: 3, + medium: 6, + high: 12, + critical: 25, +}; + +/** Trust floors per severity — minor offenses can't tank a user to zero. */ +export const INFRACTION_FLOORS: Record< + "low" | "medium" | "high" | "critical", + number +> = { + low: 10, + medium: 5, + high: 0, + critical: 0, +}; + +function clampTrust(score: number): number { + return Math.min( + TRUST_DEFAULTS.MAX_TRUST, + Math.max(TRUST_DEFAULTS.MIN_TRUST, Math.round(score)), + ); +} + +export interface InfractionContext { + totalInfractions: number; + lastInfractionAt: number | null; + severity: "low" | "medium" | "high" | "critical"; + now?: number; +} + +export interface InfractionOutcome { + penalty: number; + appliedRules: { + firstOffense: boolean; + repeatEscalation: boolean; + }; +} + +/** + * Pure penalty computation for the trust model (unit-testable, no DB). + * - First offense ever → halved (leniency for a single slip). + * - Repeat offense within the 7-day window → ×1.5 (escalation). + */ +export function computeInfractionPenalty( + ctx: InfractionContext, +): InfractionOutcome { + const basePenalty = INFRACTION_PENALTIES[ctx.severity]; + let penalty = basePenalty; + const isFirstOffense = ctx.totalInfractions === 0; + + if (isFirstOffense) { + penalty = Math.ceil(basePenalty / 2); + } else if ( + ctx.lastInfractionAt && + (ctx.now ?? Date.now()) - ctx.lastInfractionAt <= + TRUST_DEFAULTS.REPEAT_OFFENSE_WINDOW_MS + ) { + penalty = Math.ceil(basePenalty * TRUST_DEFAULTS.REPEAT_OFFENSE_MULTIPLIER); + } + + return { + penalty, + appliedRules: { + firstOffense: isFirstOffense, + repeatEscalation: !isFirstOffense && penalty > basePenalty, + }, + }; +} + +export interface CleanGainOutcome { + newStreak: number; + trustGain: number; +} + +/** + * Pure clean-message gain computation (unit-testable, no DB). + * +1 trust every CLEAN_MESSAGES_PER_POINT consecutive clean messages; + * the streak keeps counting past the threshold (gains compound). + */ +export function computeCleanTrustGain(currentStreak: number): CleanGainOutcome { + const newStreak = currentStreak + 1; + const trustGain = + newStreak % TRUST_DEFAULTS.CLEAN_MESSAGES_PER_POINT === 0 ? 1 : 0; + return { newStreak, trustGain }; +} + /** * Ensures a user reputation record exists. */ @@ -33,7 +158,7 @@ export async function initializeUserReputation( .values({ user_id: userId, guild_id: guildId, - trust_score: 50, + trust_score: TRUST_DEFAULTS.DEFAULT_TRUST, clean_message_streak: 0, total_infractions: 0, created_at: Date.now(), @@ -85,7 +210,11 @@ export async function getUserReputation( } /** - * Increment the clean message streak and update trust score if threshold is met. + * Increment the clean message streak and grow trust — +1 per + * CLEAN_MESSAGES_PER_POINT consecutive clean messages (cap 100). The streak + * keeps counting past the threshold so gains compound with continued good + * behavior (no more wasted progress at 100, and recovery is genuinely + * reachable after an infraction). */ export async function recordCleanMessage( userId: string, @@ -93,14 +222,11 @@ export async function recordCleanMessage( ): Promise { const rep = await initializeUserReputation(userId, guildId); const db = getDatabase(); - let newStreak = rep.clean_message_streak + 1; - let newScore = rep.trust_score; - - // Every 100 clean messages, give +2 trust score up to 100 - if (newStreak >= 100) { - newScore = Math.min(100, newScore + 2); - newStreak = 0; - } + const { newStreak, trustGain } = computeCleanTrustGain( + rep.clean_message_streak, + ); + const newScore = + trustGain > 0 ? clampTrust(rep.trust_score + trustGain) : rep.trust_score; await db .update(userReputationsTable) @@ -119,6 +245,12 @@ export async function recordCleanMessage( /** * Apply an infraction penalty to a user. + * + * Fairness rules: + * - First offense ever → penalty halved (leniency, rounded up). + * - Repeat offense within the 7-day window → ×1.5 (escalation). + * - Severity floor prevents minor infractions from zeroing a user. + * - Streak resets — trust must be re-earned through clean behavior. */ export async function recordInfraction( userId: string, @@ -127,23 +259,16 @@ export async function recordInfraction( ): Promise { const rep = await initializeUserReputation(userId, guildId); const db = getDatabase(); - let penalty = 0; - switch (severity) { - case "low": - penalty = 2; - break; - case "medium": - penalty = 5; - break; - case "high": - penalty = 15; - break; - case "critical": - penalty = 30; - break; - } - const newScore = Math.max(0, rep.trust_score - penalty); + const outcome = computeInfractionPenalty({ + totalInfractions: rep.total_infractions, + lastInfractionAt: rep.last_infraction_at, + severity, + }); + const { penalty } = outcome; + + const floor = INFRACTION_FLOORS[severity]; + const newScore = Math.max(floor, clampTrust(rep.trust_score - penalty)); await db .update(userReputationsTable) @@ -160,8 +285,12 @@ export async function recordInfraction( { userId, severity, + basePenalty: INFRACTION_PENALTIES[severity], penalty, + appliedRules: outcome.appliedRules, + previousScore: rep.trust_score, newScore, + floor, totalInfractions: rep.total_infractions + 1, }, "Infraction recorded", diff --git a/services/discord-gateway/src/shared/config/index.ts b/services/discord-gateway/src/shared/config/index.ts index 07ccf12..0d3196d 100644 --- a/services/discord-gateway/src/shared/config/index.ts +++ b/services/discord-gateway/src/shared/config/index.ts @@ -163,21 +163,14 @@ export const configSchema = z .int() .positive() .default(60000), - // ── AI Model (new unified keys) ─────────────────────────────────── - AI_MODEL_FAST_CLASSIFIER_ENABLED: z - .string() - .optional() - .transform((v) => v === "true") - .default(true) - .describe("Enable Layer 1 fast heuristic classifier"), - AI_MODEL_LLM_TIMEOUT_MS: z.coerce + // Text-only moderation batches are cheaper than media (no downloads / + // vision pre-pass), so they get their own (shorter) timeout instead of + // being tied to the media budget. + AI_LLM_TEXT_ANALYSIS_TIMEOUT_MS: z.coerce .number() .int() .positive() - .default(30000) - .describe("Timeout for individual LLM moderation calls"), - - + .default(30000), // ── AI Analysis Timing ────────────────────────────────────────────── AI_ANALYSIS_DEBOUNCE_MS: z.coerce.number().positive().default(500), @@ -210,7 +203,11 @@ export const configSchema = z .int() .positive() .default(50), - PISCINA_MAX_THREADS: z.coerce.number().int().positive().optional(), + // Worker pool size. Default 4 (not availableParallelism) because each + // Piscina thread owns its own pLimit(5) semaphore — on big VPSes + // availableParallelism × 5 concurrent LLM calls would overwhelm the + // router. Keep threads modest; concurrency is capped per-thread anyway. + PISCINA_MAX_THREADS: z.coerce.number().int().positive().default(4), // ── Voice Transcription ──────────────────────────────────────────────── AI_VOICE_TRANSCRIPTION_ENABLED: z @@ -219,14 +216,6 @@ export const configSchema = z .transform((v) => v === "true") .default(false), - // ── OpenAI Moderation ─────────────────────────────────────────────── - OPENAI_MODERATION_API_KEY: z.string().optional(), - OPENAI_MODERATION_BASE_URL: z - .string() - .url() - .default("https://api.openai.com/v1"), - OPENAI_MODERATION_MODEL: z.string().default("omni-moderation-latest"), - // ── Auto Delete ───────────────────────────────────────────────────── AUTO_DELETE_FLAGGED_ENABLED: z .string() diff --git a/services/discord-gateway/tests/trust-model.test.ts b/services/discord-gateway/tests/trust-model.test.ts new file mode 100644 index 0000000..2eb99a6 --- /dev/null +++ b/services/discord-gateway/tests/trust-model.test.ts @@ -0,0 +1,93 @@ +// ═══════════════════════════════════════════════════════════════════════════ +// Trust model v2 — pure math tests (no DB required) +// ═══════════════════════════════════════════════════════════════════════════ +import { describe, expect, it } from "vitest"; +import { + computeCleanTrustGain, + computeInfractionPenalty, + INFRACTION_FLOORS, + INFRACTION_PENALTIES, + TRUST_DEFAULTS, +} from "../src/modules/ai-moderation/userReputationStore.js"; + +describe("computeCleanTrustGain — trust CAN rise", () => { + it("grants +1 every CLEAN_MESSAGES_PER_POINT clean messages", () => { + const before = computeCleanTrustGain(14); + expect(before.newStreak).toBe(15); + expect(before.trustGain).toBe(1); + + const after = computeCleanTrustGain(15); + expect(after.newStreak).toBe(16); + expect(after.trustGain).toBe(0); + }); + + it("keeps compounding past the threshold (no wasted progress)", () => { + expect(computeCleanTrustGain(29).trustGain).toBe(1); + expect(computeCleanTrustGain(44).trustGain).toBe(1); + // 45 clean messages from a fresh start → 3 points of recovery + let gain = 0; + let streak = 0; + for (let i = 0; i < 45; i++) { + const r = computeCleanTrustGain(streak); + streak = r.newStreak; + gain += r.trustGain; + } + expect(gain).toBe(3); + }); +}); + +describe("computeInfractionPenalty — fair and escalating", () => { + const NOW = Date.now(); + + it("applies base penalty for a repeat offender outside the window", () => { + const r = computeInfractionPenalty({ + totalInfractions: 3, + lastInfractionAt: NOW - TRUST_DEFAULTS.REPEAT_OFFENSE_WINDOW_MS - 1000, + severity: "medium", + now: NOW, + }); + expect(r.penalty).toBe(INFRACTION_PENALTIES.medium); // 6 + expect(r.appliedRules.firstOffense).toBe(false); + expect(r.appliedRules.repeatEscalation).toBe(false); + }); + + it("halves the penalty for a first offense (leniency)", () => { + const r = computeInfractionPenalty({ + totalInfractions: 0, + lastInfractionAt: null, + severity: "high", + now: NOW, + }); + expect(r.penalty).toBe(Math.ceil(INFRACTION_PENALTIES.high / 2)); // 6 + expect(r.appliedRules.firstOffense).toBe(true); + }); + + it("escalates ×1.5 for a repeat offense within 7 days", () => { + const r = computeInfractionPenalty({ + totalInfractions: 2, + lastInfractionAt: NOW - 60 * 60 * 1000, // 1h ago + severity: "medium", + now: NOW, + }); + expect(r.penalty).toBe(Math.ceil(INFRACTION_PENALTIES.medium * 1.5)); // 9 + expect(r.appliedRules.repeatEscalation).toBe(true); + }); + + it("critical first offense still hurts but is halved", () => { + const r = computeInfractionPenalty({ + totalInfractions: 0, + lastInfractionAt: null, + severity: "critical", + now: NOW, + }); + expect(r.penalty).toBe(Math.ceil(INFRACTION_PENALTIES.critical / 2)); // 13 + }); + + it("severity floors prevent minor offenses from zeroing a user", () => { + expect(INFRACTION_FLOORS.low).toBeGreaterThan(0); + expect(INFRACTION_FLOORS.medium).toBeGreaterThan(0); + // high/critical can still reach zero — severe behavior has consequences + expect(INFRACTION_FLOORS.high).toBe(0); + expect(INFRACTION_FLOORS.critical).toBe(0); + }); +}); diff --git a/services/discord-gateway/vitest.config.ts b/services/discord-gateway/vitest.config.ts new file mode 100644 index 0000000..55fa072 --- /dev/null +++ b/services/discord-gateway/vitest.config.ts @@ -0,0 +1,23 @@ +import { fileURLToPath } from "node:url"; +import { defineConfig } from "vitest/config"; + +// Resolves the "@/*" tsconfig path alias so vitest can import src modules +// (the pre-existing test suite was broken without this). +export default defineConfig({ + resolve: { + alias: { + "@": fileURLToPath(new URL("./src", import.meta.url)), + }, + }, + test: { + include: ["tests/**/*.test.ts"], + // Loaded before module imports — satisfies the config singleton + // (DISCORD_TOKEN required) and DB-agnostic pure-function tests. + env: { + DISCORD_TOKEN: "test-discord-token", + DATABASE_URL: "postgres://localhost:5432/test", + AI_ANALYSIS_ENABLED: "true", + AI_LLM_API_KEY: "sk-test", + }, + }, +});