refactor: streamline error handling and attachment processing in message capture

src/moderation/aiAnalysisWorker.ts

@@ -97,19 +97,6 @@ async function processAnalysisRequest({
     const errorMessage = error instanceof Error ? error.message : String(error);
     const rows: MessageRecord[] = [];
 
-    for (const msg of messages) {
-      const row = await updateMessageAIAnalysis(msg.id, {
-        status: "error",
-        flags: null,
-        score: null,
-        raw: null,
-        analysis: null,
-        analyzedAt: Date.now(),
-        error: errorMessage,
-      });
-      if (row) rows.push(row);
-    }
-
     return { ok: false, conversationKey, rows, error: errorMessage };
   }
 }

src/moderation/aiAnalyzer.ts

@@ -32,7 +32,6 @@ const conversationErrorCooldown = new Map<string, number>();
 
 let activeRequests = 0;
 let lastError: string | null = null;
-const MAX_ACTIVE_REQUESTS = 2;
 const DEBOUNCE_MS = 1500;
 const RECOVERY_INTERVAL_MS = 15000;
 const ERROR_COOLDOWN_MS = 30000;
@@ -104,7 +103,7 @@ async function processBatch(
       );
       logger.error(
         { conversationKey, error: lastError },
-        "Batch analysis failed",
+        "Batch analysis failed, will retry after cooldown",
       );
       return;
     }
@@ -118,21 +117,8 @@ async function processBatch(
     );
     logger.error(
       { conversationKey, error: lastError },
-      "Analysis worker failed",
+      "Analysis worker failed, will retry after cooldown",
     );
-
-    for (const msg of messages) {
-      const row = await updateMessageAIAnalysis(msg.id, {
-        status: "error",
-        flags: null,
-        score: null,
-        raw: null,
-        analysis: null,
-        analyzedAt: Date.now(),
-        error: lastError,
-      });
-      if (row) getModerationBroadcaster()?.messageAnalyzed(row);
-    }
   } finally {
     activeRequests--;
     conversationProcessing.delete(conversationKey);
@@ -186,22 +172,13 @@ function scheduleConversationAnalysis(conversationKey: string): void {
     clearTimeout(existingTimer);
   }
 
-  // If we have available slots, process immediately with shorter debounce
+  // Always use shorter debounce for immediate processing (no concurrency limit)
-  const debounceTime =
+  const debounceTime = Math.min(DEBOUNCE_MS, 500);
-    activeRequests < MAX_ACTIVE_REQUESTS
-      ? Math.min(DEBOUNCE_MS, 500)
-      : DEBOUNCE_MS;
 
   // Set new debounced timer
   const timer = setTimeout(async () => {
     conversationDebounceTimers.delete(conversationKey);
 
-    // If activeRequests >= MAX_ACTIVE_REQUESTS, requeue instead of waiting
-    if (activeRequests >= MAX_ACTIVE_REQUESTS) {
-      scheduleConversationAnalysis(conversationKey);
-      return;
-    }
-
     // Get pending messages for this conversation
     const messages = await getPendingMessagesByConversation(
       conversationKey,
@@ -277,11 +254,6 @@ export function startPendingAIAnalysisWorker(): void {
       const conversationKeys = await getPendingConversationKeys(100);
 
       for (const key of conversationKeys) {
-        // Stop if we've reached max active requests
-        if (activeRequests >= MAX_ACTIVE_REQUESTS) {
-          break;
-        }
-
         // Skip if already scheduled
         if (conversationDebounceTimers.has(key)) {
           continue;

src/moderation/llmModerationClient.ts

@@ -81,6 +81,30 @@ export function parseModerationResponse(
   // If parsed is a direct array, wrap it in a results object to handle LLM variations
   if (Array.isArray(parsed)) {
     parsed = { results: parsed };
+  } else if (parsed && typeof parsed === "object" && !("results" in parsed)) {
+    // Handle single result object (has message_id or status)
+    if ("message_id" in parsed || "status" in parsed) {
+      const msgId = (parsed as any).message_id || (parsed as any).id;
+      parsed = {
+        results: [
+          {
+            message_id: msgId,
+            status: (parsed as any).status || "clean",
+            flags: (parsed as any).flags || [],
+            score: (parsed as any).score !== undefined ? (parsed as any).score : 0.1,
+            analysis: (parsed as any).analysis || "",
+          },
+        ],
+      };
+    } else {
+      // Look for any array property (result, data, messages, moderation, etc.)
+      const arrayKey = Object.keys(parsed).find(
+        (key) => Array.isArray((parsed as any)[key]),
+      );
+      if (arrayKey) {
+        parsed.results = (parsed as any)[arrayKey];
+      }
+    }
   }
 
   // Validate structure

src/moderation/messageCapture.ts

@@ -2,6 +2,7 @@ import type { Client, Message } from "discord.js-selfbot-v13";
 import { config } from "../config";
 import { createChildLogger } from "../logger";
 import { queueMessageAnalysis } from "./aiAnalyzer";
+import { processAttachmentUpload } from "./attachmentUploader";
 import {
   getDisplayContent,
   getMessageLocation,
@@ -87,15 +88,13 @@ export async function captureMessage(
   }
 
   const isBacklog = options.source === "backlog";
-  if (!isBacklog) {
-    queueMessageAnalysis(message.id);
-  }
 
   const broadcaster = getModerationBroadcaster();
   if (broadcaster && !isBacklog) {
     broadcaster.messageCreated(messageRecord);
   }
 
+  // Insert attachments before queuing analysis to avoid race condition
   if (message.attachments.size > 0) {
     for (const [, attachment] of message.attachments) {
       const attachmentRecord: AttachmentRecord = {
@@ -109,20 +108,39 @@ export async function captureMessage(
         size: attachment.size,
         type: attachment.contentType || "application/octet-stream",
         discord_url: attachment.url,
-        uploaded_url: attachment.url,
+        uploaded_url: null,
-        upload_status: "uploaded",
+        upload_status: "pending",
         upload_error: null,
         created_at: Date.now(),
-        uploaded_at: Date.now(),
+        uploaded_at: null,
       };
 
       await insertAttachment(attachmentRecord);
 
+      // Initiate async upload to Picser (non-blocking, fire-and-forget)
+      if (!isBacklog) {
+        processAttachmentUpload(
+          attachment.id,
+          attachment.url,
+          attachment.name || "unknown",
+        ).catch((err) => {
+          logger.error(
+            { attachmentId: attachment.id, error: err },
+            "Failed to initiate attachment upload",
+          );
+        });
+      }
+
       if (broadcaster) {
         broadcaster.attachmentCreated(attachmentRecord);
       }
     }
   }
+
+  // Queue analysis after attachments are inserted
+  if (!isBacklog) {
+    queueMessageAnalysis(message.id);
+  }
 }
 
 export function registerMessageCapture(client: Client): void {