feat: add AI analysis integration with moderation and LLM processing

2026-05-14 02:31:16 +07:00
parent b36d038eba
commit be6c9f8132
9 changed files with 322 additions and 7 deletions
@@ -0,0 +1,169 @@
+import { config } from "../config";
+import { createChildLogger } from "../logger";
+import type { SqliteDatabase } from "../muxer-queue";
+import { retryWithBackoff } from "../retry";
+import { getMessageById, updateMessageAIAnalysis } from "./messageStore";
+import type { MessageRecord } from "./types";
+
+const logger = createChildLogger("ai-analyzer");
+const queuedMessageIds = new Set<string>();
+let isProcessing = false;
+
+interface ModerationResult {
+  flagged: boolean;
+  flags: string[];
+  score: number;
+  raw: unknown;
+}
+
+interface ChatCompletionResponse {
+  choices?: Array<{
+    message?: {
+      content?: string;
+    };
+  }>;
+}
+
+function getAnalysisText(message: MessageRecord): string {
+  return (message.edited_content || message.content || "").trim();
+}
+
+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(() => ({}));
+    if (!response.ok) {
+      const message = typeof body === "object" && body && "error" in body
+        ? JSON.stringify(body)
+        : response.statusText;
+      throw new Error(`AI request failed (${response.status}): ${message}`);
+    }
+    return body;
+  } finally {
+    clearTimeout(timeout);
+  }
+}
+
+async function runModeration(text: string): Promise<ModerationResult> {
+  const response = await retryWithBackoff(
+    () => fetchJson(`${config.OPENAI_MODERATION_BASE_URL}/moderations`, {
+      method: "POST",
+      headers: {
+        "Authorization": `Bearer ${config.OPENAI_MODERATION_API_KEY}`,
+        "Content-Type": "application/json",
+      },
+      body: JSON.stringify({
+        model: config.OPENAI_MODERATION_MODEL,
+        input: text,
+      }),
+    }),
+    { retries: 2, logger },
+  ) as any;
+
+  const result = response.results?.[0] || {};
+  const categories = result.categories || {};
+  const categoryScores = result.category_scores || {};
+  const flags = Object.entries(categories)
+    .filter(([, flagged]) => Boolean(flagged))
+    .map(([name]) => name);
+  const score = Math.max(0, ...Object.values(categoryScores).map((value) => Number(value) || 0));
+
+  return {
+    flagged: Boolean(result.flagged) || flags.length > 0,
+    flags,
+    score,
+    raw: response,
+  };
+}
+
+async function runLLMAnalysis(text: string, moderation: ModerationResult): Promise<string> {
+  const response = await retryWithBackoff(
+    () => fetchJson(`${config.AI_LLM_BASE_URL}/chat/completions`, {
+      method: "POST",
+      headers: {
+        "Authorization": `Bearer ${config.AI_LLM_API_KEY}`,
+        "Content-Type": "application/json",
+      },
+      body: JSON.stringify({
+        model: config.AI_LLM_MODEL,
+        messages: [
+          {
+            role: "system",
+            content: "Kamu analis moderation Discord. Jawab singkat dalam Bahasa Indonesia: ringkasan risiko, alasan, dan aksi yang disarankan. Jangan mengulang pesan mentah secara panjang.",
+          },
+          {
+            role: "user",
+            content: JSON.stringify({
+              message: text,
+              moderationFlagged: moderation.flagged,
+              moderationFlags: moderation.flags,
+              moderationScore: moderation.score,
+            }),
+          },
+        ],
+        temperature: 0.2,
+      }),
+    }),
+    { retries: 2, logger },
+  ) as ChatCompletionResponse;
+
+  return response.choices?.[0]?.message?.content?.trim() || "Tidak ada analisis dari LLM.";
+}
+
+async function analyzeAndStore(db: SqliteDatabase, message: MessageRecord): Promise<void> {
+  const text = getAnalysisText(message);
+  if (!config.AI_ANALYSIS_ENABLED || text.length === 0) return;
+
+  try {
+    const moderation = await runModeration(text);
+    const analysis = await runLLMAnalysis(text, moderation);
+    const row = updateMessageAIAnalysis(db, message.id, {
+      status: moderation.flagged ? "flagged" : "clean",
+      flags: JSON.stringify(moderation.flags),
+      score: moderation.score,
+      raw: JSON.stringify(moderation.raw),
+      analysis,
+      analyzedAt: Date.now(),
+      error: null,
+    });
+    if (row) (globalThis as any).broadcastMessageAnalyzed?.(row);
+  } catch (error) {
+    const row = updateMessageAIAnalysis(db, message.id, {
+      status: "error",
+      flags: null,
+      score: null,
+      raw: null,
+      analysis: null,
+      analyzedAt: Date.now(),
+      error: error instanceof Error ? error.message : String(error),
+    });
+    if (row) (globalThis as any).broadcastMessageAnalyzed?.(row);
+    logger.warn({ messageId: message.id, error }, "AI analysis failed");
+  }
+}
+
+async function drainQueue(db: SqliteDatabase): Promise<void> {
+  if (isProcessing) return;
+  isProcessing = true;
+  try {
+    while (queuedMessageIds.size > 0) {
+      const [messageId] = queuedMessageIds;
+      queuedMessageIds.delete(messageId);
+      const message = getMessageById(db, messageId);
+      if (message) await analyzeAndStore(db, message);
+    }
+  } finally {
+    isProcessing = false;
+  }
+}
+
+export function queueMessageAnalysis(db: SqliteDatabase, messageId: string): void {
+  if (!config.AI_ANALYSIS_ENABLED) return;
+  queuedMessageIds.add(messageId);
+  setImmediate(() => {
+    drainQueue(db).catch((error) => logger.error({ error }, "AI analysis queue failed"));
+  });
+}
@@ -5,6 +5,7 @@ import { config } from "../config";
 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";

 const logger = createChildLogger("message-capture");
@@ -35,6 +36,7 @@ export async function captureMessage(
  };

  insertMessage(db, messageRecord);
+  queueMessageAnalysis(db, message.id);

  const broadcaster = globalThis as any;
  if (broadcaster.broadcastMessageCreated) {
@@ -126,6 +128,7 @@ export function registerMessageCapture(client: Client, db: SqliteDatabase): void
      if (existing) {
        const editedAt = Date.now();
        updateMessageAsEdited(db, newMessage.id, getDisplayContent(newMessage as Message), editedAt);
+        queueMessageAnalysis(db, newMessage.id);

        const broadcaster = globalThis as any;
        if (broadcaster.broadcastMessageUpdated) {
@@ -220,3 +220,76 @@ export function updateAttachmentAsFailedUpload(
    throw error;
  }
 }
+
+interface AIAnalysisUpdate {
+  status: "pending" | "clean" | "flagged" | "error";
+  flags?: string | null;
+  score?: number | null;
+  raw?: string | null;
+  analysis?: string | null;
+  analyzedAt?: number | null;
+  error?: string | null;
+}
+
+export function updateMessageAIAnalysis(
+  db: SqliteDatabase,
+  messageId: string,
+  result: AIAnalysisUpdate,
+): MessageRecord | null {
+  try {
+    const stmt = db.prepare(`
+      UPDATE messages
+      SET ai_status = ?, ai_moderation_flags = ?, ai_moderation_score = ?,
+          ai_moderation_raw = ?, ai_analysis = ?, ai_analyzed_at = ?, ai_error = ?
+      WHERE id = ?
+    `);
+
+    stmt.run(
+      result.status,
+      result.flags ?? null,
+      result.score ?? null,
+      result.raw ?? null,
+      result.analysis ?? null,
+      result.analyzedAt ?? Date.now(),
+      result.error ?? null,
+      messageId,
+    );
+
+    const row = db.prepare("SELECT * FROM messages WHERE id = ?").get(messageId) as MessageRecord | undefined;
+    return row ?? null;
+  } catch (error) {
+    logger.error(
+      { messageId, error: error instanceof Error ? error.message : String(error) },
+      "Failed to update message AI analysis",
+    );
+    throw error;
+  }
+}
+
+export function getPendingAIAnalysisMessages(
+  db: SqliteDatabase,
+  limit: number = 25,
+): MessageRecord[] {
+  try {
+    const stmt = db.prepare(`
+      SELECT * FROM messages
+      WHERE ai_status = 'pending'
+        AND deleted_at IS NULL
+        AND COALESCE(edited_content, content) != ''
+      ORDER BY created_at ASC
+      LIMIT ?
+    `);
+    return stmt.all(limit) as MessageRecord[];
+  } catch (error) {
+    logger.error(
+      { error: error instanceof Error ? error.message : String(error) },
+      "Failed to get pending AI analysis messages",
+    );
+    throw error;
+  }
+}
+
+export function getMessageById(db: SqliteDatabase, messageId: string): MessageRecord | null {
+  const row = db.prepare("SELECT * FROM messages WHERE id = ?").get(messageId) as MessageRecord | undefined;
+  return row ?? null;
+}
@@ -13,6 +13,13 @@ export interface MessageRecord {
  deleted_at: number | null;
  type: "text" | "edited" | "deleted";
  metadata: string | null;
+  ai_status?: "pending" | "clean" | "flagged" | "error" | null;
+  ai_moderation_flags?: string | null;
+  ai_moderation_score?: number | null;
+  ai_moderation_raw?: string | null;
+  ai_analysis?: string | null;
+  ai_analyzed_at?: number | null;
+  ai_error?: string | null;
 }

 export interface AttachmentRecord {