refactor(ai-moderation): offload individual message analysis to worker pool and add auto-delete notifications

services/discord-gateway/src/modules/ai-moderation/aiAnalysisWorker.ts

@@ -1,13 +1,13 @@
 import { config } from "../../shared/config/config.js";
 import { initializeDatabase } from "../../shared/database/drizzle.js";
 import { buildConversationContext } from "./conversationContext.js";
-import { runModerationAnalysis } from "./llmModerationClient.js";
+import { runModerationAnalysis, runSimpleTextFallback } from "./llmModerationClient.js";
 import {
   getAttachmentsForMessages,
   getConversationContextBefore,
   updateMessagesAIAnalysisBulk,
 } from "../message-capture/messageStore.js";
-import type { MessageRecord } from "../message-capture/types.js";
+import type { MessageRecord, AnalysisResult } from "../message-capture/types.js";
 
 let dbInitialized = false;
 let dbInitPromise: Promise<any> | null = null;
@@ -22,6 +22,10 @@ async function ensureDb() {
   await dbInitPromise;
 }
 
+// ---------------------------------------------------------------------------
+// Batch analysis (existing)
+// ---------------------------------------------------------------------------
+
 export interface AnalysisWorkerRequest {
   conversationKey: string;
   messages: MessageRecord[];
@@ -117,8 +121,6 @@ export default async function processAnalysisRequest({
       const rows = await updateMessagesAIAnalysisBulk(updates);
       return { ok: true, conversationKey, rows };
     } catch (dbErr) {
-      // If bulk update fails, we log it but don't fail the worker completely
-      // so it can at least retry later without blowing up the circuit breaker if it was an isolated issue
       throw new Error(
         `Failed to update DB: ${dbErr instanceof Error ? dbErr.message : String(dbErr)}`,
       );
@@ -143,3 +145,82 @@ export default async function processAnalysisRequest({
     return { ok: false, conversationKey, rows, error: errorMessage };
   }
 }
+
+// ---------------------------------------------------------------------------
+// Individual fallback analysis (offloaded from main thread)
+// ---------------------------------------------------------------------------
+
+export interface IndividualWorkerRequest {
+  message: MessageRecord;
+  /** Optional — if true, skip normal analysis and go straight to simple fallback */
+  skipNormalAnalysis: boolean;
+}
+
+export type IndividualWorkerResponse =
+  | {
+      ok: true;
+      results: AnalysisResult[];
+    }
+  | {
+      ok: false;
+      results: AnalysisResult[];
+      error: string;
+    };
+
+/**
+ * Processes a single message analysis in the worker thread.
+ * Fetches context, attachments, runs LLM analysis (or simple fallback),
+ * and returns the result — does NOT update DB or broadcast.
+ *
+ * The caller (main thread) handles DB writes, broadcasting, and auto-delete
+ * scheduling.
+ */
+export async function processIndividualAnalysis({
+  message,
+  skipNormalAnalysis,
+}: IndividualWorkerRequest): Promise<IndividualWorkerResponse> {
+  if (!config.AI_LLM_API_KEY) {
+    return { ok: false, results: [], error: "AI_LLM_API_KEY is missing" };
+  }
+
+  try {
+    await ensureDb();
+
+    const contextBefore = await getConversationContextBefore({
+      channelId: message.channel_id,
+      threadId: message.thread_id,
+      beforeCreatedAt: message.created_at,
+      limit: config.AI_ANALYSIS_CONTEXT_MESSAGE_LIMIT,
+    });
+
+    const contextLines = buildConversationContext({
+      contextBefore,
+      targets: [message],
+      maxTokens: config.AI_ANALYSIS_MAX_CONTEXT_TOKENS,
+    });
+
+    const contextIds = contextBefore.map((m) => m.id);
+    const attachments = await getAttachmentsForMessages([message.id, ...contextIds]);
+
+    let results: AnalysisResult[];
+
+    if (skipNormalAnalysis) {
+      // Go straight to simple text fallback (no JSON, no complex prompt)
+      const simpleResult = await runSimpleTextFallback(message);
+      results = [simpleResult];
+    } else {
+      // Try normal analysis first
+      const moderationResult = await runModerationAnalysis({
+        targets: [message],
+        contextText: contextLines.join("\n"),
+        attachments,
+      });
+      results = moderationResult.results;
+    }
+
+    return { ok: true, results };
+  } catch (error) {
+    const errorMessage = error instanceof Error ? error.message : String(error);
+    return { ok: false, results: [], error: errorMessage };
+  }
+}

services/discord-gateway/src/modules/ai-moderation/aiAnalyzer.ts

@@ -2,17 +2,13 @@ import { existsSync } from "node:fs";
 import { availableParallelism } from "node:os";
 import { fileURLToPath } from "node:url";
 import { createChildLogger } from "@bete/shared/logger";
-import { retryWithBackoff } from "@bete/shared/utils";
 import type { Client } from "discord.js-selfbot-v13";
-import { AbortError } from "p-retry";
 import { Piscina } from "piscina";
 import { config } from "../../shared/config/config.js";
 import type { EventBroadcaster } from "../event-broadcaster/index.js";
 import { invalidateAnalyticsCache } from "../message-capture/analyticsStore.js";
 import { isAgeRestrictedMetadata } from "../message-capture/messageMetadata.js";
 import {
-  getAttachmentsForMessages,
-  getConversationContextBefore,
   getConversationKeysWithIncompleteAnalysis,
   getIncompleteMessagesByConversation,
   getMessageById,
@@ -28,14 +24,7 @@ import type {
   ModerationBroadcaster,
 } from "../message-capture/types.js";
 import { attemptAutoDeleteFlaggedMessage } from "./autoDeleteManager.js";
-import {
+import { estimateTokens } from "./conversationContext.js";
-  buildConversationContext,
-  estimateTokens,
-} from "./conversationContext.js";
-import {
-  runModerationAnalysis,
-  runSimpleTextFallback,
-} from "./llmModerationClient.js";
 import { logModerationError } from "./responseLogger.js";
 
 const logger = createChildLogger("ai-analyzer");
@@ -314,11 +303,20 @@ function isConversationProcessingLocked(conversationKey: string): boolean {
 // ---------------------------------------------------------------------------
 
 /**
- * Processes a single message directly in the main process (no IPC/worker
+ * Processes a single message via the Piscina worker pool (offloaded from
- * pool overhead).  Never called from the batch path.
+ * main thread to avoid blocking the event loop).
  *
- * FIX #1+#5: Increments the individual circuit breaker on failure so a
+ * The worker handles:
- * sustained outage stops hammering the LLM endpoint.
+ * 1. DB initialization
+ * 2. Context fetching + conversation building
+ * 3. Attachment fetching
+ * 4. LLM analysis (normal or simple fallback)
+ *
+ * The main thread handles:
+ * - DB writes (updateMessagesAIAnalysisBulk)
+ * - WebSocket/Redis broadcast
+ * - Analytics cache invalidation
+ * - Auto-delete scheduling
  *
  * Infinite-loop prevention: if the LLM consistently drops the single target
  * message across all retries (analysis_incomplete), we write a terminal flag
@@ -336,117 +334,77 @@ async function processIndividualFallback(
   const conversationKey = getConversationKey(message);
 
   activeIndividualRequests++;
-  // Increment per-conversation counter so the recovery worker can see it.
   individualInFlightByConversation.set(
     conversationKey,
     (individualInFlightByConversation.get(conversationKey) ?? 0) + 1,
   );
   individualInFlightLastTouched.set(conversationKey, Date.now());
 
-  // Track whether all retries were exhausted specifically because the LLM
-  // consistently returned no result for this message (vs. a transient error).
   let exhaustedOnIncomplete = false;
-  let usedSimpleFallback = false;
 
   try {
-    const contextBefore = await getConversationContextBefore({
+    // ── Run the LLM-heavy work in the worker thread ──
-      channelId: message.channel_id,
+    // Try normal analysis first. The worker handles retries internally.
-      threadId: message.thread_id,
+    const workerResult = await workerPool.run({
-      beforeCreatedAt: message.created_at,
+      type: "individual",
-      limit: config.AI_ANALYSIS_CONTEXT_MESSAGE_LIMIT,
+      message,
-    });
+      skipNormalAnalysis: false,
+    } as any) as
+      | { ok: true; results: AnalysisResult[] }
+      | { ok: false; results: AnalysisResult[]; error: string };
 
-    const contextLines = buildConversationContext({
-      contextBefore,
-      targets: [message],
-      maxTokens: config.AI_ANALYSIS_MAX_CONTEXT_TOKENS,
-    });
-
-    const contextIds = contextBefore.map((m) => m.id);
-    const attachments = await getAttachmentsForMessages([
-      messageId,
-      ...contextIds,
-    ]);
-
-    // ── Step 1: Try the normal analysis path (retries on failure) ──
     let analysisResult: { results: AnalysisResult[] } | null = null;
+    let usedSimpleFallback = false;
 
-    try {
+    if (workerResult.ok) {
-      analysisResult = await retryWithBackoff(
+      const stillIncomplete = workerResult.results.some((r) =>
-        async () => {
+        r.flags.includes("analysis_incomplete"),
-          try {
-            const result = await runModerationAnalysis({
-              targets: [message],
-              contextText: contextLines.join("\n"),
-              attachments,
-            });
-
-            // If the LLM still dropped our only target, convert to a retryable
-            // throw so backoff kicks in.  Track this so the catch block can
-            // distinguish it from a transient network/parse failure.
-            const stillIncomplete = result.results.some((r) =>
-              r.flags.includes("analysis_incomplete"),
-            );
-            if (stillIncomplete) {
-              exhaustedOnIncomplete = true;
-              throw new Error(
-                `LLM returned no result for single-target message ${messageId} — will retry with backoff`,
-              );
-            }
-
-            // Got a real result — clear the incomplete flag.
-            exhaustedOnIncomplete = false;
-
-            return result;
-          } catch (err: any) {
-            // Propagate AbortError so outer retry is immediately cancelled on 429.
-            if (err instanceof AbortError) {
-              throw err;
-            }
-            if (
-              err?.status === 429 ||
-              err?.status === 401 ||
-              err?.status === 403
-            ) {
-              throw new AbortError(err);
-            }
-            throw err;
-          }
-        },
-        {
-          retries: 0,
-          minTimeout: 0,
-          maxTimeout: 0,
-        },
       );
-    } catch {
+      if (stillIncomplete) {
-      // Normal path failed — don't give up yet. Try the simple fallback.
+        exhaustedOnIncomplete = true;
-      analysisResult = null;
+        analysisResult = null;
+      } else {
+        analysisResult = workerResult;
+      }
     }
 
-    // ── Step 2: If normal analysis failed, try SIMPLE fallback ──
+    // ── Step 2: If normal analysis failed, try SIMPLE fallback via worker ──
-    // No JSON, no complex prompt — just asks the LLM for one word.
     if (!analysisResult) {
       logger.info(
         { messageId },
-        "Normal analysis failed for individual message — trying simple text fallback",
+        "Normal analysis failed (or incomplete) — trying simple text fallback via worker",
       );
-      usedSimpleFallback = true;
 
-      const simpleResult = await runSimpleTextFallback(message);
+      const simpleResult = await workerPool.run({
-      analysisResult = { results: [simpleResult] };
+        type: "individual_simple",
-      // Clear the exhausted flag since we got a result from the simple path
+        message,
-      exhaustedOnIncomplete = false;
+        skipNormalAnalysis: true,
+      } as any) as
+        | { ok: true; results: AnalysisResult[] }
+        | { ok: false; results: AnalysisResult[]; error: string };
+
+      if (simpleResult.ok) {
+        analysisResult = simpleResult;
+        usedSimpleFallback = true;
+        exhaustedOnIncomplete = false;
+      }
+    }
+
+    // If both failed, throw to go to the catch block
+    if (!analysisResult) {
+      throw new Error(
+        `Both normal and simple analysis failed for message ${messageId}`,
+      );
     }
 
-    // At this point we definitely have a result (either normal or simple)
     if (usedSimpleFallback) {
       logger.info(
         { messageId, status: analysisResult.results[0]?.status },
-        "Used simple text fallback for individual message — no JSON, one-word classification",
+        "Used simple text fallback for individual message (via worker)",
       );
     }
 
+    // ── Main thread: DB writes + broadcast (non-blocking work) ──
     const updates = analysisResult.results.map((r) => ({
       messageId: r.messageId,
       result: {
@@ -470,12 +428,11 @@ async function processIndividualFallback(
       scheduleAutoDelete(row);
     }
 
-    // Log individual analysis completion with comprehensive details
     const resultSummary = analysisResult.results[0];
     logModerationError(
       [messageId],
       config.AI_LLM_MODEL,
-      new Error("Success"), // For logging purposes only
+      new Error("Success"),
       {
         phase: "individual_fallback",
         status: resultSummary?.status,
@@ -485,15 +442,13 @@ async function processIndividualFallback(
       },
     );
 
-    // Reset individual CB on success.
     individualConsecutiveErrors = 0;
 
     logger.debug(
       { messageId, status: analysisResult.results[0]?.status },
-      "Individual fallback analysis complete",
+      "Individual fallback analysis complete (via worker)",
     );
   } catch (error) {
-    // FIX #5: individual failures now feed their own circuit breaker.
     individualConsecutiveErrors++;
     if (
       individualConsecutiveErrors >= config.AI_ANALYSIS_INDIVIDUAL_CB_THRESHOLD
@@ -510,7 +465,6 @@ async function processIndividualFallback(
 
     lastError = error instanceof Error ? error.message : String(error);
 
-    // Log error with responseLogger
     logModerationError(
       [messageId],
       config.AI_LLM_MODEL,
@@ -522,11 +476,6 @@ async function processIndividualFallback(
       },
     );
 
-    // Infinite-loop prevention: if all retries were exhausted because the LLM
-    // consistently dropped this specific message (not a transient error),
-    // overwrite the DB entry with a terminal flag that the recovery query
-    // does NOT match.  This permanently removes it from the recovery loop
-    // while keeping it visible as an error in the dashboard.
     if (exhaustedOnIncomplete) {
       await updateMessagesAIAnalysisBulk([
         {
@@ -548,31 +497,27 @@ async function processIndividualFallback(
       ]).catch((dbErr: unknown) => {
         logger.error(
           { messageId, error: String(dbErr) },
-          "Failed to write terminal exhausted status — message may re-enter recovery loop",
+          "Failed to write terminal exhausted status",
         );
       });
       logger.warn(
         { messageId },
-        "Individual fallback exhausted — marked as individual_analysis_exhausted to stop recovery loop",
+        "Individual fallback exhausted — marked as individual_analysis_exhausted",
       );
     } else {
-      // Transient failure (network/parse/DB): do NOT write terminal status.
-      // Message stays as error/analysis_incomplete in DB and will be retried
-      // by the recovery worker, subject to the individual circuit breaker.
       logger.error(
         {
           messageId,
           error: lastError,
           stack: error instanceof Error ? error.stack : undefined,
         },
-        "Individual fallback analysis failed (transient) — will be retried by recovery worker",
+        "Individual fallback analysis failed (transient) — will be retried",
       );
     }
   } finally {
     activeIndividualRequests--;
     individualInFlight.delete(messageId);
 
-    // Decrement per-conversation counter; remove key when it hits zero.
     const prev = individualInFlightByConversation.get(conversationKey) ?? 1;
     if (prev <= 1) {
       individualInFlightByConversation.delete(conversationKey);

services/discord-gateway/src/modules/ai-moderation/autoDeleteManager.ts

@@ -318,6 +318,52 @@ export async function attemptAutoDeleteFlaggedMessage(
     const discordMessage = await channel.messages.fetch(message.id);
     await discordMessage.delete();
 
+    // ── Notify user via DM ──
+    if (config.AUTO_DELETE_NOTIFY_USER) {
+      try {
+        const targetUser = await client.users.fetch(message.user_id);
+        if (targetUser) {
+          const reason = message.ai_categories ?? message.ai_moderation_flags ?? "(unknown)";
+          await targetUser.send(
+            `Pesan Anda di **${guild.name}** telah dihapus oleh sistem moderasi otomatis.\n` +
+            `Alasan: ${reason}\n` +
+            `Jika Anda merasa ini adalah kesalahan, silakan hubungi admin server.`,
+          );
+        }
+      } catch (dmErr) {
+        // DM might fail if user has DMs disabled — not critical
+        logger.debug(
+          { messageId: message.id, userId: message.user_id, error: String(dmErr) },
+          "Failed to send DM notification for auto-deleted message",
+        );
+      }
+    }
+
+    // ── Log to moderation channel ──
+    if (config.AUTO_DELETE_LOG_CHANNEL_ID) {
+      try {
+        const logChannel = guild.channels.cache.get(config.AUTO_DELETE_LOG_CHANNEL_ID);
+        if (logChannel && "send" in logChannel && typeof (logChannel as any).send === "function") {
+          const severity = message.ai_severity ?? "none";
+          const categories = message.ai_categories ?? message.ai_moderation_flags ?? "—";
+          const snippet = (message.edited_content ?? message.content).substring(0, 200);
+          await (logChannel as any).send(
+            `**🧹 Auto-Delete** — Pesan dari <@${message.user_id}> di <#${channelId}>\n` +
+            `**Status:** ${message.ai_status}\n` +
+            `**Severitas:** ${severity}\n` +
+            `**Kategori:** ${categories}\n` +
+            `**Isi:** ${snippet}\n` +
+            `**Waktu:** <t:${Math.floor(Date.now() / 1000)}:R>`,
+          );
+        }
+      } catch (logErr) {
+        logger.warn(
+          { messageId: message.id, error: String(logErr) },
+          "Failed to log auto-delete to moderation channel",
+        );
+      }
+    }
+
     const result = {
       deleted: true,
       skipped: false,