import { createChildLogger } from "@bete/shared/logger"; import { delay, retryWithBackoff } from "@bete/shared/utils"; import { LRUCache } from "lru-cache"; import type { ChatCompletion } from "openai/resources/chat/completions"; import { z } from "zod"; import { config } from "../../shared/config/config.js"; import { resizeImageForVision } from "../attachment-upload/imageResizer.js"; import { extractMessageMediaEvidence } from "../message-capture/messageMetadata.js"; import type { AnalysisResult, AttachmentRecord, MessageRecord, } from "../message-capture/types.js"; import { llmChat, llmVision } from "./llmClient.js"; import { buildSystemPrompt as buildSystemPromptModular } from "./moderationPrompt.js"; import { initializeUserReputation } from "./userReputationStore.js"; import { getChannelCulture } from "./channelCultureStore.js"; import { logModerationAnalysis, logModerationError } from "./responseLogger.js"; import { getStickerFromCache, initStickerCache, isStickerCacheReady, setStickerInCache, } from "./stickerCache.js"; import { buildCustomEmojiVisionPrompt, buildGeneralImageVisionPrompt, buildStickerTextOnlyWarning, buildStickerVisionPrompt, } from "./stickerPrompt.js"; import { acquireMediaAnalysisLock, computeImagePhash, deleteCachedMediaAnalysis, getCachedMediaAnalysis, getCachedMediaByPhash, getCachedTextModeration, getRecentCorrectedModerations, makeCustomEmojiCacheKey, makeImageCacheKey, makeStickerCacheKey, makeTextModerationCacheKey, setCachedTextModeration, upsertCachedMediaAnalysis, upsertCachedMediaByPhash, } from "./textCacheStore.js"; import { extractUrlsFromText, fetchUrlSafely } from "./urlFetcher.js"; const SeveritySchema = z.enum(["none", "low", "medium", "high", "critical"]); const RecommendedActionSchema = z.enum([ "none", "monitor", "warn", "review", "delete", "escalate", ]); const ResultItemSchema = z.object({ message_id: z.union([z.string(), z.number()]).transform(String), status: z.enum(["clean", "warn", "flagged"]), flags: z.array(z.string()).optional(), score: z.number(), analysis: z.string().nullable().optional(), categories: z.array(z.string()).optional(), severity: SeveritySchema.optional(), confidence: z.number().optional(), recommended_action: RecommendedActionSchema.optional(), policy_version: z.string().optional(), evidence: z.array(z.string()).optional(), }); const ModerationResponseSchema = z.object({ results: z.array(ResultItemSchema), }); const log = createChildLogger("llmModerationClient"); /** * Fetches recent corrected false positives from the DB and formats them * as additional few-shot examples for the moderation prompt. * * Returns an empty string if no corrections are available (so the prompt * builder simply skips the section). */ async function buildCorrectedFewShotExamples(): Promise { try { const corrections = await getRecentCorrectedModerations(5); if (corrections.length === 0) return ""; const lines = [ "## Contoh Koreksi False Positive (dari moderasi sebelumnya)", "Berikut adalah koreksi manual dari false positive yang pernah terjadi. Gunakan sebagai panduan tambahan:", ]; for (const c of corrections) { const origFlags = c.originalFlags.join(", ") || "(none)"; const corrFlags = c.correctedFlags.join(", ") || "(clean)"; const notes = c.correctionNotes ? ` — ${c.correctionNotes}` : ""; lines.push( `- Konten: "${c.contentSnippet.substring(0, 100)}" → sebelumnya di-flag sebagai [${origFlags}], dikoreksi menjadi [${corrFlags}]${notes}`, ); } lines.push( "JANGAN ulangi kesalahan yang sama. Jika konten serupa dengan contoh di atas, gunakan koreksi yang sudah ditentukan.", ); return lines.join("\n"); } catch { return ""; } } /** * Enhanced deferral detection pattern (R9). * * Only matches patterns where the model explicitly states it cannot make * a decision and needs human review. Removed overly broad patterns that * caused false positives: * - "admin (perlu|harus|sebaiknya)" → common in regular sentences * - "bisa (berpotensi|mengandung)" → decisive statements, not deferral * - "maaf|sorry" → opinions/apologies, not deferral * - "saya tidak yakin|tahu|paham" → expressing uncertainty, not deferral */ const DEFERRAL_ANALYSIS_PATTERN = /(?:kurang (?:konteks|bukti|informasi|data) (?:untuk (?:menilai|menentukan|memutuskan)|untuk moderasi)|perlu (?:dicek|diperiksa|ditinjau|dikaji|dievaluasi) (?:oleh )?(?:admin|moderator|manusia|human review)|tidak (?:bisa|dapat|mampu) (?:menentukan|menilai|memastikan|menyimpulkan|memberi keputusan|memoderasi).*(?:karena (?:konteks tidak jelas|informasi tidak cukup|bukti kurang|konteks kurang|tidak cukup konteks)|data tidak cukup|informasi tidak lengkap)|cannot determine|insufficient (?:context|evidence|information) (?:to |for )?(?:moderate|judge|evaluate|decide|classify)|(?:sepertinya|tampaknya) (?:perlu|harus) (?:ditinjau|diperiksa|dicek) (?:oleh )?(?:admin|moderator)|tidak cukup (?:bukti|informasi|konteks) (?:untuk (?:memberikan|membuat|menentukan)|memutuskan))/i; /** * Exceptions: patterns that look like deferral but are actually decisive. * Expanded to catch more variations where the model gives a clear verdict. */ const DEFERRAL_EXCEPTION_PATTERN = /tidak bisa menentukan.*(?:karena|sebab|dengan alasan|sebab tidak ada).*(?:clean|tidak (?:ada|terdapat|menunjukkan).*(?:pelanggaran|masalah|indikasi|konten)|aman|bersih|normal)/i; function hasDeferralAnalysis(analysis: string): boolean { if (DEFERRAL_EXCEPTION_PATTERN.test(analysis)) return false; return DEFERRAL_ANALYSIS_PATTERN.test(analysis); } function clampScore(value: number | undefined, fallback = 0): number { return Math.max( 0, Math.min(1, Number.isFinite(value) ? (value as number) : fallback), ); } function deriveSeverity( status: "clean" | "warn" | "flagged", score: number, ): z.infer { if (status === "clean") return "none"; if (status === "warn") return score >= 0.65 ? "medium" : "low"; if (score >= 0.9) return "critical"; return score >= 0.75 ? "high" : "medium"; } function deriveRecommendedAction( status: "clean" | "warn" | "flagged", severity: z.infer, ): z.infer { if (status === "clean") return "none"; if (status === "warn") return severity === "medium" ? "review" : "warn"; if (severity === "critical") return "escalate"; if (severity === "high") return "delete"; return "review"; } /** * Helper to extract JSON from a potentially conversational or markdown-wrapped string. */ export function extractJson(content: string): any { const codeBlockRegex = /```(?:json)?\s*([\s\S]*?)\s*```/g; const matches = content.matchAll(codeBlockRegex); for (const match of matches) { const codeContent = match[1].trim(); try { const parsed = JSON.parse(codeContent); if (parsed && typeof parsed === "object") { return parsed; } } catch (_) {} } for (let start = 0; start < content.length; start++) { const firstChar = content[start]; if (firstChar !== "{" && firstChar !== "[") continue; const stack = [firstChar]; let inString = false; let escaped = false; for (let i = start + 1; i < content.length; i++) { const char = content[i]; if (inString) { if (escaped) { escaped = false; } else if (char === "\\") { escaped = true; } else if (char === '"') { inString = false; } continue; } if (char === '"') { inString = true; continue; } if (char === "{" || char === "[") { stack.push(char); continue; } const last = stack[stack.length - 1]; if ((char === "}" && last === "{") || (char === "]" && last === "[")) { stack.pop(); if (stack.length === 0) { const candidate = content.slice(start, i + 1); try { const parsed = JSON.parse(candidate); if (parsed && typeof parsed === "object") { return parsed; } } catch (_) {} break; } } } } throw new Error("No JSON object found in response"); } /** * Sanitize error messages for client-facing output (R10). * Internal details are logged but the caller gets a generic message. */ function sanitizeErrorMessage(internalMsg: string, messageId: string): string { // Log the full error for debugging log.warn( { messageId, internalError: internalMsg }, "Internal moderation error (sanitized for client)", ); // Return generic message without internal details return `Analisis gagal dan memerlukan pemeriksaan manual. Error code: MOD_${Date.now().toString(36).slice(0, 6)}`; } export function parseModerationResponse( content: string, targetIds: string[], ): AnalysisResult[] { let parsed: any; try { parsed = JSON.parse(content); } catch (e) { parsed = extractJson(content); } if (Array.isArray(parsed)) { parsed = { results: parsed }; } else if (parsed && typeof parsed === "object" && !("results" in parsed)) { if ("message_id" in parsed) { parsed = { results: [parsed] }; } else { const arrayKey = Object.keys(parsed).find((key) => { const val = (parsed as any)[key]; return ( Array.isArray(val) && val.length > 0 && val.every( (item: unknown) => typeof item === "object" && item !== null && "message_id" in (item as any), ) ); }); if (arrayKey) { parsed.results = (parsed as any)[arrayKey]; } else { parsed = { results: [parsed] }; } } } const parseResult = ModerationResponseSchema.safeParse(parsed); if (!parseResult.success) { throw new Error(`Zod validation failed: ${parseResult.error.message}`); } const response = parseResult.data; const foundIds = new Set(); const targetIdSet = new Set(targetIds); const results: (AnalysisResult | null)[] = response.results.map((result) => { const { message_id, status, flags, score, analysis, categories, severity, confidence, recommended_action, policy_version, evidence, } = result; const finalId = message_id.trim(); if (!targetIdSet.has(finalId)) { return null; } if (foundIds.has(finalId)) { throw new Error( `Duplicate message_id in moderation response: ${finalId}`, ); } foundIds.add(finalId); const coalescedAnalysis = analysis ?? ""; if (hasDeferralAnalysis(coalescedAnalysis)) { throw new Error( `Deferral analysis is not allowed for message ${finalId}; return a direct moderation decision`, ); } const normalizedScore = clampScore(score); const normalizedConfidence = clampScore(confidence, normalizedScore); const normalizedSeverity = severity ?? deriveSeverity(status, normalizedScore); return { messageId: finalId, status: status as "clean" | "warn" | "flagged", flags: flags ?? [], score: normalizedScore, analysis: coalescedAnalysis, categories: categories ?? flags ?? [], severity: normalizedSeverity, confidence: normalizedConfidence, recommendedAction: recommended_action ?? deriveRecommendedAction(status, normalizedSeverity), policyVersion: policy_version ?? "default-2026-05-30", evidence: evidence ?? [], }; }); const filteredResults = results.filter( (r): r is AnalysisResult => r !== null, ); const missingIds = targetIds.filter((id) => !foundIds.has(id)); if (missingIds.length > 0) { log.warn( { missingIds, foundCount: foundIds.size, totalCount: targetIds.length }, "Some target IDs missing in response - marking as incomplete", ); for (const missingId of missingIds) { filteredResults.push({ messageId: missingId, status: "error", flags: ["analysis_incomplete"], score: 0, analysis: sanitizeErrorMessage( "Analysis incomplete - LLM did not process this message", missingId, ), categories: ["analysis_incomplete"], severity: "none", confidence: 0, recommendedAction: "review", policyVersion: "default-2026-05-30", evidence: [], }); } } return filteredResults; } interface ModerationInput { targets: MessageRecord[]; contextText: string; attachments?: AttachmentRecord[]; } interface ModerationOutput { results: AnalysisResult[]; raw: unknown; } /** * Sniff the first bytes of a buffer to determine if it is a supported image * format. Returns the canonical MIME type string on success, or null if the * bytes are not a recognizable image. */ function sniffImageMimeType(buf: Buffer): string | null { if (buf.length < 12) return null; if (buf[0] === 0xff && buf[1] === 0xd8 && buf[2] === 0xff) { return "image/jpeg"; } if ( buf[0] === 0x89 && buf[1] === 0x50 && buf[2] === 0x4e && buf[3] === 0x47 && buf[4] === 0x0d && buf[5] === 0x0a && buf[6] === 0x1a && buf[7] === 0x0a ) { return "image/png"; } if ( buf[0] === 0x47 && buf[1] === 0x49 && buf[2] === 0x46 && buf[3] === 0x38 ) { return "image/gif"; } if ( buf[0] === 0x52 && buf[1] === 0x49 && buf[2] === 0x46 && buf[3] === 0x46 && buf[8] === 0x57 && buf[9] === 0x45 && buf[10] === 0x42 && buf[11] === 0x50 ) { return "image/webp"; } if ( buf.length >= 12 && buf[4] === 0x66 && buf[5] === 0x74 && buf[6] === 0x79 && buf[7] === 0x70 ) { const brand = buf.subarray(8, 12).toString("ascii"); if (brand.startsWith("avif") || brand.startsWith("avis")) { return "image/avif"; } if ( brand.startsWith("mif1") || brand.startsWith("heic") || brand.startsWith("heis") ) { return "image/heic"; } } return null; } // --------------------------------------------------------------------------- // Shared types for image resolution // --------------------------------------------------------------------------- type MessageImagePart = { type: "image_url"; image_url: { url: string }; sourceLabel: string; stickerName?: string; customEmojiId?: string; customEmojiName?: string; }; // --------------------------------------------------------------------------- // Content helpers // --------------------------------------------------------------------------- /** * Returns the real text content for AI analysis, stripping fallback text * that getDisplayContent() synthesized ("[Attachment: ...]", "[Sticker: ...]", * "[Embed]"). These filenames alone are meaningless to the LLM and can * falsely inflate a "clean" verdict when the actual image failed to download. */ function getAnalysisContent(message: MessageRecord): string { const raw = message.edited_content ?? message.content; const stripped = raw.replace( /\[(?:Attachment|Sticker):[^\]]*\]|\[Embed\]/g, "", ); return stripped.trim(); } // --------------------------------------------------------------------------- // Media detection helper // --------------------------------------------------------------------------- function hasMediaContent( target: MessageRecord, attachments?: AttachmentRecord[], ): boolean { if (target.metadata) { const evidence = extractMessageMediaEvidence(target.metadata); if (evidence.stickers.length > 0 || evidence.embeds.length > 0) return true; } if (attachments?.some((a) => a.message_id === target.id)) return true; return false; } // --------------------------------------------------------------------------- // Single-image vision analysis (reused by both text-only and media paths) // --------------------------------------------------------------------------- /** * In-memory LRU cache for vision analysis results. * Fastest path — avoids DB round-trip for frequently seen images. * Max 500 entries, 24-hour TTL. */ const visionLruCache = new LRUCache({ max: 500, ttl: 24 * 60 * 60 * 1000, }); /** * In-flight deduplication map — prevents concurrent vision API calls for * the same cache key. Multiple concurrent requests for an identical image * share the same promise, eliminating the race condition between cache * check and cache write. */ const inFlightVisionCalls = new Map>(); const FAILED_ANALYSIS_PREFIX = "GAGAL DIANALISIS — gambar tidak dapat diunduh atau vision API gagal setelah 3x percobaan. JANGAN mengasumsikan gambar aman hanya karena gagal dianalisis. Gunakan metadata URL/nama file saja sebagai petunjuk."; const analyzeSingleMediaImage = async ( messageId: string, image: MessageImagePart, ): Promise => { const cacheKey = image.customEmojiId ? makeCustomEmojiCacheKey(image.customEmojiId) : image.stickerName ? makeStickerCacheKey(image.stickerName) : makeImageCacheKey(image.image_url.url); // Layer 0: In-memory LRU cache (fastest — no DB or network I/O) const lruCached = visionLruCache.get(cacheKey); if (lruCached) { log.debug({ cacheKey }, "Vision LRU cache HIT (in-memory)"); return `[Media analysis for message ${messageId}] ${image.sourceLabel}: ${lruCached}`; } // Layer 1: DB cache const cached = await getCachedMediaAnalysis(cacheKey); if (cached) { visionLruCache.set(cacheKey, cached); log.debug({ cacheKey }, "Media analysis cache HIT (DB → LRU)"); return `[Media analysis for message ${messageId}] ${image.sourceLabel}: ${cached}`; } // Deduplicate in-flight vision calls: if another caller is already // processing this exact image, wait for it instead of starting a duplicate. const existing = inFlightVisionCalls.get(cacheKey); if (existing) { log.debug( { cacheKey }, "Media analysis in-flight dedupe — waiting for existing call", ); const result = await existing; // result is never null — the promise always returns a descriptive string return `[Media analysis for message ${messageId}] ${image.sourceLabel}: ${result}`; } const promptText = image.stickerName ? buildStickerVisionPrompt(image.stickerName, messageId) : image.customEmojiName ? buildCustomEmojiVisionPrompt(image.customEmojiName, messageId) : buildGeneralImageVisionPrompt(image.sourceLabel, messageId); const visionPromise = (async (): Promise => { // Attempt to acquire DISTRIBUTED lock // Lock expires in 60 seconds (generous timeout for LLM) const locked = await acquireMediaAnalysisLock(cacheKey, Date.now() + 60000); if (!locked) { log.debug({ cacheKey }, "Media analysis distributed lock acquired by another pod. Polling..."); // Poll DB for up to 30 seconds for (let i = 0; i < 15; i++) { await new Promise((resolve) => setTimeout(resolve, 2000)); const pollCached = await getCachedMediaAnalysis(cacheKey); if (pollCached) { visionLruCache.set(cacheKey, pollCached); return pollCached; } } log.warn({ cacheKey }, "Polling for distributed media analysis timed out. Falling back."); return FAILED_ANALYSIS_PREFIX; } // Layer 2: Perceptual hash pre-check (before expensive vision API call) let phash: string | null = null; if (image.image_url.url.startsWith("data:")) { try { const base64Data = image.image_url.url.split(",")[1]; if (base64Data) { const imgBuffer = Buffer.from(base64Data, "base64"); phash = await computeImagePhash(imgBuffer); if (phash) { const phashCached = await getCachedMediaByPhash(phash); if (phashCached) { log.debug( { cacheKey, phash: phash.slice(0, 16) }, "Vision phash HIT — reusing analysis", ); visionLruCache.set(cacheKey, phashCached); await upsertCachedMediaAnalysis( cacheKey, phashCached, "vision_llm", Date.now() + 24 * 60 * 60 * 1000, ).catch(() => {}); return phashCached; } } } } catch { // phash failed — continue with normal vision API flow phash = null; } } // ── Vision API call with EXTERNAL exponential backoff ── // Uses retryWithBackoff directly so each retry has proper backoff delay. // llmVision calls llmChat which also has retryWithBackoff, but its // inner backoff has minTimeout=0 (instant). Our outer backoff ensures // meaningful delay between full attempts. let lastError: Error | null = null; for (let attempt = 0; attempt < 3; attempt++) { try { const content = await llmVision(promptText, image.image_url); if (content) { // Success — persist to all cache layers await upsertCachedMediaAnalysis( cacheKey, content, "vision_llm", Date.now() + 24 * 60 * 60 * 1000, ); visionLruCache.set(cacheKey, content); if (phash) { upsertCachedMediaByPhash( phash, content, "vision_llm", Date.now() + 7 * 24 * 60 * 60 * 1000, ).catch(() => {}); } return content; } // llmVision returned null (no API key / client unavailable) — no point retrying log.warn( { messageId }, "Vision API client unavailable (null response) — skipping retry", ); break; } catch (err) { lastError = err instanceof Error ? err : new Error(String(err)); if (attempt < 2) { const backoffMs = Math.min( 2_000 * 3 ** attempt + Math.random() * 500, 30_000, ); log.warn( { messageId, attempt: attempt + 1, backoffMs, error: lastError.message, }, "Vision API attempt failed — backing off before retry", ); await delay(backoffMs); } } } // All attempts exhausted — return descriptive failure text. // NOT null: the caller must always have a descriptive string to // inject into the prompt so the LLM knows the image was skipped. log.warn( { messageId, lastError: lastError?.message ?? "null response", }, "Vision analysis failed after all retry attempts", ); await deleteCachedMediaAnalysis(cacheKey).catch(() => {}); return FAILED_ANALYSIS_PREFIX; })(); inFlightVisionCalls.set(cacheKey, visionPromise); try { const content = await visionPromise; return `[Media analysis for message ${messageId}] ${image.sourceLabel}: ${content}`; } catch (outerErr) { // Should never reach here — visionPromise has its own catch that // returns FAILED_ANALYSIS_PREFIX. Log anyway for debugging. log.error( { messageId, cacheKey, error: outerErr instanceof Error ? outerErr.message : String(outerErr), }, "Unexpected rejection in analyzeSingleMediaImage (visionPromise threw)", ); return `[Media analysis for message ${messageId}] ${image.sourceLabel}: ${FAILED_ANALYSIS_PREFIX}`; } finally { inFlightVisionCalls.delete(cacheKey); } }; // --------------------------------------------------------------------------- // Shared LLM call + parse + fallback helper // --------------------------------------------------------------------------- /** * State object shared between the caller and callModerationLLM so that * parse-error feedback can be injected into subsequent retry attempts. * * The caller creates this object, passes it to callModerationLLM, and * the internal retry loop mutates it before re-invoking buildContent(). */ interface RetryState { lastParseError: string | null; lastInvalidContent: string | null; } /** * Execute a single LLM moderation call (batch or single-message) with retry * logic, JSON parse, and fallback error markers on failure. * * Uses JSON Schema response format (R2) and concurrency limiter (R3). */ async function callModerationLLM( buildContent: (state: RetryState) => Promise, targetIds: string[], label: string, signal?: AbortSignal, ): Promise<{ results: AnalysisResult[]; raw: ChatCompletion | null; }> { const state: RetryState = { lastParseError: null, lastInvalidContent: null, }; let parsed: AnalysisResult[]; let result: ChatCompletion | null = null; try { const analysis = await retryWithBackoff( async () => { try { const content = await buildContent(state); const completion = await llmChat({ messages: [{ role: "user", content }], max_tokens: 16384, jsonResponse: { type: "json_object" }, retries: 0, signal, }); if (!completion) { throw new Error("LLM client unavailable (no API key)"); } if ( !completion.choices || !Array.isArray(completion.choices) || !completion.choices[0] ) { throw new Error("Invalid LLM response structure"); } const rawContent = completion.choices[0].message?.content; if (!rawContent) { throw new Error("No content in LLM response"); } try { return { parsed: parseModerationResponse(rawContent, targetIds), result: completion, }; } catch (parseError) { state.lastParseError = parseError instanceof Error ? parseError.message : String(parseError); state.lastInvalidContent = rawContent; log.warn( { error: state.lastParseError, contentLength: rawContent.length, contentPreview: rawContent.substring(0, 1000), targetIds, model: config.AI_LLM_MODEL, }, `Failed to parse moderation response from LLM (${label})`, ); throw parseError; } } catch (apiError: any) { // 429 → retryable with backoff (rate limited — the provider may recover) // GitHub Issue #429-cascade: 429 was previously an AbortError which skipped // retries and caused immediate re-queues, creating a tight rate-limit cascade. // Now treated as retryable with generous backoff so the provider can recover. if (apiError?.status === 429) { log.warn( { status: 429, targetIds, model: config.AI_LLM_MODEL, label, }, "LLM API 429 rate limited — will retry with backoff", ); // Add a small jitter to prevent thundering herd on retry const jitterMs = Math.floor(Math.random() * 1000) + 500; await delay(jitterMs); throw apiError; } // 401/403 → abort immediately, never retry if ( apiError?.status === 401 || apiError?.status === 403 ) { const abortErr = new Error(String(apiError)); abortErr.name = "AbortError"; throw abortErr; } // 5xx server errors → retryable transient errors // p-retry will retry these; on final exhaustion the outer catch // will produce synthetic error results for all targets if ( apiError?.status >= 500 || apiError?.code === "ECONNRESET" || apiError?.code === "ETIMEDOUT" || apiError?.name === "APIError" ) { // re-throw as-is so p-retry can retry throw apiError; } throw apiError; } }, { retries: 3, minTimeout: 5_000, maxTimeout: 60_000, factor: 3, signal, }, ); parsed = analysis.parsed; result = analysis.result; } catch (err) { if (err instanceof Error && err.name === "AbortError") { throw err; } const errorMsg = err instanceof Error ? err.message : String(err); const isApiError = !state.lastInvalidContent; // For API errors (502, timeout, etc.) where retries exhausted, produce // synthetic error results so the batch doesn't crash entirely. // For parse errors, we already have lastInvalidContent and the existing // fallback path below handles it. const apiErrorCode = isApiError ? `MOD_${Date.now().toString(36).slice(0, 6)}` : null; if (isApiError) { log.warn( { error: errorMsg, targetIds, model: config.AI_LLM_MODEL, label, }, `LLM API error after retries exhausted (${label}) — marking all targets as analysis errors`, ); logModerationError( targetIds, config.AI_LLM_MODEL, err instanceof Error ? err : new Error(String(err)), { phase: "api_call", label, }, ); parsed = targetIds.map((id) => ({ messageId: id, status: "error", flags: ["analysis_api_failed"], score: 0, analysis: `Analisis gagal karena error pada server AI dan memerlukan pemeriksaan manual. Error code: ${apiErrorCode}`, categories: ["analysis_api_failed"], severity: "none", confidence: 0, recommendedAction: "review", policyVersion: "default-2026-05-30", evidence: [], })); } else { // Parse error fallback — existing path const parseMsg = err instanceof Error ? err.message : String(err); const contentPreview = state.lastInvalidContent?.substring(0, 500) ?? ""; const contentLen = state.lastInvalidContent?.length ?? 0; log.error( { error: parseMsg, contentLength: contentLen, contentPreview, targetIds, model: config.AI_LLM_MODEL, timestamp: new Date().toISOString(), }, `Robust Fallback (${label}): Failed to parse moderation response. Marking all targets as analysis errors.`, ); // Log error with responseLogger logModerationError( targetIds, config.AI_LLM_MODEL, err instanceof Error ? err : new Error(String(err)), { phase: "parse_response", label, contentLength: contentLen, }, ); // Sanitized error messages — no internal details exposed (R10) const errorCode = `MOD_${Date.now().toString(36).slice(0, 6)}`; parsed = targetIds.map((id) => ({ messageId: id, status: "error", flags: ["analysis_parse_failed"], score: 0, analysis: `Analisis gagal dan memerlukan pemeriksaan manual. Error code: ${errorCode}`, categories: ["analysis_parse_failed"], severity: "none", confidence: 0, recommendedAction: "review", policyVersion: "default-2026-05-30", evidence: [], })); } } return { results: parsed, raw: result }; } // --------------------------------------------------------------------------- // Text-only fast path — with batch size splitting (R6) // --------------------------------------------------------------------------- /** * Run a lightweight batch analysis on text-only messages. * * If targets exceed AI_LLM_TEXT_BATCH_SIZE, split into sub-batches * and run sequentially to avoid overwhelming the LLM (R6). */ async function runTextOnlyBatch( targets: MessageRecord[], contextText: string, ): Promise<{ results: AnalysisResult[]; raw: unknown }> { 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; // ── Fetch web content from URLs in text-only messages ── // Prevents LLM from guessing based on domain name alone (e.g., false "scam" flags). // The fetched text content is injected into the message XML so the LLM can // analyze the actual page rather than pattern-match the URL string. const urlFetchMap = new Map(); // url → fetched text content { const allUrls = new Set(); for (const msg of targets) { const content = msg.edited_content ?? msg.content; for (const url of extractUrlsFromText(content)) { allUrls.add(url); } } const urlArr = Array.from(allUrls).slice(0, 10); // cap to 10 fetches per batch if (urlArr.length > 0) { log.debug( { urlCount: urlArr.length }, "Fetching web content for text-only batch URLs", ); const results = await Promise.allSettled( urlArr.map((url) => fetchUrlSafely(url)), ); for (let i = 0; i < urlArr.length; i++) { const r = results[i]; if ( r.status === "fulfilled" && r.value.type === "text" && r.value.textContent ) { urlFetchMap.set(urlArr[i], r.value.textContent); } } } } // ── Group identical short messages (< 20 chars) to reduce redundant analysis ── // Messages with identical normalized content share a single representative. // Results are fanned out to all group members after the LLM call. const shortContentGroups = new Map(); const deduplicatedTargets: MessageRecord[] = []; const groupMapping = new Map(); // representativeId → [all memberIds] for (const msg of targets) { const rawContent = (msg.edited_content ?? msg.content).trim(); if (rawContent.length > 0 && rawContent.length < 20) { const groupKey = rawContent.toLowerCase(); if (shortContentGroups.has(groupKey)) { shortContentGroups.get(groupKey)!.push(msg); } else { shortContentGroups.set(groupKey, [msg]); deduplicatedTargets.push(msg); // first occurrence = representative } } else { deduplicatedTargets.push(msg); } } // Build group mapping for results fan-out for (const [, members] of shortContentGroups) { if (members.length > 1) { const rep = members[0]; groupMapping.set( rep.id, members.map((m) => m.id), ); } } if (groupMapping.size > 0) { log.debug( { originalCount: targets.length, deduplicatedCount: deduplicatedTargets.length, groupsFormed: groupMapping.size, }, "Grouped identical short messages for text batch", ); } // Split into sub-batches if needed (R6) — using deduplicated targets const subBatches: MessageRecord[][] = []; for (let i = 0; i < deduplicatedTargets.length; i += maxBatchSize) { subBatches.push(deduplicatedTargets.slice(i, i + maxBatchSize)); } if (subBatches.length > 1) { log.debug( { totalTargets: targets.length, subBatchCount: subBatches.length, maxBatchSize, }, "Text targets exceed batch size limit — splitting into sub-batches", ); } const allResults: AnalysisResult[] = []; let lastRaw: unknown = null; const channelId = targets.length > 0 ? targets[0].channel_id : ""; const guildId = targets.length > 0 ? targets[0].guild_id : ""; const channelCultureObj = channelId ? await getChannelCulture(channelId) : null; const channelCulture = channelCultureObj ? channelCultureObj.culture_summary : undefined; // Run sub-batches sequentially to avoid rate limits for (let i = 0; i < subBatches.length; i++) { const batch = subBatches[i]; const targetIds = batch.map((t) => t.id); // Abstract user reputation (no history — prevents confirmation bias) const userContexts = new Map(); for (const msg of batch) { if (!userContexts.has(msg.user_id)) { const rep = await initializeUserReputation(msg.user_id, msg.guild_id); const contextStr = ``; userContexts.set(msg.user_id, contextStr); } } const buildContent = async (state: RetryState): Promise => { const correction = state.lastParseError ? { error: state.lastParseError, preview: state.lastInvalidContent?.slice(0, 800) ?? "", } : undefined; // Use modular system prompt with XML delimiters (R1, R7, R8) const correctedExamples = await buildCorrectedFewShotExamples(); const systemText = buildSystemPromptModular({ contextText, mode: "text", correction, correctedExamples, channelCulture, }); const messagesBlock = batch .map((msg) => { const content = getAnalysisContent(msg); // Inject fetched web content for URLs found in this message const msgUrls = extractUrlsFromText(content); const urlContexts = msgUrls .map((url) => { const fetchedText = urlFetchMap.get(url); if (!fetchedText) return null; return `${fetchedText}`; }) .filter(Boolean) .join("\n"); const webContext = urlContexts ? `\n${urlContexts}` : ""; const userCtx = userContexts.get(msg.user_id) ?? ""; // XML delimiters wrap each message for prompt safety (R1) return `\n ${userCtx}\n ${content}${webContext}\n`; }) .join("\n"); // XML delimiter wraps the entire messages block (R1) return `${systemText}\n\n\n${messagesBlock}\n`; }; const abortController = new AbortController(); const timeoutId = setTimeout(() => { abortController.abort(); }, timeoutMs); timeoutId.unref(); let batchResult: { results: AnalysisResult[]; raw: unknown }; try { batchResult = await callModerationLLM(buildContent, targetIds, `text-batch-${i + 1}`, abortController.signal); } catch (err: any) { if (err.name === "AbortError" || abortController.signal.aborted) { throw new Error( `Text-only batch sub-batch ${i + 1} timed out for messages ${targetIds.join(", ")}`, ); } throw err; } finally { clearTimeout(timeoutId); } const rawUsage = ( batchResult.raw as { usage?: { prompt_tokens: number; completion_tokens: number; total_tokens: number; }; } )?.usage; // Fan-out results from representative messages to all group members const fannedOutResults = groupMapping.size > 0 ? batchResult.results.flatMap((result) => { const members = groupMapping.get(result.messageId); if (members) { return members.map((memberId) => ({ ...result, messageId: memberId, })); } return [result]; }) : batchResult.results; allResults.push(...fannedOutResults); if (batchResult.raw) lastRaw = batchResult.raw; // Log batch results with comprehensive details logModerationAnalysis( targetIds, config.AI_LLM_MODEL, batchResult.results, 0, // Duration will be tracked at higher level rawUsage ? { prompt_tokens: rawUsage.prompt_tokens, completion_tokens: rawUsage.completion_tokens, total_tokens: rawUsage.total_tokens, } : undefined, ); } log.debug( { targetCount: targets.length, resultCount: allResults.length, subBatchCount: subBatches.length, }, "Text-only batch analysis complete", ); return { results: allResults, raw: lastRaw }; } // --------------------------------------------------------------------------- // Single media message analysis — one LLM call per message with vision + timeout (R4, R5) // --------------------------------------------------------------------------- /** * Process a single media-bearing message: * 1. Download attachment images (resized via sharp — R5) * 2. Fetch URLs found in the message body * 3. Download sticker/embed images (resized via sharp — R5) * 4. Run vision analysis on every image (with DB + sticker cache) * 5. Build a single-message prompt with XML delimiters (R1) * 6. One LLM call → single AnalysisResult * * Wrapped with overall timeout (R4). */ async function runSingleMediaAnalysis( target: MessageRecord, contextText: string, allAttachments: AttachmentRecord[] | undefined, ): Promise<{ results: AnalysisResult[]; raw: unknown }> { const targetId = target.id; const targetIds = [targetId]; // Timeout wrapper (R4) const timeoutMs = config.AI_LLM_MEDIA_ANALYSIS_TIMEOUT_MS ?? 60000; const abortController = new AbortController(); const timeoutId = setTimeout(() => { abortController.abort(); }, timeoutMs); timeoutId.unref(); try { return await _runSingleMediaAnalysis( target, contextText, allAttachments, targetId, targetIds, abortController.signal, ); } catch (err: any) { if (err.name === "AbortError" || abortController.signal.aborted) { throw new Error( `Media analysis timed out after ${timeoutMs}ms for message ${targetId}`, ); } throw err; } finally { clearTimeout(timeoutId); } } async function _runSingleMediaAnalysis( target: MessageRecord, contextText: string, allAttachments: AttachmentRecord[] | undefined, targetId: string, targetIds: string[], signal?: AbortSignal, ): Promise<{ results: AnalysisResult[]; raw: unknown }> { // Lazy init sticker cache if (!isStickerCacheReady()) { await initStickerCache().catch((err: unknown) => { log.warn( { error: err instanceof Error ? err.message : String(err) }, "Sticker cache init failed — continuing without cache", ); }); } // ── State maps for this single message ── const imageMap = new Map(); const webTextMap = new Map(); const mediaAnalysisMap = new Map(); const getAttachmentImageUrl = (att: AttachmentRecord): string | null => att.uploaded_url ?? att.discord_url ?? null; const maxDimension = config.AI_LLM_IMAGE_MAX_DIMENSION ?? 1024; const content = getAnalysisContent(target); // ── 1-3. Parallel download of ALL media sources ── // Build all download promises upfront and execute them in one Promise.all. // Attachment, URL, sticker/emoji downloads are fully independent of each other. // An 8-image cap is enforced across all sources combined. const downloadPromises: Array> = []; // ── Attachment downloads ── const msgAttachments = (allAttachments ?? []) .filter( (att) => att.message_id === targetId && getAttachmentImageUrl(att) && att.type.startsWith("image/"), ) .slice(0, 8); for (const att of msgAttachments) { downloadPromises.push( (async () => { const urlToUse = getAttachmentImageUrl(att); if (!urlToUse) { log.warn( { attachmentId: att.id, messageId: att.message_id }, "Skipping attachment: no uploaded URL available", ); return; } const controller = new AbortController(); const timeoutId = setTimeout(() => controller.abort(), 15000); try { const res = await fetch(urlToUse, { signal: controller.signal }); if (!res.ok || !res.body) { log.warn( { attachmentId: att.id, url: urlToUse, status: res.status }, "Failed to download attachment: HTTP error or no body", ); return; } let totalBytes = 0; const chunks: Uint8Array[] = []; const reader = res.body.getReader(); while (true) { const { done, value } = await reader.read(); if (done) break; if (value) { totalBytes += value.length; if (totalBytes > 10 * 1024 * 1024) { log.warn( { attachmentId: att.id, totalBytes }, "Attachment too large (>10MB) — skipping", ); reader.cancel(); return; } chunks.push(value); } } const imageBytes = Buffer.concat(chunks); const sniffedMime = sniffImageMimeType(imageBytes); if (!sniffedMime) { log.warn( { attachmentId: att.id }, "Skipping attachment: not a recognised image format", ); return; } const { data: resizedBuffer, mimeType: resizedMime } = await resizeImageForVision(imageBytes, maxDimension); const dataUrl = `data:${resizedMime};base64,${resizedBuffer.toString("base64")}`; const part: MessageImagePart = { type: "image_url", image_url: { url: dataUrl }, sourceLabel: `[gambar di atas adalah attachment ${att.filename} dari pesan id=${att.message_id}]`, }; const existing = imageMap.get(targetId) ?? []; if (existing.length < 8) { existing.push(part); imageMap.set(targetId, existing); } } catch (err) { log.warn( { attachmentId: att.id, error: err instanceof Error ? err.message : String(err), }, "Error downloading attachment", ); } finally { clearTimeout(timeoutId); } })(), ); } // ── URL fetch promises ── const urls = extractUrlsFromText(content).slice(0, 3); const urlWebTexts: string[] = []; for (const url of urls) { downloadPromises.push( (async () => { const result = await fetchUrlSafely(url); if (result.type === "image" && result.data && result.mimeType) { const { data: resizedBuffer, mimeType: resizedMime } = await resizeImageForVision(result.data, maxDimension); const dataUrl = `data:${resizedMime};base64,${resizedBuffer.toString("base64")}`; const part: MessageImagePart = { type: "image_url", image_url: { url: dataUrl }, sourceLabel: `[gambar di atas berasal dari link ${url} pada pesan id=${targetId}]`, }; const existing = imageMap.get(targetId) ?? []; if (existing.length < 8) { existing.push(part); imageMap.set(targetId, existing); } } else if (result.type === "text" && result.textContent) { urlWebTexts.push(`[Isi Web dari ${url}]: ${result.textContent}`); } })(), ); } // ── Sticker / embed / custom emoji download promises ── const mediaEvidence = extractMessageMediaEvidence(target.metadata); const mediaCandidates: Array<{ messageId: string; url: string; label: string; stickerName?: string; customEmojiId?: string; customEmojiName?: string; }> = [ ...mediaEvidence.stickers .filter((s) => s.url) .map((s) => ({ messageId: targetId, url: s.url, label: `[gambar di atas adalah sticker "${s.name}" dari pesan id=${targetId}]`, stickerName: s.name, })), ...mediaEvidence.embeds.flatMap((embed) => [ embed.image ? { messageId: targetId, url: embed.image, label: `[gambar di atas berasal dari embed image pada pesan id=${targetId}]`, } : null, embed.thumbnail ? { messageId: targetId, url: embed.thumbnail, label: `[gambar di atas berasal dari embed thumbnail pada pesan id=${targetId}]`, } : null, ].filter( ( c, ): c is { messageId: string; url: string; label: string; stickerName?: string; customEmojiId?: string; customEmojiName?: string; } => c !== null, ), ), ...mediaEvidence.customEmojis.map((emoji) => ({ messageId: targetId, url: emoji.url, label: `[gambar di atas adalah custom emoji "${emoji.name}" dari pesan id=${targetId}]`, customEmojiId: emoji.id, customEmojiName: emoji.name, })), ]; for (const candidate of mediaCandidates) { downloadPromises.push( (async () => { // Skip if we already have 8 images if ((imageMap.get(targetId)?.length ?? 0) >= 8) return; // Vision cache check before download (sticker & emoji keys only, since // their cache keys are consistent between check and store — embed URLs // use base64 data URL keys that never match the CDN URL). if (candidate.customEmojiId || candidate.stickerName) { const visionCacheKey = candidate.customEmojiId ? makeCustomEmojiCacheKey(candidate.customEmojiId) : makeStickerCacheKey(candidate.stickerName!); const cachedVision = await getCachedMediaAnalysis(visionCacheKey); if (cachedVision) { log.debug( { cacheKey: visionCacheKey }, "Vision cache HIT for media candidate — skipped download", ); const analysisText = `[Media analysis for message ${candidate.messageId}] ${candidate.label}: ${cachedVision}`; const existing = mediaAnalysisMap.get(targetId) ?? []; existing.push(analysisText); mediaAnalysisMap.set(targetId, existing); return; } } // Sticker download cache if (candidate.stickerName && isStickerCacheReady()) { try { const cached = await getStickerFromCache(candidate.stickerName); if (cached) { const part: MessageImagePart = { type: "image_url", image_url: { url: `data:${cached.mimeType};base64,${cached.base64}`, }, sourceLabel: candidate.label, stickerName: candidate.stickerName, }; const existing = imageMap.get(targetId) ?? []; if (existing.length < 8) { existing.push(part); imageMap.set(targetId, existing); } return; } } catch (stickerErr) { log.warn( { stickerName: candidate.stickerName, error: stickerErr instanceof Error ? stickerErr.message : String(stickerErr), }, "Sticker cache lookup failed — falling through to network fetch", ); } } const result = await fetchUrlSafely(candidate.url); if (result.type !== "image" || !result.data || !result.mimeType) { log.warn( { url: candidate.url, resultType: result.type, resultHasData: !!result.data, messageId: candidate.messageId, label: candidate.stickerName ? `sticker:${candidate.stickerName}` : candidate.customEmojiName ? `emoji:${candidate.customEmojiName}` : "embed/other", }, "Media candidate fetch did not return a usable image — skipping", ); return; } const { data: resizedBuffer, mimeType: resizedMime } = await resizeImageForVision(result.data, maxDimension); const base64 = resizedBuffer.toString("base64"); if (candidate.stickerName) { setStickerInCache(candidate.stickerName, base64, resizedMime).catch( () => {}, ); } const part: MessageImagePart = { type: "image_url", image_url: { url: `data:${resizedMime};base64,${base64}` }, sourceLabel: candidate.label, stickerName: candidate.stickerName, customEmojiId: candidate.customEmojiId, customEmojiName: candidate.customEmojiName, }; const existing = imageMap.get(targetId) ?? []; if (existing.length < 8) { existing.push(part); imageMap.set(targetId, existing); } })(), ); } // Execute ALL media downloads in parallel await Promise.all(downloadPromises); // Collect web text results from URL fetches if (urlWebTexts.length > 0) webTextMap.set(targetId, urlWebTexts); // ── 4. Vision analysis for every image ── await Promise.all( Array.from(imageMap.entries()).flatMap(([msgId, images]) => images.map(async (image) => { const summary = await analyzeSingleMediaImage(msgId, image); // summary is never null — always returns either the analysis or a failure description const existing = mediaAnalysisMap.get(msgId) ?? []; existing.push(summary); mediaAnalysisMap.set(msgId, existing); }), ), ); // ── 5. Build single-message prompt with XML delimiters (R1) ── const webTexts = webTextMap.get(targetId) ?? []; const mediaAnalyses = mediaAnalysisMap.get(targetId) ?? []; const webContext = webTexts.length > 0 ? `\n${webTexts.join("\n")}` : ""; const mediaAnalysisContext = mediaAnalyses.length > 0 ? `\n${mediaAnalyses.join("\n")}` : ""; const mediaContext = [ mediaEvidence.stickers.length > 0 ? mediaEvidence.stickers .map((s) => buildStickerTextOnlyWarning(s.name, s.url)) .join(" ") : null, mediaEvidence.embeds.length > 0 ? `[embed evidence: ${mediaEvidence.embeds .map((e) => [e.title, e.description, e.url, e.image, e.thumbnail] .filter(Boolean) .join(" | "), ) .join(" || ")}]` : null, ] .filter(Boolean) .join(" "); const channelId = target.channel_id; const channelCultureObj = channelId ? await getChannelCulture(channelId) : null; const channelCulture = channelCultureObj ? channelCultureObj.culture_summary : undefined; const rep = await initializeUserReputation(target.user_id, target.guild_id); const userCtx = ``; // XML delimiters wrap the message content (R1) const messageBlock = `\n ${userCtx}\n ${content}${mediaContext ? ` ${mediaContext}` : ""}${webContext}${mediaAnalysisContext}\n`; // Modular system prompt with XML delimiters (R1, R7, R8) const correctedExamples = await buildCorrectedFewShotExamples(); const systemText = buildSystemPromptModular({ contextText, mode: "mixed", correctedExamples, channelCulture, }); const userContent = `${systemText}\n\n\n${messageBlock}\n`; // ── 6. LLM call ── const result = await callModerationLLM( async (_state: RetryState) => userContent, targetIds, `media:${targetId}`, signal, ); return result; } // --------------------------------------------------------------------------- // Main entry point — splits text-only vs media, runs both paths in parallel // --------------------------------------------------------------------------- /** * Runs LLM-based moderation analysis on messages. * * Architecture: * - **Text-only messages** → single batch LLM call (fast, no image processing) * - Split into sub-batches if exceeding AI_LLM_TEXT_BATCH_SIZE (R6) * - **Media messages** → each gets its own LLM call with vision API (R5: resized images) * - Both paths execute **in parallel** — text batch does NOT wait for media. * - All LLM calls go through concurrency limiter (R3). */ export async function runModerationAnalysis( input: ModerationInput, ): Promise { const { targets, contextText, attachments } = input; if (!targets.length) { throw new Error("No targets provided for analysis"); } // ── Per-user moderation cache check ── // For text-only messages: check if we've already analyzed the same // (user, content) pair within the last 24 hours. If so, reuse the // cached result to save LLM calls (especially for repeat spam). const cacheHits: AnalysisResult[] = []; const uncachedTargets: MessageRecord[] = []; const seenCacheKeys = new Set(); // dedupe identical content within same batch for (const target of targets) { // Only cache text-only messages (media has dynamic image fetches) const hasMedia = hasMediaContent(target, attachments); if (hasMedia) { uncachedTargets.push(target); continue; } const rawContent = target.edited_content ?? target.content; if (!rawContent.trim()) { uncachedTargets.push(target); continue; } const cacheKey = makeTextModerationCacheKey(rawContent); // Deduplicate: if two identical messages from same user in this batch, // skip the cache lookup for the second and reuse the first's result. if (seenCacheKeys.has(cacheKey)) { // Synthesize a copy of the previous cache hit result for this duplicate const previousHit = cacheHits.find((h) => h.messageId !== target.id); if (previousHit) { cacheHits.push({ ...previousHit, messageId: target.id, }); } else { uncachedTargets.push(target); } continue; } seenCacheKeys.add(cacheKey); try { const cached = await getCachedTextModeration(cacheKey); if (cached) { // Safety: skip cache entries that are artifacts of API/parse errors. // A previous bug cached error results as "flagged", causing 24h false positives. // This guards against both legacy corrupt entries and any future write-path bugs. if ( cached.flags.some((f) => [ "analysis_api_failed", "analysis_parse_failed", "analysis_incomplete", ].includes(f), ) ) { log.warn( { messageId: target.id, cacheKey }, "Cache entry contains error artifact — treating as miss", ); } else { cacheHits.push({ messageId: target.id, status: cached.status, flags: cached.flags, score: cached.score, analysis: cached.analysis, categories: cached.categories, severity: cached.severity as AnalysisResult["severity"], confidence: cached.confidence, recommendedAction: cached.recommendedAction as AnalysisResult["recommendedAction"], policyVersion: "cached-user-moderation-2026-06", evidence: [], }); log.debug( { messageId: target.id, userId: target.user_id, cacheKey }, "User moderation cache HIT — reusing previous result", ); continue; } } } catch { // Cache lookup failed — proceed with uncached path } uncachedTargets.push(target); } if (cacheHits.length > 0) { log.info( { cacheHits: cacheHits.length, uncached: uncachedTargets.length, total: targets.length, }, "User moderation cache applied — skipping LLM call for cached targets", ); } // If all targets were cache hits, return early if (uncachedTargets.length === 0) { return { results: cacheHits, raw: null }; } // ── Split uncached targets ── const textOnlyTargets: MessageRecord[] = []; const mediaTargets: MessageRecord[] = []; for (const target of uncachedTargets) { if (hasMediaContent(target, attachments)) { mediaTargets.push(target); } else { textOnlyTargets.push(target); } } log.debug( { total: targets.length, textOnly: textOnlyTargets.length, media: mediaTargets.length, cacheHits: cacheHits.length, }, "Split uncached targets for parallel moderation analysis", ); // ── Run both paths in parallel ── const [textBatchResult, ...mediaResults] = await Promise.all([ // Text-only: one fast batch call (or multiple sub-batches) textOnlyTargets.length > 0 ? runTextOnlyBatch(textOnlyTargets, contextText) : Promise.resolve({ results: [] as AnalysisResult[], raw: null }), // Media: each message gets its own LLM call (all in parallel, but limited by semaphore — R3) ...mediaTargets.map((target) => runSingleMediaAnalysis(target, contextText, attachments), ), ]); // ── Store uncached text-only results in cache ── const textResults = textBatchResult.results; for (const result of textResults) { const target = textOnlyTargets.find((t) => t.id === result.messageId); if (!target) continue; const rawContent = target.edited_content ?? target.content; if (!rawContent.trim()) continue; // Do NOT cache error results (API failures, parse failures, incomplete). // Caching a transient error would turn it into a 24h false positive. if (result.status === "error") continue; const cacheKey = makeTextModerationCacheKey(rawContent); setCachedTextModeration(cacheKey, { flags: result.flags ?? [], score: result.score ?? 0, analysis: result.analysis ?? "", categories: result.categories ?? result.flags ?? [], severity: result.severity ?? "none", confidence: result.confidence ?? result.score ?? 0, recommendedAction: result.recommendedAction ?? "none", status: result.status, }).catch(() => {}); } // ── Merge cache hits + new results ── const allResults = [ ...cacheHits, ...textResults, ...mediaResults.flatMap((r) => r.results), ]; const raw = textBatchResult.raw ?? (mediaResults.length > 0 ? mediaResults[0].raw : null); log.debug( { targetCount: targets.length, resultCount: allResults.length, cacheHits: cacheHits.length, textBatchResults: textResults.length, mediaResults: mediaResults.length, }, "Moderation analysis complete", ); return { results: allResults, raw }; } // --------------------------------------------------------------------------- // Simple text-only fallback — uses a MINIMAL prompt that returns a single // word ("clean", "warn", or "flagged") instead of a complex JSON object. // // This is designed for cheap/small models that struggle with: // 1. Multi-target JSON output (confusing message_ids) // 2. Complex JSON schema compliance (9+ fields) // // Trade-off: less detail (no flags/evidence/categories), but ZERO parse // errors and much faster. Field values are derived heuristically. // --------------------------------------------------------------------------- /** * Simple two-step text fallback for cheap/small models. * * Step 1: Ask the LLM for a single-word classification (clean/warn/flagged). * Step 2: If not clean, ask the LLM again for a real reason — no dummy text. * * NO JSON at either step. Just raw text that we parse by simple rules. */ export async function runSimpleTextFallback( message: MessageRecord, ): Promise { const content = getAnalysisContent(message); const MAX_CONTENT_CHARS = 500; const truncatedContent = content.length > MAX_CONTENT_CHARS ? content.slice(0, MAX_CONTENT_CHARS) + "..." : content; // ── Step 1: Single-word classification ── const classifyPrompt = `Pesan berikut perlu diklasifikasikan sebagai: clean, warn, atau flagged. Aturan: - clean: pesan biasa, percakapan normal, tidak ada pelanggaran - warn: spam ringan, promosi tidak jelas, atau pelanggaran ringan - flagged: harassment, SARA, NSFW, judi, ancaman, atau pelanggaran serius PENTING: Slang Indonesia ("anjay", "wkwk", "njir", "gws", dll) dan makian umum ("asu", "anjing", "bangsat") yang TIDAK ditujukan ke orang lain = clean. Pesan: "${truncatedContent}" Jawab HANYA dengan satu kata: clean, warn, atau flagged`; let status: "clean" | "warn" | "flagged"; let rawClassify = ""; try { const completion = await llmChat({ messages: [{ role: "user", content: classifyPrompt }], max_tokens: 10, temperature: 0.1, }); rawClassify = completion?.choices[0]?.message?.content?.trim().toLowerCase() ?? ""; if (rawClassify.includes("flagged")) { status = "flagged"; } else if (rawClassify.includes("warn")) { status = "warn"; } else { status = "clean"; } log.info( { messageId: message.id, status, raw: rawClassify }, "Simple fallback step 1 — classification", ); } catch (error) { log.warn( { messageId: message.id, error: error instanceof Error ? error.message : String(error), }, "Simple fallback step 1 failed — defaulting to clean", ); status = "clean"; } // ── Step 2: Real analysis text + category (only if not clean) ── // We ask the LLM for a real reason and a category word — no complex JSON. let analysis: string; let category = ""; if (status === "clean") { analysis = `${message.username ?? "user"}: ${content.length > 200 ? content.slice(0, 200) + "..." : content}. Percakapan normal, tidak ada pelanggaran.`; } else { category = status === "flagged" ? "harassment" : "spam"; // default fallback const categoryOptions = status === "flagged" ? "harassment, gambling, atau sara" : "spam"; const reasonPrompt = `Pesan berikut telah diklasifikasikan sebagai "${status}". Pesan: "${truncatedContent}" Jelaskan dalam 1-2 kalimat Bahasa Indonesia: APA yang melanggar dan KENAPA. Jangan gunakan kata "mungkin" atau "sepertinya". Jangan tulis ulang pesan. Langsung ke alasan. Setelah alasan, sebutkan Kategori: ${categoryOptions} Contoh untuk "flagged": Mengandung kata kasar terarah ke individu tertentu sebagai hinaan. Kategori: harassment Contoh untuk "flagged": Promosi situs judi online dengan link dan ajakan. Kategori: gambling Contoh untuk "warn": Promosi channel Discord tanpa konteks, berpotensi spam. Kategori: spam Contoh untuk "warn": Bahasa kasar ringan yang tidak terarah. Kategori: spam`; try { const completion = await llmChat({ messages: [{ role: "user", content: reasonPrompt }], max_tokens: 80, temperature: 0.3, }); analysis = completion?.choices[0]?.message?.content?.trim() ?? ""; // Guard against empty or non-answers if (!analysis || analysis.length < 5) { analysis = `Pesan diklasifikasikan sebagai ${status} oleh sistem moderasi otomatis.`; } // ── Parse category from "Kategori: xxx" line ── const categoryMatch = analysis.match(/[Kk]ategori:\s*(\w+)/i); if (categoryMatch) { const parsedCat = categoryMatch[1].toLowerCase(); // Only accept known categories if (["harassment", "spam", "gambling", "sara"].includes(parsedCat)) { category = parsedCat; } // Strip the "Kategori:" line from the analysis text so it's cleaner analysis = analysis.replace(/[Kk]ategori:\s*\w+\s*/i, "").trim(); } log.info( { messageId: message.id, status, category, analysis: analysis.slice(0, 100), }, "Simple fallback step 2 — reason + category", ); } catch (error) { analysis = `Pesan diklasifikasikan sebagai ${status} oleh sistem moderasi otomatis berdasarkan analisis konten.`; log.warn( { messageId: message.id, error: error instanceof Error ? error.message : String(error), }, "Simple fallback step 2 failed — using fallback reason text", ); } } // Build the result fields using parsed category const flags: string[] = status === "clean" ? [] : [category]; const categories: string[] = status === "clean" ? [] : [category]; const score = status === "flagged" ? 0.7 : status === "warn" ? 0.4 : 0; const severity: "none" | "low" | "medium" | "high" | "critical" = status === "flagged" ? "medium" : status === "warn" ? "low" : "none"; const confidence = 0.6; return { messageId: message.id, status, flags, score, analysis, categories, severity, confidence, recommendedAction: status === "flagged" ? "review" : status === "warn" ? "warn" : "none", policyVersion: "default-simple-2026-06", evidence: status !== "clean" ? [content.length > 120 ? content.slice(0, 120) + "..." : content] : [], }; }