diff --git a/services/backend/src/modules/messages/messages.routes.ts b/services/backend/src/modules/messages/messages.routes.ts index 06001a6..4d66dd4 100644 --- a/services/backend/src/modules/messages/messages.routes.ts +++ b/services/backend/src/modules/messages/messages.routes.ts @@ -14,14 +14,21 @@ import { messagesService } from "./messages.service.js"; const logger = createChildLogger("messages.routes"); /** - * Per-message in-flight guard for the reanalyze endpoint. + * Per-message in-flight guard for the single reanalyze endpoint. * Prevents concurrent spam-clicks from issuing duplicate UPDATE + recovery - * worker triggers for the same message. Released as soon as the DB write - * completes (or fails), which is fast enough that false-positive blocking - * is not a practical concern. + * worker triggers for the same message. */ const reanalyzeInFlight = new Set(); +/** + * Per-scope in-flight guard for the batch reanalyze endpoint. + * Scope key = "guildId:channelId" (empty string used for undefined parts). + * Two concurrent batch-reanalyze requests for the same scope are rejected + * with 409 so the recovery worker is not triggered multiple times for the + * same set of error messages. + */ +const reanalyzeBatchInFlight = new Set(); + export function createMessagesRouter(): Router { const router = express.Router(); @@ -50,22 +57,40 @@ export function createMessagesRouter(): Router { messageIds?: string[]; }; - const count = await messagesService.reanalyzeErrorBatch({ - guildId, - channelId, - messageIds, - }); + // Idempotency guard: one concurrent batch-reanalyze per scope. + // Prevents two admin sessions clicking simultaneously from each + // triggering the recovery worker for the same set of messages. + const scopeKey = `${guildId ?? ""}:${channelId ?? ""}`; + if (reanalyzeBatchInFlight.has(scopeKey)) { + res + .status(409) + .json({ error: "REANALYZE_BATCH_IN_PROGRESS", scope: scopeKey }); + return; + } + + reanalyzeBatchInFlight.add(scopeKey); + let count = 0; + try { + count = await messagesService.reanalyzeErrorBatch({ + guildId, + channelId, + messageIds, + }); + } finally { + reanalyzeBatchInFlight.delete(scopeKey); + } logger.info({ count, guildId, channelId }, "Batch reanalyze completed"); res.status(200).json({ ok: true, count }); }), ); + // POST /api/messages/:id/reanalyze - Mark single message for re-analysis router.post( "/messages/:id/reanalyze", asyncHandler(async (req: Request, res: Response) => { - const id = req.params.id; + const id = String(req.params.id ?? ""); if (!id) { res.status(400).json({ error: "MISSING_ID" }); return; diff --git a/services/discord-gateway/src/modules/ai-moderation/aiAnalyzer.ts b/services/discord-gateway/src/modules/ai-moderation/aiAnalyzer.ts index ce7e13f..3247e03 100644 --- a/services/discord-gateway/src/modules/ai-moderation/aiAnalyzer.ts +++ b/services/discord-gateway/src/modules/ai-moderation/aiAnalyzer.ts @@ -61,9 +61,21 @@ function broadcastAnalysisCompleted(row: MessageRecord): void { function scheduleAutoDelete(row: MessageRecord): void { if (row.ai_status !== "flagged" && row.ai_status !== "warn") return; + + // Idempotency guard: if a concurrent path (batch + individual fallback) both + // produce a result for the same message, only the first call proceeds. + if (autoDeleteInFlight.has(row.id)) { + logger.debug( + { messageId: row.id }, + "Auto-delete skipped: already in-flight for this message", + ); + return; + } + autoDeleteInFlight.add(row.id); + const run = () => { - attemptAutoDeleteFlaggedMessage(moderationClient, row).catch( - (error: unknown) => { + attemptAutoDeleteFlaggedMessage(moderationClient, row) + .catch((error: unknown) => { logger.error( { messageId: row.id, @@ -71,8 +83,10 @@ function scheduleAutoDelete(row: MessageRecord): void { }, "Unexpected auto-delete error", ); - }, - ); + }) + .finally(() => { + autoDeleteInFlight.delete(row.id); + }); }; if (config.AUTO_DELETE_FLAGGED_DELAY_MS > 0) { @@ -82,6 +96,7 @@ function scheduleAutoDelete(row: MessageRecord): void { setImmediate(run); } + function isAgeRestrictedMessage(message: MessageRecord): boolean { return isAgeRestrictedMetadata(message.metadata); } @@ -148,6 +163,16 @@ const conversationProcessing = new Map(); /** Cooldown expiry timestamp per conversation key after an error. */ const conversationErrorCooldown = new Map(); +/** + * Per-message in-flight guard for the auto-delete side-effect. + * `scheduleAutoDelete` is called from both `processBatch` (on batch success) + * and `processIndividualFallback` (on individual success). For a message that + * races through both paths, without this guard two concurrent + * `attemptAutoDeleteFlaggedMessage` calls would be launched — producing a + * duplicate moderation-action log and an unnecessary Discord 10008 error. + */ +const autoDeleteInFlight = new Set(); + let activeRequests = 0; let lastError: string | null = null; let moderationClient: Client | undefined; @@ -875,6 +900,11 @@ async function processBatch( * .catch() so DB errors don't produce unhandled promise rejections. * FIX #6: Calls pickBatchWithinBudget after fetching messages so token budget * is respected before handing the batch to the LLM. + * FIX #7: Unified single-timer path — always clear-and-reset one timer per + * conversation key regardless of whether a cooldown is active. The delay is + * simply max(cooldownRemainder+500, debounce) so the same timer serves both + * the "throttled by error cooldown" and "normal debounce" cases, eliminating + * the previous two-path logic that could leave both timers live simultaneously. */ function scheduleConversationAnalysis(conversationKey: string): void { if (isConversationProcessingLocked(conversationKey)) { @@ -884,24 +914,20 @@ function scheduleConversationAnalysis(conversationKey: string): void { const convoCooldown = conversationErrorCooldown.get(conversationKey) ?? 0; const convoErrors = conversationConsecutiveErrors.get(conversationKey) ?? 0; - if (convoCooldown && Date.now() < convoCooldown) { - if (!conversationDebounceTimers.has(conversationKey)) { - const remaining = convoCooldown - Date.now(); - const timer = setTimeout(() => { - conversationDebounceTimers.delete(conversationKey); - scheduleConversationAnalysis(conversationKey); - }, remaining + 500); - conversationDebounceTimers.set(conversationKey, timer); - } - return; - } - - // Block scheduling if this conversation already hit the per-conversation - // error threshold — prevents rapid retry loops on the same broken convo. + // Hard-block: circuit breaker threshold reached AND cooldown still active. if (convoErrors >= MAX_CONSECUTIVE_ERRORS && Date.now() < convoCooldown) { return; } + // Unified delay: honour the cooldown window if active, otherwise use the + // normal debounce interval. Always clear-and-reset so only ONE timer is + // ever pending per conversation key regardless of call source. + const now = Date.now(); + const delayMs = + convoCooldown > now + ? convoCooldown - now + 500 + : config.AI_ANALYSIS_DEBOUNCE_MS; + const existingTimer = conversationDebounceTimers.get(conversationKey); if (existingTimer) { clearTimeout(existingTimer); @@ -956,11 +982,12 @@ function scheduleConversationAnalysis(conversationKey: string): void { "Failed to fetch or dispatch pending messages for scheduled analysis", ); }); - }, config.AI_ANALYSIS_DEBOUNCE_MS); + }, delayMs); conversationDebounceTimers.set(conversationKey, timer); } + // --------------------------------------------------------------------------- // Public API // --------------------------------------------------------------------------- diff --git a/services/discord-gateway/src/modules/message-capture/messageCapture.ts b/services/discord-gateway/src/modules/message-capture/messageCapture.ts index 2b861ef..16066ac 100644 --- a/services/discord-gateway/src/modules/message-capture/messageCapture.ts +++ b/services/discord-gateway/src/modules/message-capture/messageCapture.ts @@ -236,6 +236,21 @@ export function registerMessageCapture(client: Client): void { const existing = await getMessageById(newMessage.id); if (existing) { + const newContent = getDisplayContent(newMessage as Message); + const existingContent = existing.edited_content ?? existing.content; + + // Skip if the displayed text is identical — Discord fires `messageUpdate` + // for embed resolution (link previews) which does NOT change the message + // body. Re-setting ai_status + re-queuing LLM in that case wastes a call + // and risks overwriting a valid completed analysis with a duplicate. + if (newContent === existingContent) { + logger.debug( + { messageId: newMessage.id }, + "messageUpdate skipped: content unchanged (embed resolution or no-op)", + ); + return; + } + const editedAt = Date.now(); await updateMessageAsEdited( newMessage.id, diff --git a/services/discord-gateway/src/modules/message-capture/messageStore.ts b/services/discord-gateway/src/modules/message-capture/messageStore.ts index fb488de..ae4d482 100644 --- a/services/discord-gateway/src/modules/message-capture/messageStore.ts +++ b/services/discord-gateway/src/modules/message-capture/messageStore.ts @@ -807,15 +807,21 @@ export async function getConversationKeysWithIncompleteAnalysis( try { const database = db(); const rows = await database - .selectDistinct>({ - thread_id: messagesTable.thread_id, - channel_id: messagesTable.channel_id, - }) + .selectDistinct>( + { + thread_id: messagesTable.thread_id, + channel_id: messagesTable.channel_id, + }, + ) .from(messagesTable) .where( and( eq(messagesTable.ai_status, "error"), sql`${messagesTable.ai_moderation_flags} LIKE ${"%analysis_incomplete%"}`, + // Exclude rows that have already been exhausted by the individual + // fallback pipeline — prevents an infinite recovery loop if both + // flags are ever written to the same row due to a bug. + sql`(${messagesTable.ai_moderation_flags} IS NULL OR ${messagesTable.ai_moderation_flags} NOT LIKE ${"%individual_analysis_exhausted%"})`, isNull(messagesTable.deleted_at), ), ) @@ -861,6 +867,9 @@ export async function getIncompleteMessagesByConversation( ), eq(messagesTable.ai_status, "error"), sql`${messagesTable.ai_moderation_flags} LIKE ${"%analysis_incomplete%"}`, + // Same guard as getConversationKeysWithIncompleteAnalysis: exclude + // rows that are already exhausted to prevent re-entry to recovery. + sql`(${messagesTable.ai_moderation_flags} IS NULL OR ${messagesTable.ai_moderation_flags} NOT LIKE ${"%individual_analysis_exhausted%"})`, isNull(messagesTable.deleted_at), ), ) @@ -880,6 +889,7 @@ export async function getIncompleteMessagesByConversation( } } + // Message Reviews CRUD // ==================== diff --git a/services/frontend/src/features/messages/hooks/useMessages.ts b/services/frontend/src/features/messages/hooks/useMessages.ts index 80126ea..a40ee54 100644 --- a/services/frontend/src/features/messages/hooks/useMessages.ts +++ b/services/frontend/src/features/messages/hooks/useMessages.ts @@ -78,8 +78,13 @@ export function useMessages() { }, [cursor, loadingMore]); const reanalyze = useCallback(async (id: string): Promise => { - setMessages((prev) => - prev.map((message) => + // Capture prior state inside the functional updater so we don't need + // `messages` as a useCallback dependency (avoids stale closure churn). + let saved: MessageRecord | undefined; + + setMessages((prev) => { + saved = prev.find((m) => m.id === id); + return prev.map((message) => message.id === id ? { ...message, @@ -88,11 +93,24 @@ export function useMessages() { ai_analysis: null, } : message, - ), - ); - await reanalyzeMessage(id); + ); + }); + + try { + await reanalyzeMessage(id); + } catch (err) { + // HTTP failed — revert the optimistic update so the UI stays truthful. + if (saved) { + const snapshot = saved; + setMessages((prev) => + prev.map((message) => (message.id === id ? snapshot : message)), + ); + } + throw err; + } }, []); + const reanalyzeAllErrors = useCallback(async (): Promise => { // Optimistically mark all error messages as pending setMessages((prev) =>