feat(moderation): add support for vision model in moderation analysis

.env.example

@@ -48,6 +48,8 @@ AI_ANALYSIS_ENABLED=false
 AI_LLM_API_KEY=your_9router_key_here
 AI_LLM_BASE_URL=https://9router.asepharyana.tech/v1
 AI_LLM_MODEL=free
+# Vision model for image/video moderation (falls back to AI_LLM_MODEL if unset)
+AI_LLM_VISION_MODEL=multimodal
 
 # NVIDIA Nemotron Content Safety Configuration
 NVIDIA_NEMOTRON_API_KEY=your_nvidia_api_key_here

src/config.ts

@@ -71,14 +71,18 @@ const configSchema = z
       .string()
       .url()
       .default("https://9router.asepharyana.tech/v1"),
-    AI_LLM_MODEL: z.string().default("free"),
+    /** Model used for text-only moderation (messages, badword analysis). */
+    AI_LLM_MODEL: z.string().default("text"),
+    /** Model used for image/video moderation (vision-capable model). */
+    AI_LLM_VISION_MODEL: z.string().optional(),
     AI_ANALYSIS_DEBOUNCE_MS: z.coerce.number().positive().default(500),
     AI_ANALYSIS_RECOVERY_INTERVAL_MS: z.coerce
       .number()
       .positive()
       .default(15000),
     AI_ANALYSIS_ERROR_COOLDOWN_MS: z.coerce.number().positive().default(30000),
-    AI_ANALYSIS_MAX_BATCH_SIZE: z.coerce.number().int().positive().default(25),
+    /** Max messages fetched per conversation batch (token budget is the real constraint). */
+    AI_ANALYSIS_MAX_BATCH_SIZE: z.coerce.number().int().positive().default(200),
     AI_ANALYSIS_MAX_CONTEXT_TOKENS: z.coerce.number().positive().default(8000),
     /** Token budget for target messages specifically (separate from context window). */
     AI_ANALYSIS_MAX_TARGET_TOKENS: z.coerce.number().positive().default(4000),
@@ -96,15 +100,12 @@ const configSchema = z
       .number()
       .positive()
       .default(120000),
-    /**
-     * Maximum number of concurrent individual-fallback LLM calls.
-     * Prevents OOM/connection exhaustion when many messages miss a batch.
-     */
+    /** Max concurrent individual-fallback LLM calls (effectively unlimited). */
     AI_ANALYSIS_INDIVIDUAL_MAX_CONCURRENT: z.coerce
       .number()
       .int()
       .positive()
-      .default(20),
+      .default(1000),
     /**
      * How many consecutive individual-fallback errors trigger the individual
      * circuit breaker (separate from the batch circuit breaker).
@@ -113,7 +114,7 @@ const configSchema = z
       .number()
       .int()
       .positive()
-      .default(10),
+      .default(50),
     /** NVIDIA Nemotron-3 Content Safety API key for badword detection. */
     NVIDIA_NEMOTRON_API_KEY: z.string().optional(),
     /** NVIDIA Nemotron model identifier. */

src/moderation/aiAnalyzer.ts

@@ -486,34 +486,15 @@ function enqueueIndividualFallbacks(messages: MessageRecord[]): void {
   const newMessages = messages.filter((m) => !individualInFlight.has(m.id));
   if (newMessages.length === 0) return;
 
-  // FIX #1: Enforce concurrency cap.
-  const availableSlots =
-    config.AI_ANALYSIS_INDIVIDUAL_MAX_CONCURRENT - individualInFlight.size;
-  if (availableSlots <= 0) {
-    logger.warn(
-      {
-        cap: config.AI_ANALYSIS_INDIVIDUAL_MAX_CONCURRENT,
-        inFlight: individualInFlight.size,
-        skipped: newMessages.length,
-      },
-      "Individual fallback concurrency cap reached — messages will be recovered by recovery worker",
-    );
-    return;
-  }
-
-  const toProcess = newMessages.slice(0, availableSlots);
-  const skipped = newMessages.length - toProcess.length;
-
   logger.info(
     {
-      count: toProcess.length,
-      skipped,
-      messageIds: toProcess.map((m) => m.id),
+      count: newMessages.length,
+      messageIds: newMessages.map((m) => m.id),
     },
     "Enqueueing individual fallback analysis for batch-incomplete messages",
   );
 
-  for (const msg of toProcess) {
+  for (const msg of newMessages) {
     individualInFlight.add(msg.id);
     // Fire-and-forget: processIndividualFallback handles all errors internally.
     processIndividualFallback(msg).catch((err) => {
@@ -855,8 +836,8 @@ export function startPendingAIAnalysisWorker(client?: Client): void {
   setInterval(() => {
     // FIX #3 pattern: no async arrow — chain promises explicitly.
     Promise.all([
-      getPendingConversationKeys(100),
-      getConversationKeysWithIncompleteAnalysis(50),
+      getPendingConversationKeys(500),
+      getConversationKeysWithIncompleteAnalysis(200),
     ])
       .then(([pendingKeys, incompleteKeys]) => {
         const now = Date.now();
@@ -902,10 +883,7 @@ export function startPendingAIAnalysisWorker(client?: Client): void {
             if (isConversationProcessingLocked(key)) continue;
 
             promises.push(
-              getIncompleteMessagesByConversation(
-                key,
-                config.AI_ANALYSIS_INDIVIDUAL_MAX_CONCURRENT,
-              )
+              getIncompleteMessagesByConversation(key, 500)
                 .then(async (msgs) => {
                   const processableMessages =
                     await skipAgeRestrictedMessages(msgs);

src/moderation/llmModerationClient.ts

@@ -769,7 +769,7 @@ export async function runModerationAnalysis(
   ): Promise<string | null> => {
     try {
       const completion = await openai.chat.completions.create({
-        model: config.AI_LLM_MODEL,
+        model: config.AI_LLM_VISION_MODEL ?? config.AI_LLM_MODEL,
         messages: [
           {
             role: "user",

src/moderation/messageStore.ts

@@ -630,7 +630,7 @@ export async function getConversationContextBefore(input: {
 
 export async function getPendingMessagesByConversation(
   conversationKey: string,
-  limit: number = 25,
+  limit: number = 200,
 ): Promise<MessageRecord[]> {
   try {
     const database = db();
@@ -667,7 +667,7 @@ export async function getPendingMessagesByConversation(
 }
 
 export async function getPendingConversationKeys(
-  limit: number = 100,
+  limit: number = 500,
 ): Promise<string[]> {
   try {
     const database = db();
@@ -782,7 +782,7 @@ export async function searchMessages(input: {
  * the individual-fallback queue.
  */
 export async function getConversationKeysWithIncompleteAnalysis(
-  limit: number = 50,
+  limit: number = 200,
 ): Promise<string[]> {
   try {
     const database = db();
@@ -826,7 +826,7 @@ export async function getConversationKeysWithIncompleteAnalysis(
  */
 export async function getIncompleteMessagesByConversation(
   conversationKey: string,
-  limit: number = 20,
+  limit: number = 500,
 ): Promise<MessageRecord[]> {
   try {
     const database = db();