diff --git a/src/moderation/aiAnalysisWorker.ts b/src/moderation/aiAnalysisWorker.ts index baed009..b08363a 100644 --- a/src/moderation/aiAnalysisWorker.ts +++ b/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 }; } } diff --git a/src/moderation/aiAnalyzer.ts b/src/moderation/aiAnalyzer.ts index 54ff87e..366b0d3 100644 --- a/src/moderation/aiAnalyzer.ts +++ b/src/moderation/aiAnalyzer.ts @@ -32,7 +32,6 @@ const conversationErrorCooldown = new Map(); 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 - const debounceTime = - activeRequests < MAX_ACTIVE_REQUESTS - ? Math.min(DEBOUNCE_MS, 500) - : DEBOUNCE_MS; + // Always use shorter debounce for immediate processing (no concurrency limit) + const debounceTime = Math.min(DEBOUNCE_MS, 500); // 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; diff --git a/src/moderation/llmModerationClient.ts b/src/moderation/llmModerationClient.ts index f2e242b..fb4e487 100644 --- a/src/moderation/llmModerationClient.ts +++ b/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 diff --git a/src/moderation/messageCapture.ts b/src/moderation/messageCapture.ts index 9c28f31..49225b8 100644 --- a/src/moderation/messageCapture.ts +++ b/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, - upload_status: "uploaded", + uploaded_url: null, + 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 {