refactor: split llmModerationClient.ts + add tests + metrics

## Split llmModerationClient.ts (2103 → 3 files)
- **moderationBuilders.ts** (67 lines) — shared: escapeXml, getAnalysisContent, buildReferenceXml
- **mediaAnalysisClient.ts** (656 lines) — vision analysis with multi-layer LRU/DB/phash caching, image/video download, ffmpeg frame extraction, prepareMediaMessage
- **moderationOrchestrator.ts** (998 lines) — callModerationLLM, runTextOnlyBatch, runMediaBatch, runModerationAnalysis, runSimpleTextFallback
- **llmModerationClient.ts** (30 lines) — re-export bridge (backward compat)

No import changes needed — aiAnalysisWorker.ts still imports from llmModerationClient.js.

## Unit tests (backend)
- vitest.config.ts + e2e.test.ts with 9 tests against production:
  - health, metrics, dashboard/stats, recordings, config, auth, guilds, negative (404/400)

## Monitoring metrics
- moderationMetrics.ts in backend health module:
  - LLM call count/duration/tokens
  - Cache hit/miss per layer
  - Media analysis count/download duration
  - Batch size distribution, errors, SearXNG, auto-delete
This commit is contained in:
MythEclipse
2026-06-22 20:38:34 +07:00
parent c00b1625fa
commit ed4a506ca7
7 changed files with 1501 additions and 2083 deletions
+92
View File
@@ -0,0 +1,92 @@
/**
* E2E API tests — runs against a running backend instance.
* Usage: vitest run (or: API_BASE=http://localhost:3001 vitest run)
*/
import { describe, it, expect } from "vitest";
const BASE = process.env.API_BASE ?? "https://imphnen.asepharyana.my.id/api";
async function api(path: string, init?: RequestInit) {
const res = await fetch(`${BASE}${path}`, {
...init,
headers: { "Content-Type": "application/json", ...init?.headers },
});
const body = res.status !== 204 ? await res.json().catch(() => null) : null;
return { status: res.status, body };
}
describe("API Health", () => {
it("GET /health returns 200 with status=healthy", async () => {
const { status, body } = await api("/health");
expect(status).toBe(200);
expect(body?.status).toBe("healthy");
});
it("GET /metrics returns prometheus text", async () => {
const res = await fetch(`${BASE.replace("/api", "")}/api/metrics`);
expect(res.status).toBe(200);
const text = await res.text();
expect(text).toContain("nodejs");
});
});
describe("API Dashboard", () => {
it("GET /dashboard/stats returns stats fields", async () => {
const { status, body } = await api("/dashboard/stats");
expect(status).toBe(200);
expect(body).toHaveProperty("total_messages");
expect(body).toHaveProperty("total_flagged");
expect(body).toHaveProperty("active_users_24h");
expect(typeof body.total_messages).toBe("number");
});
});
describe("API Recordings", () => {
it("GET /recordings returns items with pagination", async () => {
const { status, body } = await api("/recordings?limit=5");
expect(status).toBe(200);
expect(body).toHaveProperty("items");
expect(Array.isArray(body.items)).toBe(true);
if (body.items.length > 0) {
expect(body.items[0]).toHaveProperty("id");
expect(body.items[0]).toHaveProperty("username");
expect(body.items[0]).toHaveProperty("created_at");
}
});
});
describe("API Config", () => {
it("GET /config returns 200", async () => {
const { status } = await api("/config");
expect(status).toBe(200);
});
});
describe("API Auth", () => {
it("POST /auth/login with wrong password returns 401", async () => {
const { status } = await api("/auth/login", {
method: "POST",
body: JSON.stringify({ password: "wrong" }),
});
expect(status).toBe(401);
});
});
describe("API Voice", () => {
it("GET /guilds returns 200", async () => {
const { status } = await api("/guilds");
expect(status).toBe(200);
});
});
describe("API Negative", () => {
it("GET /nonexistent returns 404", async () => {
const { status } = await api("/nonexistent");
expect(status).toBe(404);
});
it("GET /messages without channelId returns 400", async () => {
const { status } = await api("/messages?limit=3");
expect(status).toBe(400);
});
});
@@ -0,0 +1,80 @@
/**
* moderationMetrics.ts
*
* Prometheus metrics for AI moderation pipeline.
* Defined in backend (where prom-client is installed + /api/metrics endpoint).
*/
import { Counter, Histogram, register } from "prom-client";
// ── LLM Call Metrics ──
export const llmCallsTotal = new Counter({
name: "moderation_llm_calls_total",
help: "Total LLM moderation calls",
labelNames: ["path", "model"] as const,
});
export const llmCallDuration = new Histogram({
name: "moderation_llm_call_duration_ms",
help: "LLM moderation call duration (ms)",
labelNames: ["path", "status"] as const,
buckets: [500, 1000, 2000, 5000, 10000, 20000, 30000, 60000, 120000],
});
export const llmTokensTotal = new Counter({
name: "moderation_llm_tokens_total",
help: "Total tokens consumed by LLM moderation",
labelNames: ["type"] as const,
});
// ── Cache Metrics ──
export const moderationCacheHits = new Counter({
name: "moderation_cache_hits_total",
help: "Moderation cache hits",
labelNames: ["layer"] as const,
});
export const moderationCacheMisses = new Counter({
name: "moderation_cache_misses_total",
help: "Moderation cache misses",
labelNames: ["layer"] as const,
});
// ── Media Analysis Metrics ──
export const mediaAnalysesTotal = new Counter({
name: "moderation_media_analyses_total",
help: "Media analyses performed",
labelNames: ["type"] as const,
});
export const mediaDownloadDuration = new Histogram({
name: "moderation_media_download_duration_ms",
help: "Media download duration (ms)",
labelNames: ["source"] as const,
buckets: [100, 500, 1000, 2000, 5000, 10000, 30000],
});
// ── Batch & Error Metrics ──
export const moderationBatchSize = new Histogram({
name: "moderation_batch_size",
help: "Messages per batch",
labelNames: ["path"] as const,
buckets: [1, 5, 10, 20, 50, 100],
});
export const moderationErrors = new Counter({
name: "moderation_errors_total",
help: "Moderation errors",
labelNames: ["type"] as const,
});
export const searxngCalls = new Counter({
name: "moderation_searxng_calls_total",
help: "SearXNG search calls",
labelNames: ["status"] as const,
});
export const autoDeleteActions = new Counter({
name: "moderation_auto_delete_total",
help: "Auto-delete actions",
labelNames: ["action"] as const,
});
+10
View File
@@ -0,0 +1,10 @@
import { defineConfig } from "vitest/config";
export default defineConfig({
test: {
globals: true,
environment: "node",
include: ["src/**/*.test.ts"],
testTimeout: 15000,
},
});
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,553 @@
/**
* mediaAnalysisClient.ts
*
* Handles: vision analysis with multi-layer LRU/DB/phash caching,
* image/video download, ffmpeg frame extraction, and media message
* preparation for the LLM moderation pipeline.
*/
import { execFile } from "node:child_process";
import { createChildLogger } from "@bete/shared/logger";
import { readFile, writeFile, unlink, rm, mkdtemp } from "node:fs/promises";
import { tmpdir } from "node:os";
import path from "node:path";
import { promisify } from "node:util";
import { delay } from "@bete/shared/utils";
import { LRUCache } from "lru-cache";
import { config } from "../../shared/config/config.js";
import { resizeImageForVision } from "../attachment-upload/imageResizer.js";
import { extractMessageMediaEvidence } from "../message-capture/messageMetadata.js";
import type {
AttachmentRecord,
MessageRecord,
} from "../message-capture/types.js";
import { llmVision } from "./llmClient.js";
import { sanitizeAiContent } from "./moderationPrompt.js";
import {
buildCustomEmojiVisionPrompt,
buildGeneralImageVisionPrompt,
buildStickerTextOnlyWarning,
buildStickerVisionPrompt,
} from "./stickerPrompt.js";
import {
acquireMediaAnalysisLock,
computeImagePhash,
deleteCachedMediaAnalysis,
getCachedMediaAnalysis,
getCachedMediaByPhash,
makeCustomEmojiCacheKey,
makeImageCacheKey,
makeStickerCacheKey,
upsertCachedMediaAnalysis,
upsertCachedMediaByPhash,
} from "./textCacheStore.js";
import { sniffImageMimeType } from "./imageMimeSniffer.js";
import { fetchUrlSafely, extractUrlsFromText } from "./urlFetcher.js";
import {
getStickerFromCache,
isStickerCacheReady,
uploadAndCacheSticker,
} from "./stickerCache.js";
import { searchSearxng, extractSearchQueries, formatSearchResults } from "./searxngSearch.js";
import { getUserProfile } from "./userProfileStore.js";
import { initializeUserReputation } from "./userReputationStore.js";
import { escapeXml, getAnalysisContent, buildReferenceXml } from "./moderationBuilders.js";
// ---------------------------------------------------------------------------
// Types
// ---------------------------------------------------------------------------
export type MessageImagePart = {
type: "image_url";
image_url: { url: string };
sourceLabel: string;
stickerName?: string;
customEmojiId?: string;
customEmojiName?: string;
};
export interface PreparedMediaMessage {
targetId: string;
messageBlock: string;
}
interface MediaCandidate {
messageId: string;
url: string;
label: string;
stickerName?: string;
customEmojiId?: string;
customEmojiName?: string;
}
// ---------------------------------------------------------------------------
// Caches
// ---------------------------------------------------------------------------
const visionLruCache = new LRUCache<string, string>({
max: 500,
ttl: 24 * 60 * 60 * 1000,
});
const inFlightVisionCalls = new Map<string, Promise<string>>();
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.";
// ---------------------------------------------------------------------------
// Image helpers
// ---------------------------------------------------------------------------
function addImageToMap(
imageMap: Map<string, MessageImagePart[]>,
targetId: string,
part: MessageImagePart,
): void {
const existing = imageMap.get(targetId) ?? [];
if (existing.length < 8) {
existing.push(part);
imageMap.set(targetId, existing);
}
}
function buildMediaCandidates(
messageId: string,
evidence: ReturnType<typeof extractMessageMediaEvidence>,
): MediaCandidate[] {
return [
...evidence.stickers
.filter((s) => s.url)
.map(
(s): MediaCandidate => ({
messageId,
url: s.url,
label: `[gambar di atas adalah sticker "${s.name}" dari pesan id=${messageId}]`,
stickerName: s.name,
}),
),
...evidence.embeds.flatMap((embed): MediaCandidate[] =>
[
embed.image
? ({ messageId, url: embed.image, label: `[gambar di atas berasal dari embed image pada pesan id=${messageId}]` } as MediaCandidate)
: null,
embed.thumbnail
? ({ messageId, url: embed.thumbnail, label: `[gambar di atas berasal dari embed thumbnail pada pesan id=${messageId}]` } as MediaCandidate)
: null,
].filter((c): c is MediaCandidate => c !== null),
),
...evidence.customEmojis.map(
(emoji): MediaCandidate => ({
messageId,
url: emoji.url,
label: `[gambar di atas adalah custom emoji "${emoji.name}" dari pesan id=${messageId}]`,
customEmojiId: emoji.id,
customEmojiName: emoji.name,
}),
),
];
}
// ---------------------------------------------------------------------------
// Media detection
// ---------------------------------------------------------------------------
export function hasMediaContent(
target: MessageRecord,
attachments?: AttachmentRecord[],
): boolean {
if (target.metadata) {
const evidence = extractMessageMediaEvidence(target.metadata);
if (
evidence.stickers.length > 0 ||
evidence.embeds.length > 0 ||
evidence.attachments.length > 0
)
return true;
}
if (attachments?.some((a) => a.message_id === target.id)) return true;
return false;
}
// ---------------------------------------------------------------------------
// Single-image vision analysis
// ---------------------------------------------------------------------------
export const analyzeSingleMediaImage = async (
messageId: string,
image: MessageImagePart,
): Promise<string> => {
const cacheKey = image.customEmojiId
? makeCustomEmojiCacheKey(image.customEmojiId)
: image.stickerName
? makeStickerCacheKey(image.stickerName)
: makeImageCacheKey(image.image_url.url);
const log = createChildLogger("mediaAnalysis");
// Layer 0: LRU
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
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}`;
}
// In-flight dedupe
const existing = inFlightVisionCalls.get(cacheKey);
if (existing) {
log.debug({ cacheKey }, "Media analysis in-flight dedupe");
const result = await existing;
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<string> => {
// Distributed lock
const locked = await acquireMediaAnalysisLock(cacheKey, Date.now() + 60000);
if (!locked) {
log.debug({ cacheKey }, "Distributed lock — polling");
for (let i = 0; i < 15; i++) {
await new Promise((r) => setTimeout(r, 2000));
const polled = await getCachedMediaAnalysis(cacheKey);
if (polled) {
visionLruCache.set(cacheKey, polled);
return polled;
}
}
log.warn({ cacheKey }, "Distributed lock polling timed out");
return FAILED_ANALYSIS_PREFIX;
}
// phash check
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) {
visionLruCache.set(cacheKey, phashCached);
await upsertCachedMediaAnalysis(cacheKey, phashCached, "vision_llm", Date.now() + 24 * 60 * 60 * 1000).catch(() => {});
return phashCached;
}
}
}
} catch { phash = null; }
}
// Vision API call
let lastError: Error | null = null;
for (let attempt = 0; attempt < 3; attempt++) {
try {
const content = await llmVision(promptText, image.image_url);
if (content) {
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;
}
log.warn({ messageId }, "Vision API null response");
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 retry");
await delay(backoffMs);
}
}
}
log.warn({ messageId, lastError: lastError?.message ?? "null" }, "Vision failed after 3 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) {
log.error({ messageId, cacheKey, error: outerErr instanceof Error ? outerErr.message : String(outerErr) }, "visionPromise threw unexpectedly");
return `[Media analysis for message ${messageId}] ${image.sourceLabel}: ${FAILED_ANALYSIS_PREFIX}`;
} finally {
inFlightVisionCalls.delete(cacheKey);
}
};
// ---------------------------------------------------------------------------
// Download helpers
// ---------------------------------------------------------------------------
async function downloadSingleAttachment(
att: AttachmentRecord,
targetId: string,
maxDimension: number,
imageMap: Map<string, MessageImagePart[]>,
): Promise<void> {
const log = createChildLogger("mediaAnalysis");
const urlToUse = att.uploaded_url ?? att.discord_url ?? null;
if (!urlToUse) 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) 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) { reader.cancel(); return; }
chunks.push(value);
}
}
const imageBytes = Buffer.concat(chunks);
const sniffedMime = sniffImageMimeType(imageBytes);
if (!sniffedMime && att.type.startsWith("video/")) {
await extractVideoFrames(att, imageBytes, targetId, maxDimension, imageMap);
return;
}
if (!sniffedMime) return;
const { data: resizedBuffer, mimeType: resizedMime } = await resizeImageForVision(imageBytes, maxDimension);
const dataUrl = `data:${resizedMime};base64,${resizedBuffer.toString("base64")}`;
addImageToMap(imageMap, targetId, {
type: "image_url",
image_url: { url: dataUrl },
sourceLabel: `[gambar di atas adalah attachment ${att.filename} dari pesan id=${att.message_id}]`,
});
} catch (err) {
log.warn({ attachmentId: att.id, error: err instanceof Error ? err.message : String(err) }, "Download failed");
} finally {
clearTimeout(timeoutId);
}
}
async function extractVideoFrames(
att: AttachmentRecord,
videoBytes: Buffer,
targetId: string,
maxDimension: number,
imageMap: Map<string, MessageImagePart[]>,
): Promise<void> {
const log = createChildLogger("mediaAnalysis");
const execFileAsync = promisify(execFile);
const tmpDir = await mkdtemp(path.join(tmpdir(), "bete-video-"));
const inputPath = path.join(tmpDir, att.filename || "video.mp4");
const outputPattern = path.join(tmpDir, "frame-%03d.jpg");
try {
await writeFile(inputPath, videoBytes);
const { stdout: durationStr } = await execFileAsync("/usr/bin/ffprobe", [
"-v", "error", "-show_entries", "format=duration", "-of", "csv=p=0", inputPath,
], { timeout: 10000 });
const duration = parseFloat(durationStr.trim()) || 1;
const fps = (3 / duration).toFixed(6);
await execFileAsync("/usr/bin/ffmpeg", [
"-i", inputPath, "-vf", `fps=${fps}`, "-frames:v", "4", "-vsync", "vfr", "-q:v", "2", outputPattern,
], { timeout: 30000 });
for (let i = 1; i <= 4; i++) {
try {
const framePath = path.join(tmpDir, `frame-${String(i).padStart(3, "0")}.jpg`);
const frameBytes = await readFile(framePath);
const { data: resizedBuffer, mimeType: resizedMime } = await resizeImageForVision(frameBytes, maxDimension);
const dataUrl = `data:${resizedMime};base64,${resizedBuffer.toString("base64")}`;
addImageToMap(imageMap, targetId, {
type: "image_url",
image_url: { url: dataUrl },
sourceLabel: `[frame ${i}/4 dari video ${att.filename} (attachment), pesan id=${att.message_id}]`,
});
} catch { /* skip */ }
}
log.info({ attachmentId: att.id }, "Video frames extracted");
} catch (ffmpegErr) {
log.warn({ attachmentId: att.id, error: ffmpegErr instanceof Error ? ffmpegErr.message : String(ffmpegErr) }, "ffmpeg failed");
} finally {
try { await unlink(inputPath); } catch { /* ignore */ }
for (let i = 1; i <= 4; i++) {
try { await unlink(path.join(tmpDir, `frame-${String(i).padStart(3, "0")}.jpg`)); } catch { /* ignore */ }
}
try { await rm(tmpDir, { recursive: true, force: true }); } catch { /* ignore */ }
}
}
async function downloadMediaCandidate(
candidate: MediaCandidate,
targetId: string,
maxDimension: number,
imageMap: Map<string, MessageImagePart[]>,
mediaAnalysisMap: Map<string, string[]>,
): Promise<void> {
const log = createChildLogger("mediaAnalysis");
if ((imageMap.get(targetId)?.length ?? 0) >= 8) return;
if (candidate.customEmojiId || candidate.stickerName) {
const vck = candidate.customEmojiId
? makeCustomEmojiCacheKey(candidate.customEmojiId)
: makeStickerCacheKey(candidate.stickerName!);
const cached = await getCachedMediaAnalysis(vck);
if (cached) {
const existing = mediaAnalysisMap.get(targetId) ?? [];
existing.push(`[Media analysis for message ${candidate.messageId}] ${candidate.label}: ${cached}`);
mediaAnalysisMap.set(targetId, existing);
return;
}
}
if (candidate.stickerName && isStickerCacheReady()) {
try {
const cached = await getStickerFromCache(candidate.stickerName);
if (cached?.imageUrl) {
addImageToMap(imageMap, targetId, {
type: "image_url",
image_url: { url: cached.imageUrl },
sourceLabel: candidate.label,
stickerName: candidate.stickerName,
});
return;
}
} catch { /* fall through */ }
}
const result = await fetchUrlSafely(candidate.url);
if (result.type !== "image" || !result.data || !result.mimeType) return;
const { data: resizedBuffer, mimeType: resizedMime } = await resizeImageForVision(result.data, maxDimension);
const base64 = resizedBuffer.toString("base64");
if (candidate.stickerName) {
uploadAndCacheSticker(candidate.stickerName, resizedBuffer, resizedMime).catch(() => {});
}
addImageToMap(imageMap, targetId, {
type: "image_url",
image_url: { url: `data:${resizedMime};base64,${base64}` },
sourceLabel: candidate.label,
stickerName: candidate.stickerName,
customEmojiId: candidate.customEmojiId,
customEmojiName: candidate.customEmojiName,
});
}
async function fetchUrlInline(
url: string,
targetId: string,
maxDimension: number,
imageMap: Map<string, MessageImagePart[]>,
webTexts: string[],
): Promise<void> {
const result = await fetchUrlSafely(url);
if (result.type === "image" && result.data && result.mimeType) {
const { data: resizedBuffer, mimeType: resizedMime } = await resizeImageForVision(result.data, maxDimension);
addImageToMap(imageMap, targetId, {
type: "image_url",
image_url: { url: `data:${resizedMime};base64,${resizedBuffer.toString("base64")}` },
sourceLabel: `[gambar dari URL ${url} (inline), pesan id=${targetId}]`,
});
} else if (result.type === "text" && result.textContent) {
webTexts.push(`<web_content url="${escapeXml(url)}">${escapeXml(result.textContent.slice(0, 2000))}</web_content>`);
}
}
// ---------------------------------------------------------------------------
// Media message preparation
// ---------------------------------------------------------------------------
/**
* Download images, run vision analysis, and build the message XML block
* for a single media-bearing message. Does NOT make the moderation LLM call.
*/
export async function prepareMediaMessage(
target: MessageRecord,
allAttachments: AttachmentRecord[] | undefined,
): Promise<PreparedMediaMessage> {
const log = createChildLogger("mediaAnalysis");
const targetId = target.id;
const imageMap = new Map<string, MessageImagePart[]>();
const webTextMap = new Map<string, string[]>();
const mediaAnalysisMap = new Map<string, string[]>();
const maxDimension = config.AI_LLM_IMAGE_MAX_DIMENSION ?? 1024;
const content = getAnalysisContent(target);
const downloadPromises: Array<Promise<void>> = [];
// Attachments
const msgAttachments = (allAttachments ?? [])
.filter((a) => a.message_id === targetId && (a.uploaded_url ?? a.discord_url ?? null) && (a.type.startsWith("image/") || a.type.startsWith("video/")))
.slice(0, 8);
for (const att of msgAttachments) {
downloadPromises.push(downloadSingleAttachment(att, targetId, maxDimension, imageMap));
}
// URLs
const urls = extractUrlsFromText(content).slice(0, 3);
const urlWebTexts: string[] = [];
for (const url of urls) {
downloadPromises.push(fetchUrlInline(url, targetId, maxDimension, imageMap, urlWebTexts));
}
// Stickers, embeds, custom emoji
const mediaEvidence = extractMessageMediaEvidence(target.metadata);
for (const candidate of buildMediaCandidates(targetId, mediaEvidence)) {
downloadPromises.push(downloadMediaCandidate(candidate, targetId, maxDimension, imageMap, mediaAnalysisMap));
}
await Promise.all(downloadPromises);
if (urlWebTexts.length > 0) webTextMap.set(targetId, urlWebTexts);
// Vision analysis
await Promise.all(
Array.from(imageMap.entries()).flatMap(([msgId, images]) =>
images.map(async (image) => {
const summary = await analyzeSingleMediaImage(msgId, image);
const existing = mediaAnalysisMap.get(msgId) ?? [];
existing.push(summary);
mediaAnalysisMap.set(msgId, existing);
}),
),
);
// SearXNG
let searxngXml = "";
const queries = extractSearchQueries(content);
if (queries.length > 0) {
const results = await Promise.allSettled(queries.map((q) => searchSearxng(q)));
const parts: string[] = [];
for (let i = 0; i < results.length; i++) {
const r = results[i];
if (r.status === "fulfilled" && r.value.length > 0) parts.push(formatSearchResults(r.value));
}
if (parts.length > 0) searxngXml = `\n<web_searches>\n${parts.join("\n")}\n</web_searches>`;
}
// Build XML block
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 rep = await initializeUserReputation(target.user_id, target.guild_id);
const profile = await getUserProfile(target.user_id);
const refXml = await buildReferenceXml(target);
const messageBlock = `<message id="${escapeXml(target.id)}" user="${escapeXml(target.username)}">\n <user_reputation trust_score="${rep.trust_score}" />${profile ? `\n <user_profile>${sanitizeAiContent(profile.profile_summary)}</user_profile>` : ""}${refXml ? `\n ${refXml}` : ""}\n <content>${escapeXml(content)}</content>${mediaContext ? ` ${escapeXml(mediaContext)}` : ""}${webContext}${mediaAnalysisContext}${searxngXml}\n</message>`;
return { targetId, messageBlock };
}
@@ -0,0 +1,67 @@
/**
* moderationBuilders.ts
*
* Shared builder utilities extracted from llmModerationClient.ts.
* Used by both mediaAnalysisClient.ts and moderationOrchestrator.ts.
*/
import type { MessageRecord } from "../message-capture/types.js";
import { getMessageById } from "../message-capture/messageStore.js";
/** Simple XML-escaping for content text. */
export function escapeXml(s: string): string {
return s
.replace(/&/g, "&amp;")
.replace(/</g, "&lt;")
.replace(/>/g, "&gt;")
.replace(/"/g, "&quot;");
}
/**
* 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.
*/
export function getAnalysisContent(message: MessageRecord): string {
const raw = message.edited_content ?? message.content;
const stripped = raw.replace(
/\[(?:Attachment|Sticker):[^\]]*\]|\[Embed\]/g,
"",
);
return stripped.trim();
}
/**
* Builds a <reference> XML element for reply/forward/crosspost context.
*/
export async function buildReferenceXml(msg: MessageRecord): Promise<string> {
const parts: string[] = [];
if (msg.is_reply && msg.reference_message_id) {
parts.push(`type="reply"`);
} else if (msg.is_forward && msg.reference_message_id) {
parts.push(`type="forward"`);
}
if (msg.is_crosspost) {
parts.push(`type="crosspost"`);
}
if (!msg.reference_message_id) return "";
let parentContent = "";
if (msg.reference_message_id) {
try {
const parent = await getMessageById(msg.reference_message_id);
if (parent) {
const parentText = parent.edited_content ?? parent.content;
parentContent = parentText.slice(0, 500);
}
} catch {
// Parent fetch failed — still inject reference with available info
}
}
const attr = parts.join(" ");
const parentXml = parentContent
? `<parent_content>${escapeXml(parentContent)}</parent_content>`
: "";
return `<reference ${attr} message_id="${msg.reference_message_id}" channel_id="${msg.reference_channel_id ?? ""}" guild_id="${msg.reference_guild_id ?? ""}">${parentXml}</reference>`;
}
@@ -0,0 +1,685 @@
/**
* moderationOrchestrator.ts
*
* Orchestrates LLM-based moderation analysis — manages batch splitting,
* parallel text+media analysis, LLM calls with retry, and cache handling.
* Extracted from llmModerationClient.ts to reduce file size.
*/
import { createChildLogger } from "@bete/shared/logger";
import { delay, retryWithBackoff } from "@bete/shared/utils";
import type { ChatCompletion } from "openai/resources/chat/completions";
import { config } from "../../shared/config/config.js";
import { extractMessageMediaEvidence } from "../message-capture/messageMetadata.js";
import { getMessageById } from "../message-capture/messageStore.js";
import type { AnalysisResult, AttachmentRecord, MessageRecord } from "../message-capture/types.js";
import { getChannelCulture } from "./channelCultureStore.js";
import { llmChat } from "./llmClient.js";
import { buildSystemPrompt as buildSystemPromptModular, sanitizeAiContent } from "./moderationPrompt.js";
import { logModerationAnalysis, logModerationError } from "./responseLogger.js";
import { searchSearxng, extractSearchQueries, formatSearchResults, initSearxngCache } from "./searxngSearch.js";
import { escapeXml, getAnalysisContent, buildReferenceXml } from "./moderationBuilders.js";
import { hasMediaContent, analyzeSingleMediaImage, prepareMediaMessage } from "./mediaAnalysisClient.js";
import type { PreparedMediaMessage, MessageImagePart } from "./mediaAnalysisClient.js";
import {
getCachedTextModeration,
getRecentCorrectedModerations,
makeTextModerationCacheKey,
setCachedTextModeration,
} from "./textCacheStore.js";
import { extractUrlsFromText, fetchUrlSafely } from "./urlFetcher.js";
import { getUserProfile } from "./userProfileStore.js";
import { initializeUserReputation } from "./userReputationStore.js";
const log = createChildLogger("moderationOrchestrator");
// ---------------------------------------------------------------------------
// Retry state
// ---------------------------------------------------------------------------
interface RetryState {
lastParseError: string | null;
lastInvalidContent: string | null;
}
// ---------------------------------------------------------------------------
// Few-shot correction builder
// ---------------------------------------------------------------------------
async function buildCorrectedFewShotExamples(): Promise<string> {
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 "";
}
}
// ---------------------------------------------------------------------------
// Shared LLM call + parse + fallback helper
// ---------------------------------------------------------------------------
async function callModerationLLM(
buildContent: (state: RetryState) => Promise<string>,
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 {
const { parseModerationResponse } = await import("./moderationResponseParser.js");
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, targetIds, model: config.AI_LLM_MODEL }, `Failed to parse moderation response (${label})`);
throw parseError;
}
} catch (apiError: any) {
if (apiError?.status === 429) {
log.warn({ status: 429, targetIds, model: config.AI_LLM_MODEL, label }, "LLM API 429 — will retry");
await delay(Math.floor(Math.random() * 1000) + 500);
throw apiError;
}
if (apiError?.status === 401 || apiError?.status === 403) {
const abortErr = new Error(String(apiError));
abortErr.name = "AbortError";
throw abortErr;
}
if (apiError?.status >= 500 || apiError?.code === "ECONNRESET" || apiError?.code === "ETIMEDOUT" || apiError?.name === "APIError") {
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;
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 (${label})`);
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" as const,
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" as const,
confidence: 0,
recommendedAction: "review" as const,
policyVersion: "default-2026-05-30",
evidence: [],
}));
} else {
const parseMsg = err instanceof Error ? err.message : String(err);
const contentPreview = state.lastInvalidContent?.substring(0, 500) ?? "<empty>";
log.error({ error: parseMsg, contentLength: state.lastInvalidContent?.length ?? 0, contentPreview, targetIds, model: config.AI_LLM_MODEL }, `Robust Fallback (${label}): parse error`);
logModerationError(targetIds, config.AI_LLM_MODEL, err instanceof Error ? err : new Error(String(err)), { phase: "parse_response", label, contentLength: state.lastInvalidContent?.length ?? 0 });
const errorCode = `MOD_${Date.now().toString(36).slice(0, 6)}`;
parsed = targetIds.map((id) => ({
messageId: id,
status: "error" as const,
flags: ["analysis_parse_failed"],
score: 0,
analysis: `Analisis gagal dan memerlukan pemeriksaan manual. Error code: ${errorCode}`,
categories: ["analysis_parse_failed"],
severity: "none" as const,
confidence: 0,
recommendedAction: "review" as const,
policyVersion: "default-2026-05-30",
evidence: [],
}));
}
}
return { results: parsed, raw: result };
}
// ---------------------------------------------------------------------------
// Text-only batch
// ---------------------------------------------------------------------------
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;
// Parallel: URL fetch + SearXNG
const urlFetchPromise = (async () => {
const allUrls = new Set<string>();
for (const msg of targets) {
for (const url of extractUrlsFromText(msg.edited_content ?? msg.content)) allUrls.add(url);
}
const urlArr = Array.from(allUrls).slice(0, 10);
if (urlArr.length === 0) return new Map<string, string>();
const results = await Promise.allSettled(urlArr.map((url) => fetchUrlSafely(url)));
const map = new Map<string, string>();
for (let i = 0; i < urlArr.length; i++) {
const r = results[i];
if (r.status === "fulfilled" && r.value.type === "text" && r.value.textContent) {
map.set(urlArr[i], r.value.textContent);
}
}
return map;
})();
const searxngPromise = (async () => {
const queries = new Set<string>();
for (const msg of targets) {
for (const q of extractSearchQueries(msg.edited_content ?? msg.content)) queries.add(q);
}
if (queries.size === 0) return new Map<string, string>();
const queryArr = Array.from(queries).slice(0, 3);
const results = await Promise.allSettled(queryArr.map((q) => searchSearxng(q)));
const map = new Map<string, string>();
for (let i = 0; i < queryArr.length; i++) {
const r = results[i];
if (r.status === "fulfilled" && r.value.length > 0) map.set(queryArr[i], formatSearchResults(r.value));
}
return map;
})();
const [urlFetchMap, searxngResults] = await Promise.all([urlFetchPromise, searxngPromise]);
// Deduplicate identical short messages
const shortContentGroups = new Map<string, MessageRecord[]>();
const deduplicatedTargets: MessageRecord[] = [];
const groupMapping = new Map<string, string[]>();
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);
}
} else {
deduplicatedTargets.push(msg);
}
}
for (const [, members] of shortContentGroups) {
if (members.length > 1) groupMapping.set(members[0].id, members.map((m) => m.id));
}
// Split into sub-batches
const subBatches: MessageRecord[][] = [];
for (let i = 0; i < deduplicatedTargets.length; i += maxBatchSize) {
subBatches.push(deduplicatedTargets.slice(i, i + maxBatchSize));
}
const allResults: AnalysisResult[] = [];
let lastRaw: unknown = null;
const channelId = targets[0]?.channel_id ?? "";
const channelCultureObj = channelId ? await getChannelCulture(channelId) : null;
const channelCulture = channelCultureObj?.culture_summary;
for (let i = 0; i < subBatches.length; i++) {
const batch = subBatches[i];
const targetIds = batch.map((t) => t.id);
// User reputation + profiles
const userContexts = new Map<string, string>();
const userProfiles = new Map<string, string>();
for (const msg of batch) {
if (!userContexts.has(msg.user_id)) {
const rep = await initializeUserReputation(msg.user_id, msg.guild_id);
userContexts.set(msg.user_id, `<user_reputation trust_score="${rep.trust_score}" />`);
}
if (!userProfiles.has(msg.user_id)) {
const profile = await getUserProfile(msg.user_id);
userProfiles.set(msg.user_id, profile ? `<user_profile>${sanitizeAiContent(profile.profile_summary)}</user_profile>` : "");
}
}
const buildContent = async (state: RetryState): Promise<string> => {
const correction = state.lastParseError ? { error: state.lastParseError, preview: state.lastInvalidContent?.slice(0, 800) ?? "<empty>" } : undefined;
const correctedExamples = await buildCorrectedFewShotExamples();
const systemText = buildSystemPromptModular({ contextText, mode: "text", correction, correctedExamples, channelCulture });
const messagesBlock = (await Promise.all(batch.map(async (msg) => {
const content = getAnalysisContent(msg);
const msgUrls = extractUrlsFromText(content);
const urlContexts = msgUrls.map((url) => {
const ft = urlFetchMap.get(url);
return ft ? `<web_content url="${escapeXml(url)}">${escapeXml(ft)}</web_content>` : null;
}).filter(Boolean).join("\n");
const webContext = urlContexts ? `\n${urlContexts}` : "";
const userCtx = userContexts.get(msg.user_id) ?? "";
const userProfileCtx = userProfiles.get(msg.user_id) ?? "";
const refXml = await buildReferenceXml(msg);
return `<message id="${msg.id}" user="${msg.username}">\n ${userCtx}${userProfileCtx ? `\n ${userProfileCtx}` : ""}${refXml ? `\n ${refXml}` : ""}\n <content>${escapeXml(content)}</content>${webContext}\n</message>`;
}))).join("\n");
const searxngBlock = searxngResults.size > 0
? `\n\n<web_searches>\n${Array.from(searxngResults.entries()).map(([q, xml]) => ` <search_query query="${escapeXml(q)}">\n${xml} </search_query>`).join("\n")}\n</web_searches>`
: "";
return `${systemText}${searxngBlock}\n\n<messages_to_analyze>\n${messagesBlock}\n</messages_to_analyze>`;
};
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);
}
// Fan-out results for deduplicated messages
const fannedOutResults = groupMapping.size > 0
? batchResult.results.flatMap((result) => {
const members = groupMapping.get(result.messageId);
return members ? members.map((memberId) => ({ ...result, messageId: memberId })) : [result];
})
: batchResult.results;
allResults.push(...fannedOutResults);
if (batchResult.raw) lastRaw = batchResult.raw;
logModerationAnalysis(targetIds, config.AI_LLM_MODEL, batchResult.results, 0, undefined);
}
log.debug({ targetCount: targets.length, resultCount: allResults.length, subBatchCount: subBatches.length }, "Text-only batch analysis complete");
return { results: allResults, raw: lastRaw };
}
// ---------------------------------------------------------------------------
// Media batch — download + vision + single LLM call
// ---------------------------------------------------------------------------
async function runMediaBatch(
targets: MessageRecord[],
contextText: string,
attachments: AttachmentRecord[] | undefined,
): Promise<{ results: AnalysisResult[]; raw: unknown }> {
if (!targets.length) return { results: [], raw: null };
// Lazy init sticker cache
const { isStickerCacheReady, initStickerCache } = await import("./stickerCache.js");
if (!isStickerCacheReady()) {
await initStickerCache().catch((err: unknown) => log.warn({ error: err instanceof Error ? err.message : String(err) }, "Sticker cache init failed"));
}
// Phase A: Prepare ALL messages in parallel
const prepared = await Promise.all(targets.map((target) => prepareMediaMessage(target, attachments)));
// Phase B: ONE batched LLM call
const targetIds = targets.map((t) => t.id);
const channelId = targets[0].channel_id;
const channelCultureObj = channelId ? await getChannelCulture(channelId) : null;
const channelCulture = channelCultureObj?.culture_summary;
const correctedExamples = await buildCorrectedFewShotExamples();
const systemText = buildSystemPromptModular({ contextText, mode: "mixed", correctedExamples, channelCulture });
const messagesBlock = prepared.map((p) => p.messageBlock).join("\n");
const userContent = `${systemText}\n\n<messages_to_analyze>\n${messagesBlock}\n</messages_to_analyze>`;
const perMsgTimeout = config.AI_LLM_MEDIA_ANALYSIS_TIMEOUT_MS ?? 60000;
const batchTimeout = Math.min(Math.max(perMsgTimeout, perMsgTimeout * targets.length), 300_000);
const abortController = new AbortController();
const timeoutId = setTimeout(() => abortController.abort(), batchTimeout);
timeoutId.unref();
try {
const result = await callModerationLLM(
async (_state: RetryState) => userContent,
targetIds,
`media-batch:${targetIds.length}msgs`,
abortController.signal,
);
log.info({ mediaCount: targets.length, resultCount: result.results.length }, "Media batch analysis complete");
return result;
} catch (err: any) {
if (err.name === "AbortError" || abortController.signal.aborted) {
throw new Error(`Media batch analysis timed out after ${batchTimeout}ms for ${targets.length} messages`);
}
throw err;
} finally {
clearTimeout(timeoutId);
}
}
// ---------------------------------------------------------------------------
// Types
// ---------------------------------------------------------------------------
export interface ModerationInput {
targets: MessageRecord[];
contextText: string;
attachments?: AttachmentRecord[];
}
export interface ModerationOutput {
results: AnalysisResult[];
raw: unknown;
}
// ---------------------------------------------------------------------------
// Main entry point
// ---------------------------------------------------------------------------
/**
* Runs LLM-based moderation analysis on messages.
* Splits text-only vs media, runs both paths in parallel, applies caching.
*/
export async function runModerationAnalysis(
input: ModerationInput,
): Promise<ModerationOutput> {
const { targets, contextText, attachments } = input;
initSearxngCache(config.REDIS_URL);
if (!targets.length) throw new Error("No targets provided for analysis");
// Per-user moderation cache check (text-only)
const cacheHits: AnalysisResult[] = [];
const uncachedTargets: MessageRecord[] = [];
const seenCacheKeys = new Set<string>();
for (const target of targets) {
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);
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);
}
continue;
}
seenCacheKeys.add(cacheKey);
try {
const cached = await getCachedTextModeration(cacheKey);
if (cached) {
const hasMediaInMeta = target.metadata && (() => {
const ev = extractMessageMediaEvidence(target.metadata);
return ev.attachments.length > 0 || ev.stickers.length > 0 || ev.embeds.length > 0;
})();
if (hasMediaInMeta) {
log.debug({ messageId: target.id, cacheKey }, "Cache entry but message has media — treating as miss");
} else 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: [],
} as AnalysisResult);
continue;
}
}
} catch { /* proceed */ }
uncachedTargets.push(target);
}
if (cacheHits.length > 0) {
log.info({ cacheHits: cacheHits.length, uncached: uncachedTargets.length, total: targets.length }, "User moderation cache applied");
}
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");
// Run both paths in parallel
const [textBatchResult, mediaBatchResult] = await Promise.all([
textOnlyTargets.length > 0
? runTextOnlyBatch(textOnlyTargets, contextText)
: Promise.resolve({ results: [] as AnalysisResult[], raw: null }),
mediaTargets.length > 0
? runMediaBatch(mediaTargets, contextText, attachments)
: Promise.resolve({ results: [] as AnalysisResult[], raw: null }),
]);
// Store uncached text-only results in cache
for (const result of textBatchResult.results) {
const target = textOnlyTargets.find((t) => t.id === result.messageId);
if (!target) continue;
const rawContent = target.edited_content ?? target.content;
if (!rawContent.trim()) continue;
if (result.status === "error") continue;
if (target.metadata) {
const evidence = extractMessageMediaEvidence(target.metadata);
if (evidence.attachments.length > 0 || evidence.stickers.length > 0 || evidence.embeds.length > 0) 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(() => {});
}
const allResults = [...cacheHits, ...textBatchResult.results, ...mediaBatchResult.results];
const raw = textBatchResult.raw ?? mediaBatchResult.raw;
log.debug({ targetCount: targets.length, resultCount: allResults.length, cacheHits: cacheHits.length }, "Moderation analysis complete");
return { results: allResults, raw };
}
// ---------------------------------------------------------------------------
// Simple text-only fallback
// ---------------------------------------------------------------------------
/**
* Simple two-step text fallback for cheap/small models.
* Step 1: Single-word classification (clean/warn/flagged).
* Step 2: Real analysis text (only if not clean).
*/
export async function runSimpleTextFallback(
message: MessageRecord,
): Promise<AnalysisResult> {
const content = getAnalysisContent(message);
const MAX_CONTENT_CHARS = 500;
const truncatedContent = content.length > MAX_CONTENT_CHARS ? content.slice(0, MAX_CONTENT_CHARS) + "..." : content;
let userProfileCtx = "";
try {
const profile = await getUserProfile(message.user_id);
if (profile?.profile_summary) {
userProfileCtx = `\n\nProfil pengirim pesan:\n${sanitizeAiContent(profile.profile_summary, 2000, false)}\n`;
}
} catch { /* non-fatal */ }
// 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 (False Positive Prevention):
- Slang Indonesia ("anjay", "wkwk", "njir", "gws", dll) dan makian umum ("asu", "anjing", "bangsat") yang TIDAK ditujukan ke orang lain = clean.
- Konten coding/programming (kode, log error, SQL, command line, error message, stack trace, nama library) = clean. JANGAN flag hanya karena ada kata "error" atau "crash" dalam konteks teknis.
- Nama proyek, tools, framework (IMPHNEN, Bete, Cursor, Claude, React, Discord) = clean.
- Percakapan multilingual (campuran Indonesia-Inggris) = clean.
${userProfileCtx}
Pesan: "${truncatedContent}"
Jawab HANYA dengan satu kata: clean, warn, atau flagged`;
let status: "clean" | "warn" | "flagged";
try {
const completion = await llmChat({
messages: [{ role: "user", content: classifyPrompt }],
max_tokens: 10,
temperature: 0.1,
});
const raw = completion?.choices[0]?.message?.content?.trim().toLowerCase() ?? "";
if (raw.includes("flagged")) status = "flagged";
else if (raw.includes("warn")) status = "warn";
else status = "clean";
log.info({ messageId: message.id, status, raw }, "Simple fallback step 1");
} 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: Reason + category (only if not clean)
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";
const categoryOptions = status === "flagged" ? "harassment, gambling, atau sara" : "spam";
const reasonPrompt = `Pesan berikut telah diklasifikasikan sebagai "${status}".
${userProfileCtx}
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() ?? "";
if (!analysis || analysis.length < 5) {
analysis = `Pesan diklasifikasikan sebagai ${status} oleh sistem moderasi otomatis.`;
}
const categoryMatch = analysis.match(/[Kk]ategori:\s*(\w+)/i);
if (categoryMatch) {
const parsedCat = categoryMatch[1].toLowerCase();
if (["harassment", "spam", "gambling", "sara"].includes(parsedCat)) category = parsedCat;
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");
} 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");
}
}
return {
messageId: message.id,
status,
flags: status === "clean" ? [] : [category],
score: status === "flagged" ? 0.7 : status === "warn" ? 0.4 : 0,
analysis,
categories: status === "clean" ? [] : [category],
severity: status === "flagged" ? "medium" : status === "warn" ? "low" : "none",
confidence: 0.6,
recommendedAction: status === "flagged" ? "review" : status === "warn" ? "warn" : "none",
policyVersion: "default-simple-2026-06",
evidence: status !== "clean" ? [content.length > 120 ? content.slice(0, 120) + "..." : content] : [],
};
}