perf: maximize AI batches by token budget

Batch AI moderation by estimated token budget instead of fixed message count. Send as many messages as fit within an 80k token request budget while keeping one concurrent API request. Include message metadata and chronological conversation context so the model can judge provocation and replies from surrounding discussion.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

src/moderation/aiAnalyzer.ts

@@ -8,6 +8,10 @@ import type { MessageRecord } from "./types";
 const logger = createChildLogger("ai-analyzer");
 const queuedMessageIds = new Set<string>();
 let isProcessing = false;
+let activeRequests = 0;
+const MAX_CONCURRENT_REQUESTS = 1;
+const MAX_AI_REQUEST_TOKENS = 80_000;
+const AI_PROMPT_TOKEN_RESERVE = 6_000;
 
 interface ChatCompletionResponse {
   choices?: Array<{
@@ -18,7 +22,7 @@ interface ChatCompletionResponse {
 }
 
 interface LLMAnalysis {
-  status: "clean" | "flagged";
+  status: "clean" | "warn" | "flagged";
   flags: string[];
   score: number;
   analysis: string;
@@ -28,20 +32,47 @@ function getAnalysisText(message: MessageRecord): string {
   return (message.edited_content || message.content || "").trim();
 }
 
+function estimateTokens(text: string): number {
+  return Math.ceil(text.length / 4);
+}
+
+function formatMessageForAnalysis(message: MessageRecord, index: number): string {
+  const text = getAnalysisText(message);
+  const time = new Date(message.created_at).toISOString();
+  return `${index + 1}. id=${message.id} time=${time} user=${message.username}: ${text}`;
+}
+
+function estimateMessageTokens(message: MessageRecord): number {
+  return estimateTokens(formatMessageForAnalysis(message, 0)) + 16;
+}
+
 async function fetchJson(url: string, init: RequestInit): Promise<unknown> {
   const controller = new AbortController();
   const timeout = setTimeout(() => controller.abort(), config.AI_ANALYSIS_TIMEOUT_MS);
 
   try {
     const response = await fetch(url, { ...init, signal: controller.signal });
-    const body = await response.json().catch(() => ({}));
+    const text = await response.text();
+
     if (!response.ok) {
-      const message = typeof body === "object" && body && "error" in body
-        ? JSON.stringify(body)
-        : response.statusText;
+      const message = text.includes("{")
+        ? JSON.stringify(JSON.parse(text.substring(text.indexOf("{"))))
+        : text;
       throw new Error(`AI request failed (${response.status}): ${message}`);
     }
-    return body;
+
+    // Handle streaming response: extract JSON from response text
+    const jsonStart = text.indexOf("{");
+    const jsonEnd = text.lastIndexOf("}");
+    if (jsonStart >= 0 && jsonEnd > jsonStart) {
+      try {
+        return JSON.parse(text.substring(jsonStart, jsonEnd + 1));
+      } catch {
+        // Fall through to parse full text
+      }
+    }
+
+    return JSON.parse(text);
   } finally {
     clearTimeout(timeout);
   }
@@ -53,7 +84,7 @@ function parseLLMAnalysis(content: string): LLMAnalysis {
   if (jsonStart >= 0 && jsonEnd > jsonStart) {
     try {
       const parsed = JSON.parse(content.slice(jsonStart, jsonEnd + 1));
-      const status = parsed.status === "flagged" ? "flagged" : "clean";
+      const status = parsed.status === "flagged" ? "flagged" : parsed.status === "warn" ? "warn" : "clean";
       const flags = Array.isArray(parsed.flags) ? parsed.flags.map(String) : [];
       const score = Math.max(0, Math.min(1, Number(parsed.score) || 0));
       const analysis = typeof parsed.analysis === "string" ? parsed.analysis : content;
@@ -64,14 +95,14 @@ function parseLLMAnalysis(content: string): LLMAnalysis {
   }
 
   return {
-    status: /flagged|bahaya|berisiko|toxic|hate|harassment|violence|sexual|self-harm/i.test(content) ? "flagged" : "clean",
+    status: /flagged|bahaya|berisiko|toxic|hate|harassment|violence|sexual|self-harm|illegal|scam|hacking/i.test(content) ? "flagged" : /warn|provokasi|hinaan|menyerang/i.test(content) ? "warn" : "clean",
     flags: [],
     score: 0,
     analysis: content.trim() || "Tidak ada analisis dari LLM.",
   };
 }
 
-async function runLLMAnalysis(text: string): Promise<{ result: LLMAnalysis; raw: unknown }> {
+async function runLLMAnalysis(messages: MessageRecord[]): Promise<{ results: LLMAnalysis[]; raw: unknown }> {
   const response = await retryWithBackoff(
     () => fetchJson(`${config.AI_LLM_BASE_URL}/chat/completions`, {
       method: "POST",
@@ -84,31 +115,119 @@ async function runLLMAnalysis(text: string): Promise<{ result: LLMAnalysis; raw:
         messages: [
           {
             role: "system",
-            content: "Kamu analis moderation Discord. Nilai pesan untuk toxic, harassment, hate, violence, sexual, self-harm, spam, scam, atau unsafe content. Balas JSON valid saja dengan schema: {\"status\":\"clean|flagged\",\"flags\":[\"...\"],\"score\":0..1,\"analysis\":\"ringkasan singkat Bahasa Indonesia + alasan + aksi disarankan\"}.",
+            content: `Kamu moderator Discord komunitas. Analisis setiap pesan dengan 3 kategori:
+- CLEAN: Pesan normal, tidak melanggar aturan
+- WARN: Melanggar aturan minor yang menarget orang lain (tone menyerang, hinaan ringan, konflik kecil) - butuh peringatan tapi tidak dihapus
+- FLAGGED: Melanggar aturan berat (NSFW, ilegal, hacking, scam, harassment, violence, SARA, gore, spam, promosi judi) - butuh review moderator untuk penghapusan
+
+ATURAN KOMUNITAS LENGKAP:
+
+1. JAGA SIKAP DAN HORMATI SESAMA
+   - Gunakan bahasa yang sopan dan menghormati semua anggota
+   - Tanpa memandang latar belakang, usia, gender, atau pandangan
+   - Dilarang keras: pelecehan, rasisme, seksisme, diskriminasi
+
+2. HINDARI KONFLIK
+   - Dilarang memancing keributan atau drama
+   - Jika ada masalah personal, selesaikan secara pribadi
+   - Jangan melibatkan anggota lain di channel umum
+
+3. KONTEN EKSPLISIT DILARANG
+   - Dilarang keras: NSFW, ilegal, pornografi, kekerasan (gore), SARA
+   - Tidak ada tempat untuk penyimpangan atau LGBT
+   - Tidak ada promosi aktivitas atau ideologi LGBT
+
+4. JAGA PRIVASI
+   - Dilarang menyebarkan informasi pribadi milik anggota lain tanpa izin
+
+5. PROFIL YANG SOPAN
+   - Username, foto profil, dan server tag harus pantas
+   - Jangan gunakan unsur ofensif atau vulgar
+
+6. DILARANG SPAM DAN PENIPUAN
+   - Dilarang: hoaks, link berbahaya (phishing/scam), spam
+   - Dilarang: promosi, judi, link referral
+
+7. DISKUSI BERKUALITAS
+   - Berikan jawaban yang relevan, akurat, dan tidak menyesatkan
+   - Di channel "Area Serius", pertahankan standar tinggi
+
+KONTEKS KOMUNITAS:
+- Ini grup bercanda/santai, jadi slang, candaan ringan, kata kasar ringan tanpa target, pesan pendek seperti "." atau "P", dan pertanyaan tidak jelas tetap CLEAN
+- Jangan beri WARN hanya karena pesan singkat, informal, ambigu, low-quality, atau kurang konteks
+- Pahami alur pembahasan antar pesan: pesan yang sendiri terlihat normal bisa WARN/FLAGGED jika dalam konteks percakapan sedang memancing konflik, menormalisasi pelanggaran, atau melanjutkan provokasi
+- Jangan menghukum orang yang sedang menasehati, menjelaskan bahaya, mengutip, atau menolak tindakan buruk; nilai maksud dan konteksnya
+- WARN hanya jika ada orang/kelompok yang diserang, dihina, diprovokasi, atau konflik mulai dipancing
+
+PENENTUAN STATUS:
+- WARN jika: hinaan ringan yang menarget orang/kelompok, provokasi konflik kecil, username/profil kurang pantas
+- FLAGGED jika: profanity berat, harassment, threats, violence, illegal activity, hacking, scam, NSFW, SARA, gore, spam, judi, LGBT content
+
+Balas JSON array dengan schema: [{"status":"clean|warn|flagged","flags":["..."],"score":0..1,"analysis":"ringkasan Bahasa Indonesia + alasan + aksi disarankan"}]
+Satu JSON object per pesan dalam array.`,
           },
           {
             role: "user",
-            content: text,
+            content: `Analisis ${messages.length} pesan berikut sebagai satu alur percakapan. Tetap kembalikan satu hasil per pesan dengan urutan yang sama:\n${messages.map(formatMessageForAnalysis).join("\n")}`,
           },
         ],
         temperature: 0.2,
       }),
+      signal: AbortSignal.timeout(config.AI_ANALYSIS_TIMEOUT_MS),
     }),
     { retries: 2, logger },
   ) as ChatCompletionResponse;
 
   const content = response.choices?.[0]?.message?.content?.trim() || "";
-  return { result: parseLLMAnalysis(content), raw: response };
+
+  // Extract JSON array from response
+  const jsonStart = content.indexOf("[");
+  const jsonEnd = content.lastIndexOf("]");
+  let results: LLMAnalysis[] = [];
+
+  if (jsonStart >= 0 && jsonEnd > jsonStart) {
+    try {
+      const parsed = JSON.parse(content.substring(jsonStart, jsonEnd + 1));
+      if (Array.isArray(parsed)) {
+        results = parsed.map((item: any) => {
+          const status = item.status === "flagged" ? "flagged" : item.status === "warn" ? "warn" : "clean";
+          return {
+            status,
+            flags: Array.isArray(item.flags) ? item.flags.map(String) : [],
+            score: Math.max(0, Math.min(1, Number(item.score) || 0)),
+            analysis: typeof item.analysis === "string" ? item.analysis : content,
+          };
+        });
+      }
+    } catch {
+      // Fall through to individual parsing
+    }
+  }
+
+  // If batch parsing failed, parse as individual responses
+  if (results.length === 0) {
+    results = messages.map(() => parseLLMAnalysis(content));
+  }
+
+  return { results, raw: response };
 }
 
-async function analyzeAndStore(db: SqliteDatabase, message: MessageRecord): Promise<void> {
-  const text = getAnalysisText(message);
-  if (!config.AI_ANALYSIS_ENABLED || text.length === 0) return;
+async function analyzeAndStoreBatch(db: SqliteDatabase, messages: MessageRecord[]): Promise<void> {
+  if (messages.length === 0) return;
 
+  const analyzableMessages = messages.filter((message) => getAnalysisText(message).length > 0);
+  if (analyzableMessages.length === 0) return;
+
+  activeRequests++;
   try {
-    const { result, raw } = await runLLMAnalysis(text);
+    const { results, raw } = await runLLMAnalysis(analyzableMessages);
+
+    for (let i = 0; i < analyzableMessages.length; i++) {
+      const message = analyzableMessages[i];
+      const result = results[i] || parseLLMAnalysis("");
+
       const row = updateMessageAIAnalysis(db, message.id, {
-      status: result.status,
+        status: result.status as "pending" | "clean" | "warn" | "flagged" | "error",
         flags: JSON.stringify(result.flags),
         score: result.score,
         raw: JSON.stringify(raw),
@@ -117,7 +236,10 @@ async function analyzeAndStore(db: SqliteDatabase, message: MessageRecord): Prom
         error: null,
       });
       if (row) (globalThis as any).broadcastMessageAnalyzed?.(row);
+    }
   } catch (error) {
+    const errorMsg = error instanceof Error ? error.message : String(error);
+    for (const message of analyzableMessages) {
       const row = updateMessageAIAnalysis(db, message.id, {
         status: "error",
         flags: null,
@@ -125,10 +247,13 @@ async function analyzeAndStore(db: SqliteDatabase, message: MessageRecord): Prom
         raw: null,
         analysis: null,
         analyzedAt: Date.now(),
-      error: error instanceof Error ? error.message : String(error),
+        error: errorMsg,
       });
       if (row) (globalThis as any).broadcastMessageAnalyzed?.(row);
-    logger.warn({ messageId: message.id, error }, "AI analysis failed");
+    }
+    logger.warn({ count: messages.length, error }, "AI batch analysis failed");
+  } finally {
+    activeRequests--;
   }
 }
 
@@ -136,12 +261,34 @@ async function drainQueue(db: SqliteDatabase): Promise<void> {
   if (isProcessing) return;
   isProcessing = true;
   try {
+    const batchTokenLimit = MAX_AI_REQUEST_TOKENS - AI_PROMPT_TOKEN_RESERVE;
+
     while (queuedMessageIds.size > 0) {
-      const messageId = queuedMessageIds.values().next().value as string | undefined;
-      if (!messageId) break;
-      queuedMessageIds.delete(messageId);
+      while (activeRequests >= MAX_CONCURRENT_REQUESTS) {
+        await new Promise((resolve) => setTimeout(resolve, 100));
+      }
+
+      const batch: MessageRecord[] = [];
+      let tokenEstimate = 0;
+      for (const messageId of Array.from(queuedMessageIds)) {
         const message = getMessageById(db, messageId);
-      if (message) await analyzeAndStore(db, message);
+        queuedMessageIds.delete(messageId);
+        if (!message) continue;
+
+        const messageTokens = estimateMessageTokens(message);
+        if (batch.length > 0 && tokenEstimate + messageTokens > batchTokenLimit) {
+          queuedMessageIds.add(messageId);
+          break;
+        }
+
+        batch.push(message);
+        tokenEstimate += messageTokens;
+      }
+
+      if (batch.length > 0) {
+        logger.info({ count: batch.length, tokenEstimate }, "Processing AI analysis batch");
+        await analyzeAndStoreBatch(db, batch);
+      }
     }
   } finally {
     isProcessing = false;
@@ -166,7 +313,7 @@ export function startPendingAIAnalysisWorker(db: SqliteDatabase): void {
   logger.info("AI analysis worker started");
   setInterval(() => {
     if (isProcessing) return;
-    const pendingMessages = getPendingAIAnalysisMessages(db, 3);
+    const pendingMessages = getPendingAIAnalysisMessages(db, 500);
     if (pendingMessages.length === 0) return;
     logger.info({ count: pendingMessages.length }, "Queueing pending AI analysis messages");
     for (const message of pendingMessages) {

src/moderation/backlogSync.ts

@@ -6,65 +6,6 @@ import { captureMessage } from "./messageCapture";
 
 const logger = createChildLogger("backlog-sync");
 
-function isWatchableChannel(channel: { type?: string; messages?: unknown }): boolean {
-  return Boolean(
-    channel.messages &&
-      ["GUILD_TEXT", "GUILD_PUBLIC_THREAD", "GUILD_PRIVATE_THREAD"].includes(
-        channel.type ?? "",
-      ),
-  );
-}
-
-async function collectWatchableChannels(guild: any): Promise<any[]> {
-  const channels: any[] = [];
-
-  // Fast pass: collect text channels from cache only
-  for (const channel of guild.channels.cache.values()) {
-    if (isWatchableChannel(channel)) {
-      channels.push(channel);
-    }
-  }
-
-  // Slow pass: discover threads with timeout per channel (non-blocking to message sync)
-  const threadPromises: Promise<void>[] = [];
-  for (const channel of guild.channels.cache.values()) {
-    if (!channel.threads?.fetch) continue;
-
-    threadPromises.push(
-      (async () => {
-        for (const archived of [false, true]) {
-          try {
-            const controller = new AbortController();
-            const timeout = setTimeout(() => controller.abort(), 5000);
-            const fetched = await Promise.race([
-              channel.threads.fetch({ archived, limit: 100 }),
-              new Promise((_, reject) => controller.signal.addEventListener('abort', () => reject(new Error('timeout')))),
-            ]).catch(() => null);
-            clearTimeout(timeout);
-
-            if (!fetched?.threads) continue;
-            for (const thread of fetched.threads.values()) {
-              if (isWatchableChannel(thread)) channels.push(thread);
-            }
-          } catch {
-            // Skip this channel's threads on timeout/error
-          }
-        }
-      })()
-    );
-  }
-
-  // Wait for all thread discoveries with overall timeout
-  await Promise.race([
-    Promise.all(threadPromises),
-    new Promise((resolve) => setTimeout(resolve, 30000)),
-  ]).catch(() => {
-    logger.warn("Thread discovery timeout, proceeding with cached channels");
-  });
-
-  return Array.from(new Map(channels.map((channel) => [channel.id, channel])).values());
-}
-
 async function syncChannelMessages(
   db: SqliteDatabase,
   channel: any,
@@ -116,45 +57,45 @@ export async function syncBacklogMessages(
     return;
   }
 
+  logger.info({ guildId: guild.id }, "Backlog sync ready (will sync on-demand per selected channel)");
+}
+
+export async function syncSelectedChannelBacklog(
+  client: Client,
+  db: SqliteDatabase,
+  guildId: string,
+  channelId: string,
+): Promise<number> {
+  const guild = client.guilds.cache.get(guildId);
+  if (!guild) {
+    logger.warn({ guildId }, "Guild not found for backlog sync");
+    return 0;
+  }
+
+  const channel = guild.channels.cache.get(channelId);
+  if (!channel) {
+    logger.warn({ guildId, channelId }, "Channel not found for backlog sync");
+    return 0;
+  }
+
   const cutoffTime = Date.now() - config.BACKLOG_SYNC_HOURS * 60 * 60 * 1000;
   logger.info(
-    { guildId: guild.id, hours: config.BACKLOG_SYNC_HOURS },
-    "Starting message backlog sync",
+    { guildId, channelId, hours: config.BACKLOG_SYNC_HOURS },
+    "Starting backlog sync for selected channel",
   );
 
-  logger.info({ guildId: guild.id }, "Fetching guild channels for backlog sync");
-  await guild.channels.fetch().catch((error) => {
-    logger.warn(
-      { guildId: guild.id, error: error instanceof Error ? error.message : String(error) },
-      "Failed to fetch guild channels before backlog sync",
-    );
-    return null;
-  });
-
-  logger.info({ guildId: guild.id }, "Collecting watchable channels for backlog sync");
-  const channels = await collectWatchableChannels(guild);
-
-  let total = 0;
-  logger.info(
-    { guildId: guild.id, channels: channels.length, hours: config.BACKLOG_SYNC_HOURS },
-    "Watchable channels collected for backlog sync",
-  );
-
-  for (const channel of channels) {
   try {
     const count = await syncChannelMessages(db, channel as any, cutoffTime);
-      total += count;
-      logger.info({ channelId: channel.id, count }, "Backlog channel sync completed");
+    logger.info({ channelId, count }, "Backlog sync completed for selected channel");
+    return count;
   } catch (error) {
     logger.warn(
       {
-          channelId: channel.id,
+        channelId,
         error: error instanceof Error ? error.message : String(error),
       },
-        "Backlog channel sync failed",
+      "Backlog sync failed for selected channel",
     );
+    return 0;
   }
-  }
-
-  logger.info({ total }, "Message backlog sync completed");
 }

src/moderation/messageCapture.ts

@@ -1,9 +1,8 @@
 import type { Client, Message } from "discord.js-selfbot-v13";
 import { createChildLogger } from "../logger";
-import type { SqliteDatabase } from "../muxer-queue";
 import { config } from "../config";
+import type { SqliteDatabase } from "../muxer-queue";
 import { insertMessage, insertAttachment } from "./messageStore";
-import { processAttachmentUpload } from "./attachmentUploader";
 import { getDisplayContent, getMessageLocation, getMessageMetadata } from "./messageMetadata";
 import { queueMessageAnalysis } from "./aiAnalyzer";
 import type { MessageRecord, AttachmentRecord } from "./types";
@@ -59,17 +58,15 @@ export async function captureMessage(
         size: attachment.size,
         type: attachment.contentType || "application/octet-stream",
         discord_url: attachment.url,
-        uploaded_url: null,
-        upload_status: "pending",
+        uploaded_url: attachment.url,
+        upload_status: "uploaded",
         upload_error: null,
         created_at: Date.now(),
-        uploaded_at: null,
+        uploaded_at: Date.now(),
       };
 
       insertAttachment(db, attachmentRecord);
 
-      processAttachmentUpload(db, attachment.id, attachment.url, attachment.name || "unknown")
-        .then(() => {
       if (broadcaster.broadcastAttachmentUploaded) {
         broadcaster.broadcastAttachmentUploaded({
           id: attachment.id,
@@ -79,13 +76,6 @@ export async function captureMessage(
           created_at: Date.now(),
         });
       }
-        })
-        .catch((error) => {
-          logger.error(
-            { attachmentId: attachment.id, error: error instanceof Error ? error.message : String(error) },
-            "Background attachment upload failed",
-          );
-        });
     }
   }
 

src/moderation/messageStore.ts

@@ -222,7 +222,7 @@ export function updateAttachmentAsFailedUpload(
 }
 
 interface AIAnalysisUpdate {
-  status: "pending" | "clean" | "flagged" | "error";
+  status: "pending" | "clean" | "warn" | "flagged" | "error";
   flags?: string | null;
   score?: number | null;
   raw?: string | null;

src/moderation/types.ts

@@ -13,7 +13,7 @@ export interface MessageRecord {
   deleted_at: number | null;
   type: "text" | "edited" | "deleted";
   metadata: string | null;
-  ai_status?: "pending" | "clean" | "flagged" | "error" | null;
+  ai_status?: "pending" | "clean" | "warn" | "flagged" | "error" | null;
   ai_moderation_flags?: string | null;
   ai_moderation_score?: number | null;
   ai_moderation_raw?: string | null;

src/webserver.ts

@@ -12,6 +12,7 @@ import { discordPlayer } from "./player";
 import type { VoiceController } from "./voiceController";
 import { getDatabase, getPersistedValue, setPersistedValue } from "./muxer-queue";
 import { getMessagesByChannel, getAttachmentsByChannel } from "./moderation/messageStore";
+import { syncSelectedChannelBacklog } from "./moderation/backlogSync";
 
 const wsLogger = createChildLogger("webserver");
 
@@ -277,6 +278,32 @@ export function startWebserver(
     }
   });
 
+  app.post("/api/backlog-sync", async (req, res, next) => {
+    try {
+      const { guildId, channelId } = req.body as {
+        guildId?: string;
+        channelId?: string;
+      };
+
+      if (!guildId || !channelId) {
+        throw new AppError(
+          "guildId and channelId are required",
+          "MISSING_BACKLOG_PARAMS",
+          400,
+        );
+      }
+
+      const count = await syncSelectedChannelBacklog(_client, getDatabase(), guildId, channelId);
+      res.json({
+        success: true,
+        channelId,
+        messagesSync: count,
+      });
+    } catch (error) {
+      next(error);
+    }
+  });
+
   // Inbound: Discord PCM → tagged chunks → browser
   (global as any).broadcastPcmToWeb = (chunk: Buffer, userId: string) => {
     let hash = 0;