feat: add multimodal analysis support to LLM moderation client by processing image attachments

src/errors.ts

@@ -2,11 +2,7 @@ export class AppError extends Error {
   public code: string;
   public statusCode: number;
 
-  constructor(
-    message: string,
-    code: string,
-    statusCode: number = 500,
-  ) {
+  constructor(message: string, code: string, statusCode: number = 500) {
     super(message);
     this.code = code;
     this.statusCode = statusCode;
@@ -39,10 +35,7 @@ export class VoiceConnectionError extends AppError {
 export class ValidationError extends AppError {
   public details?: Record<string, string[]>;
 
-  constructor(
-    message: string,
-    details?: Record<string, string[]>,
-  ) {
+  constructor(message: string, details?: Record<string, string[]>) {
     super(message, "VALIDATION_ERROR", 400);
     this.details = details;
     this.name = "ValidationError";

src/index.ts

@@ -74,7 +74,12 @@ async function initializeApp() {
   }
 
   client.on("debug", (msg) => {
-    if (msg.includes("[VOICE") || msg.includes("[ffmpeg") || msg.toLowerCase().includes("error") || msg.toLowerCase().includes("stream")) {
+    if (
+      msg.includes("[VOICE") ||
+      msg.includes("[ffmpeg") ||
+      msg.toLowerCase().includes("error") ||
+      msg.toLowerCase().includes("stream")
+    ) {
       logger.info({ debugMsg: msg }, "Discord Client Debug");
     } else if (config.VERBOSE) {
       logger.debug({ debugMsg: msg }, "Discord Client Debug");

src/media/mediaController.ts

@@ -17,7 +17,10 @@ import { createMusicPlayer } from "./musicPlayer";
 export interface MediaControllerDependencies {
   isVoiceConnected?: () => boolean;
   isBrowserStreaming?: () => boolean;
-  resolveMediaSource?: (source: string, mode?: MediaMode) => Promise<ResolvedMediaSource>;
+  resolveMediaSource?: (
+    source: string,
+    mode?: MediaMode,
+  ) => Promise<ResolvedMediaSource>;
   musicPlayer?: MusicPlayer;
   screenController?: ScreenShareController;
   onStateChange?: (state: MediaState) => void;
@@ -91,7 +94,10 @@ export class MediaController {
     // reject to avoid stealing the shared player. If this controller started
     // the screenPlayback, stop it and proceed.
     if (this.screenPlayback || this.dependencies.screenController?.isActive()) {
-      if (this.dependencies.screenController?.isActive() && !this.screenPlayback) {
+      if (
+        this.dependencies.screenController?.isActive() &&
+        !this.screenPlayback
+      ) {
         throw new AppError("Another media mode is active", "MEDIA_BUSY", 409);
       }
       this.screenPlayback?.stop();

src/media/mediaResolver.ts

@@ -20,7 +20,7 @@ export function createMediaResolver(
 
   return async function resolve(
     input: string,
-    mode: MediaMode = "music"
+    mode: MediaMode = "music",
   ): Promise<ResolvedMediaSource> {
     const source = input.trim();
     if (!source) {
@@ -34,17 +34,19 @@ export function createMediaResolver(
     const url = parseUrl(source);
     if (url && isYouTubeUrl(url)) {
       const metadata = await ytdlp.getMetadata(source);
-      const directUrl = mode === "screen" 
-        ? await ytdlp.getDirectVideoUrl(source)
-        : await ytdlp.getDirectAudioUrl(source);
+      const directUrl =
+        mode === "screen"
+          ? await ytdlp.getDirectVideoUrl(source)
+          : await ytdlp.getDirectAudioUrl(source);
       return { source: directUrl, title: metadata.title, kind: "youtube" };
     }
 
     if (url && isSpotifyTrackUrl(url)) {
       const result = await playDlResolver.resolveSpotifyTrack(source);
-      const directUrl = mode === "screen"
-        ? await ytdlp.getDirectVideoUrl(result.url)
-        : await ytdlp.getDirectAudioUrl(result.url);
+      const directUrl =
+        mode === "screen"
+          ? await ytdlp.getDirectVideoUrl(result.url)
+          : await ytdlp.getDirectAudioUrl(result.url);
       return { source: directUrl, title: result.title, kind: "spotify" };
     }
 
@@ -62,9 +64,10 @@ export function createMediaResolver(
 
     if (!url && !looksLikeUrl(source)) {
       const result = await playDlResolver.searchYouTube(source);
-      const directUrl = mode === "screen"
-        ? await ytdlp.getDirectVideoUrl(result.url)
-        : await ytdlp.getDirectAudioUrl(result.url);
+      const directUrl =
+        mode === "screen"
+          ? await ytdlp.getDirectVideoUrl(result.url)
+          : await ytdlp.getDirectAudioUrl(result.url);
       return { source: directUrl, title: result.title, kind: "search" };
     }
 

src/media/screenShareController.ts

@@ -1,7 +1,4 @@
-import {
-  Streamer,
-  playPreparedStream,
-} from "../streaming";
+import { Streamer, playPreparedStream } from "../streaming";
 import { AppError } from "../errors";
 import { createChildLogger } from "../logger";
 import { discordPlayer } from "../player";
@@ -23,8 +20,14 @@ export interface ScreenShareControllerDependencies {
   getDirectVideoUrl?: (source: string) => Promise<string>;
   streamer: Streamer;
   useTranscoder?: boolean;
-  onBeforeStreamStart?: (guildId: string, channelId: string) => Promise<void> | void;
-  onAfterStreamEnd?: (guildId: string, channelId: string) => Promise<void> | void;
+  onBeforeStreamStart?: (
+    guildId: string,
+    channelId: string,
+  ) => Promise<void> | void;
+  onAfterStreamEnd?: (
+    guildId: string,
+    channelId: string,
+  ) => Promise<void> | void;
   onStreamStart?: () => void;
   onStreamEnd?: () => void;
 }

src/moderation/aiAnalysisWorker.ts

@@ -3,6 +3,7 @@ import { initializeDatabase } from "../database/drizzle.ts";
 import { buildConversationPromptMessages } from "./conversationContext.ts";
 import { runModerationAnalysis } from "./llmModerationClient.ts";
 import {
+  getAttachmentsForMessages,
   getConversationContextBefore,
   updateMessageAIAnalysis,
 } from "./messageStore.ts";
@@ -66,9 +67,15 @@ async function processAnalysisRequest({
       maxTokens: MAX_CONTEXT_TOKENS,
     });
 
+    const targetIds = messages.map((m) => m.id);
+    const contextIds = contextBefore.map((m) => m.id);
+    const allMessageIds = [...targetIds, ...contextIds];
+    const attachments = await getAttachmentsForMessages(allMessageIds);
+
     const result = await runModerationAnalysis({
       targets: messages,
       contextText: promptMessages.join("\n"),
+      attachments,
     });
 
     const rows: MessageRecord[] = [];

src/moderation/aiAnalyzer.ts

@@ -146,7 +146,7 @@ async function runAnalysisInWorker(
   return new Promise((resolve, reject) => {
     const worker = new Worker(
       new URL("./aiAnalysisWorker.ts", import.meta.url),
-      { execArgv: process.execArgv }
+      { execArgv: process.execArgv },
     );
 
     worker.once("message", (response: AnalysisWorkerResponse) => {

src/moderation/llmModerationClient.ts

@@ -1,7 +1,7 @@
 import { config } from "../config.ts";
 import { createChildLogger } from "../logger.ts";
 import { retryWithBackoff } from "../retry.ts";
-import type { AnalysisResult, MessageRecord } from "./types";
+import type { AnalysisResult, AttachmentRecord, MessageRecord } from "./types";
 
 const log = createChildLogger("llmModerationClient");
 
@@ -174,6 +174,7 @@ export function parseModerationResponse(
 interface ModerationInput {
   targets: MessageRecord[];
   contextText: string;
+  attachments?: AttachmentRecord[];
 }
 
 interface ModerationOutput {
@@ -188,7 +189,7 @@ interface ModerationOutput {
 export async function runModerationAnalysis(
   input: ModerationInput,
 ): Promise<ModerationOutput> {
-  const { targets, contextText } = input;
+  const { targets, contextText, attachments } = input;
 
   if (!targets.length) {
     throw new Error("No targets provided for analysis");
@@ -220,6 +221,88 @@ Each result must have:
 
 Return ONLY valid JSON, no other text.`;
 
+  // Check for image attachments to support multimodal analysis
+  const imageAttachments = (attachments || []).filter(
+    (att) =>
+      (att.uploaded_url || att.discord_url) && att.type.startsWith("image/"),
+  );
+
+  let messageContent:
+    | string
+    | Array<{ type: string; text?: string; image_url?: { url: string } }>;
+  if (imageAttachments.length > 0) {
+    const contentParts: Array<{
+      type: string;
+      text?: string;
+      image_url?: { url: string };
+    }> = [];
+
+    // Download and convert all images to base64 data URLs
+    for (const att of imageAttachments) {
+      try {
+        const urlToUse = att.uploaded_url || att.discord_url;
+        log.info(
+          { attachmentId: att.id, url: urlToUse },
+          "Downloading attachment for base64 encoding",
+        );
+        const res = await fetch(urlToUse);
+        if (res.ok) {
+          const buffer = await res.arrayBuffer();
+          const base64Str = Buffer.from(buffer).toString("base64");
+          const dataUrl = `data:${att.type};base64,${base64Str}`;
+
+          contentParts.push({
+            type: "image_url",
+            image_url: {
+              url: dataUrl,
+            },
+          });
+
+          contentParts.push({
+            type: "text",
+            text: `\n[Image Attachment for Message ID: ${att.message_id}, Filename: ${att.filename}]`,
+          });
+        } else {
+          log.warn(
+            { attachmentId: att.id, status: res.status },
+            "Failed to fetch attachment image",
+          );
+        }
+      } catch (err) {
+        log.warn(
+          {
+            attachmentId: att.id,
+            error: err instanceof Error ? err.message : String(err),
+          },
+          "Error base64 encoding attachment",
+        );
+      }
+    }
+
+    contentParts.push({
+      type: "text",
+      text: prompt,
+    });
+
+    messageContent = contentParts;
+  } else {
+    // If no image is present, send a transparent 1x1 dummy PNG to satisfy multimodal omni requirements
+    const dummyPng =
+      "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg==";
+    messageContent = [
+      {
+        type: "image_url",
+        image_url: {
+          url: dummyPng,
+        },
+      },
+      {
+        type: "text",
+        text: prompt,
+      },
+    ];
+  }
+
   const result = await retryWithBackoff(
     async () => {
       const controller = new AbortController();
@@ -243,10 +326,14 @@ Return ONLY valid JSON, no other text.`;
               messages: [
                 {
                   role: "user",
-                  content: prompt,
+                  content: messageContent,
                 },
               ],
-              temperature: 0.3,
+              temperature: 0.6,
+              top_p: 0.95,
+              max_tokens: 65536,
+              reasoning_budget: 16384,
+              chat_template_kwargs: { enable_thinking: true },
             }),
           },
         );

src/moderation/messageStore.ts

@@ -1,4 +1,14 @@
-import { and, asc, desc, eq, isNull, or, type SQL, sql } from "drizzle-orm";
+import {
+  and,
+  asc,
+  desc,
+  eq,
+  inArray,
+  isNull,
+  or,
+  type SQL,
+  sql,
+} from "drizzle-orm";
 import { getDatabase } from "../database/drizzle.ts";
 import { attachmentsTable, messagesTable } from "../database/schema.ts";
 import { createChildLogger } from "../logger.ts";
@@ -605,3 +615,27 @@ export async function getPendingConversationKeys(
     throw error;
   }
 }
+
+export async function getAttachmentsForMessages(
+  messageIds: string[],
+): Promise<AttachmentRecord[]> {
+  try {
+    if (messageIds.length === 0) return [];
+    const database = db();
+    const rows = await database
+      .select()
+      .from(attachmentsTable)
+      .where(inArray(attachmentsTable.message_id, messageIds));
+
+    return rows as AttachmentRecord[];
+  } catch (error) {
+    logger.error(
+      {
+        messageIds,
+        error: error instanceof Error ? error.message : String(error),
+      },
+      "Failed to get attachments for messages",
+    );
+    throw error;
+  }
+}

src/streaming/index.ts

@@ -41,7 +41,10 @@ export class Streamer {
     this.dankStreamer = new DankStreamer(client);
   }
 
-  async createSession(guildId: string, channelId: string): Promise<StreamSession> {
+  async createSession(
+    guildId: string,
+    channelId: string,
+  ): Promise<StreamSession> {
     await this.dankStreamer.joinVoice(guildId, channelId);
 
     let stopped = false;
@@ -62,7 +65,10 @@ export class Streamer {
     return {
       connection: {} as any,
       stream: {} as any,
-      play: async (source: string | Readable, options: StreamPlayOptions = {}) => {
+      play: async (
+        source: string | Readable,
+        options: StreamPlayOptions = {},
+      ) => {
         if (stopped) return;
 
         let targetSource: string | Readable = source;
@@ -75,7 +81,12 @@ export class Streamer {
         const bitrateStr = String(options.bitrate ?? 8000).replace(/k$/i, "");
         const bitrateVideo = parseInt(bitrateStr, 10) || 8000;
 
-        console.log("[Streamer] Starting screen share for source:", typeof targetSource === "string" ? targetSource.slice(0, 50) + "..." : "ReadableStream");
+        console.log(
+          "[Streamer] Starting screen share for source:",
+          typeof targetSource === "string"
+            ? targetSource.slice(0, 50) + "..."
+            : "ReadableStream",
+        );
         const { command, output } = dankPrepareStream(targetSource, {
           encoder: Encoders.software({
             x264: { preset: (options.presetH26x as any) ?? "ultrafast" },
@@ -100,7 +111,7 @@ export class Streamer {
 
         const webOutput = new PassThrough();
         const discordOutput = new PassThrough();
-        
+
         output.pipe(webOutput);
         output.pipe(discordOutput);
 

src/streaming/transcoder.ts

@@ -21,7 +21,10 @@ export class Transcoder {
   restartTimer: NodeJS.Timeout | null = null;
   maxRestarts = 6;
 
-  constructor(private source: string, private opts: TranscoderOptions = {}) {}
+  constructor(
+    private source: string,
+    private opts: TranscoderOptions = {},
+  ) {}
 
   start(): { command: ChildProcess; output: Readable } {
     const fps = this.opts.fps ?? 30;
@@ -99,13 +102,19 @@ export class Transcoder {
 
   scheduleRestart() {
     if (this.restartAttempts >= this.maxRestarts) {
-      logger.error({ attempts: this.restartAttempts }, "transcoder reached max restart attempts");
+      logger.error(
+        { attempts: this.restartAttempts },
+        "transcoder reached max restart attempts",
+      );
       return;
     }
     const delay = Math.min(30000, 1000 * Math.pow(2, this.restartAttempts));
     this.restartAttempts += 1;
     transcoderRestartsCounter.inc();
-    logger.info({ delay, attempt: this.restartAttempts }, "scheduling transcoder restart");
+    logger.info(
+      { delay, attempt: this.restartAttempts },
+      "scheduling transcoder restart",
+    );
     this.restartTimer = setTimeout(() => {
       try {
         this.start();
@@ -129,9 +138,9 @@ export class Transcoder {
       clearTimeout(this.restartTimer);
       this.restartTimer = null;
     }
-      if (this.proc && !this.proc.killed) {
-        return new Promise<void>((resolve) => {
-          this.proc?.once("exit", () => resolve());
+    if (this.proc && !this.proc.killed) {
+      return new Promise<void>((resolve) => {
+        this.proc?.once("exit", () => resolve());
         try {
           this.proc?.kill("SIGTERM");
         } catch {
@@ -141,7 +150,7 @@ export class Transcoder {
             resolve();
           }
         }
-          setTimeout(() => resolve(), 5000);
+        setTimeout(() => resolve(), 5000);
       }).then(() => {
         this.proc = null;
         this.output = null;
@@ -151,7 +160,10 @@ export class Transcoder {
   }
 }
 
-export function prepareTranscoder(source: string, options: TranscoderOptions = {}) {
+export function prepareTranscoder(
+  source: string,
+  options: TranscoderOptions = {},
+) {
   const t = new Transcoder(source, options);
   const { command, output } = t.start();
   return { transcoder: t, command, output };