diff --git a/services/discord-gateway/src/modules/ai-moderation/llmModerationClient.ts b/services/discord-gateway/src/modules/ai-moderation/llmModerationClient.ts index 0b23bf6..e22378d 100644 --- a/services/discord-gateway/src/modules/ai-moderation/llmModerationClient.ts +++ b/services/discord-gateway/src/modules/ai-moderation/llmModerationClient.ts @@ -29,16 +29,18 @@ import { buildStickerVisionPrompt, } from "./stickerPrompt.js"; import { + acquireMediaAnalysisLock, computeImagePhash, + deleteCachedMediaAnalysis, getCachedMediaAnalysis, getCachedMediaByPhash, - getCachedUserModeration, + getCachedTextModeration, getRecentCorrectedModerations, makeCustomEmojiCacheKey, makeImageCacheKey, makeStickerCacheKey, - makeUserModerationCacheKey, - setCachedUserModeration, + makeTextModerationCacheKey, + setCachedTextModeration, upsertCachedMediaAnalysis, upsertCachedMediaByPhash, } from "./textCacheStore.js"; @@ -583,6 +585,25 @@ const analyzeSingleMediaImage = async ( : buildGeneralImageVisionPrompt(image.sourceLabel, messageId); const visionPromise = (async (): Promise => { + // Attempt to acquire DISTRIBUTED lock + // Lock expires in 60 seconds (generous timeout for LLM) + const locked = await acquireMediaAnalysisLock(cacheKey, Date.now() + 60000); + + if (!locked) { + log.debug({ cacheKey }, "Media analysis distributed lock acquired by another pod. Polling..."); + // Poll DB for up to 30 seconds + for (let i = 0; i < 15; i++) { + await new Promise((resolve) => setTimeout(resolve, 2000)); + const pollCached = await getCachedMediaAnalysis(cacheKey); + if (pollCached) { + visionLruCache.set(cacheKey, pollCached); + return pollCached; + } + } + log.warn({ cacheKey }, "Polling for distributed media analysis timed out. Falling back."); + return FAILED_ANALYSIS_PREFIX; + } + // Layer 2: Perceptual hash pre-check (before expensive vision API call) let phash: string | null = null; if (image.image_url.url.startsWith("data:")) { @@ -680,6 +701,7 @@ const analyzeSingleMediaImage = async ( }, "Vision analysis failed after all retry attempts", ); + await deleteCachedMediaAnalysis(cacheKey).catch(() => {}); return FAILED_ANALYSIS_PREFIX; })(); @@ -1662,7 +1684,7 @@ export async function runModerationAnalysis( continue; } - const cacheKey = makeUserModerationCacheKey(target.user_id, rawContent); + const cacheKey = makeTextModerationCacheKey(rawContent); // Deduplicate: if two identical messages from same user in this batch, // skip the cache lookup for the second and reuse the first's result. if (seenCacheKeys.has(cacheKey)) { @@ -1681,7 +1703,7 @@ export async function runModerationAnalysis( seenCacheKeys.add(cacheKey); try { - const cached = await getCachedUserModeration(cacheKey); + const cached = await getCachedTextModeration(cacheKey); if (cached) { // Safety: skip cache entries that are artifacts of API/parse errors. // A previous bug cached error results as "flagged", causing 24h false positives. @@ -1792,8 +1814,8 @@ export async function runModerationAnalysis( // Caching a transient error would turn it into a 24h false positive. if (result.status === "error") continue; - const cacheKey = makeUserModerationCacheKey(target.user_id, rawContent); - setCachedUserModeration(cacheKey, { + const cacheKey = makeTextModerationCacheKey(rawContent); + setCachedTextModeration(cacheKey, { flags: result.flags ?? [], score: result.score ?? 0, analysis: result.analysis ?? "", diff --git a/services/discord-gateway/src/modules/ai-moderation/moderationPrompt.ts b/services/discord-gateway/src/modules/ai-moderation/moderationPrompt.ts index 5df6c11..ada854e 100644 --- a/services/discord-gateway/src/modules/ai-moderation/moderationPrompt.ts +++ b/services/discord-gateway/src/modules/ai-moderation/moderationPrompt.ts @@ -195,8 +195,8 @@ Input: [target] id=13131 user=nazi_babi_itu: bener tuh nih ras emang harus dibas Output: {"results":[{"message_id":"13131","status":"flagged","flags":["offensive_username","hate_speech","sara"],"score":0.9,"categories":["offensive_username","hate_speech","sara"],"severity":"high","confidence":0.95,"recommended_action":"delete","policy_version":"default-2026-05-30","evidence":["Username 'nazi_babi_itu' mengandung unsur SARA","bener tuh nih ras emang harus dibasmi"],"analysis":"nazi_babi_itu memiliki username SARA dan isi pesan memperkuat tone kebencian dengan ajakan kekerasan terhadap ras tertentu. Pelanggaran berat."}]} Contoh 13 — Obfuscation / Zalgo Text (Evasion): -Input: [target] id=14141 user=hater: ++++++K1[[ your $€/F" "~`| \0ve $ 1F ¥°U |}iE ®© -Output: {"results":[{"message_id":"14141","status":"flagged","flags":["harassment","hate_speech"],"score":0.95,"categories":["harassment","hate_speech"],"severity":"critical","confidence":0.95,"recommended_action":"delete","policy_version":"default-2026-05-30","evidence":["++++++K1[[ your $€/F","\\0ve $ 1F ¥°U |}iE"],"analysis":"Pesan menggunakan teknik obfuscation/simbol untuk menyembunyikan frasa 'Kill yourself I love if you die'. Ini adalah ancaman dan pelecehan berat yang disamarkan."}]} +Input: [target] id=14141 user=hater: ++++++K1[[ your $€/F" "~\`| \\0ve $ 1F ¥°U |}iE ®© +Output: {"results":[{"message_id":"14141","status":"flagged","flags":["harassment","hate_speech"],"score":0.95,"categories":["harassment","hate_speech"],"severity":"critical","confidence":0.95,"recommended_action":"delete","policy_version":"default-2026-05-30","evidence":["++++++K1[[ your $€/F","\\\\0ve $ 1F ¥°U |}iE"],"analysis":"Pesan menggunakan teknik obfuscation/simbol untuk menyembunyikan frasa 'Kill yourself I love if you die'. Ini adalah ancaman dan pelecehan berat yang disamarkan."}]} Contoh 14 — Vulgaritas Bahasa Asing / All-Caps: Input: [target] id=15151 user=troll: AKU RAJA TITTEN @@ -245,8 +245,8 @@ Input: [target] id=13131 user=nazi_babi_itu: bener tuh nih ras emang harus dibas Output: {"results":[{"message_id":"13131","status":"flagged","flags":["offensive_username","hate_speech","sara"],"score":0.9,"categories":["offensive_username","hate_speech","sara"],"severity":"high","confidence":0.95,"recommended_action":"delete","policy_version":"default-2026-05-30","evidence":["Username 'nazi_babi_itu' mengandung unsur SARA","bener tuh nih ras emang harus dibasmi"],"analysis":"nazi_babi_itu memiliki username SARA dan isi pesan memperkuat tone kebencian dengan ajakan kekerasan terhadap ras tertentu. Pelanggaran berat."}]} Contoh 13 — Obfuscation / Zalgo Text (Evasion): -Input: [target] id=14141 user=hater: ++++++K1[[ your $€/F" "~`| \0ve $ 1F ¥°U |}iE ®© -Output: {"results":[{"message_id":"14141","status":"flagged","flags":["harassment","hate_speech"],"score":0.95,"categories":["harassment","hate_speech"],"severity":"critical","confidence":0.95,"recommended_action":"delete","policy_version":"default-2026-05-30","evidence":["++++++K1[[ your $€/F","\\0ve $ 1F ¥°U |}iE"],"analysis":"Pesan menggunakan teknik obfuscation/simbol untuk menyembunyikan frasa 'Kill yourself I love if you die'. Ini adalah ancaman dan pelecehan berat yang disamarkan."}]} +Input: [target] id=14141 user=hater: ++++++K1[[ your $€/F" "~\`| \\0ve $ 1F ¥°U |}iE ®© +Output: {"results":[{"message_id":"14141","status":"flagged","flags":["harassment","hate_speech"],"score":0.95,"categories":["harassment","hate_speech"],"severity":"critical","confidence":0.95,"recommended_action":"delete","policy_version":"default-2026-05-30","evidence":["++++++K1[[ your $€/F","\\\\0ve $ 1F ¥°U |}iE"],"analysis":"Pesan menggunakan teknik obfuscation/simbol untuk menyembunyikan frasa 'Kill yourself I love if you die'. Ini adalah ancaman dan pelecehan berat yang disamarkan."}]} Contoh 14 — Vulgaritas Bahasa Asing / All-Caps: Input: [target] id=15151 user=troll: AKU RAJA TITTEN @@ -300,8 +300,8 @@ Input: [target] id=13131 user=nazi_babi_itu: bener tuh nih ras emang harus dibas Output: {"results":[{"message_id":"13131","status":"flagged","flags":["offensive_username","hate_speech","sara"],"score":0.9,"categories":["offensive_username","hate_speech","sara"],"severity":"high","confidence":0.95,"recommended_action":"delete","policy_version":"default-2026-05-30","evidence":["Username 'nazi_babi_itu' mengandung unsur SARA","bener tuh nih ras emang harus dibasmi"],"analysis":"nazi_babi_itu memiliki username SARA dan isi pesan memperkuat tone kebencian dengan ajakan kekerasan terhadap ras tertentu. Pelanggaran berat."}]} Contoh 13 — Obfuscation / Zalgo Text (Evasion): -Input: [target] id=14141 user=hater: ++++++K1[[ your $€/F" "~`| \0ve $ 1F ¥°U |}iE ®© -Output: {"results":[{"message_id":"14141","status":"flagged","flags":["harassment","hate_speech"],"score":0.95,"categories":["harassment","hate_speech"],"severity":"critical","confidence":0.95,"recommended_action":"delete","policy_version":"default-2026-05-30","evidence":["++++++K1[[ your $€/F","\\0ve $ 1F ¥°U |}iE"],"analysis":"Pesan menggunakan teknik obfuscation/simbol untuk menyembunyikan frasa 'Kill yourself I love if you die'. Ini adalah ancaman dan pelecehan berat yang disamarkan."}]} +Input: [target] id=14141 user=hater: ++++++K1[[ your $€/F" "~\`| \\0ve $ 1F ¥°U |}iE ®© +Output: {"results":[{"message_id":"14141","status":"flagged","flags":["harassment","hate_speech"],"score":0.95,"categories":["harassment","hate_speech"],"severity":"critical","confidence":0.95,"recommended_action":"delete","policy_version":"default-2026-05-30","evidence":["++++++K1[[ your $€/F","\\\\0ve $ 1F ¥°U |}iE"],"analysis":"Pesan menggunakan teknik obfuscation/simbol untuk menyembunyikan frasa 'Kill yourself I love if you die'. Ini adalah ancaman dan pelecehan berat yang disamarkan."}]} Contoh 14 — Vulgaritas Bahasa Asing / All-Caps: Input: [target] id=15151 user=troll: AKU RAJA TITTEN diff --git a/services/discord-gateway/src/modules/ai-moderation/textCacheStore.ts b/services/discord-gateway/src/modules/ai-moderation/textCacheStore.ts index 9807967..1ef6c5b 100644 --- a/services/discord-gateway/src/modules/ai-moderation/textCacheStore.ts +++ b/services/discord-gateway/src/modules/ai-moderation/textCacheStore.ts @@ -192,7 +192,7 @@ export async function getCachedMediaAnalysis( const row = await executeGet( `SELECT flags, hit_count FROM text_analysis_cache - WHERE text = $1 AND expires_at > $2`, + WHERE text = $1 AND expires_at > $2 AND source != 'vision_llm_processing'`, [cacheKey, Date.now()], ); @@ -240,6 +240,50 @@ export async function upsertCachedMediaAnalysis( } } +export async function acquireMediaAnalysisLock( + cacheKey: string, + expiresAt: number, +): Promise { + try { + const rows = await executeAll( + `INSERT INTO text_analysis_cache (text, flags, source, analyzed_at, expires_at, hit_count) + VALUES ($1, $2, $3, $4, $5, 0) + ON CONFLICT (text) DO UPDATE SET + flags = EXCLUDED.flags, + source = EXCLUDED.source, + analyzed_at = EXCLUDED.analyzed_at, + expires_at = EXCLUDED.expires_at + WHERE text_analysis_cache.expires_at < $4 + RETURNING text`, + [cacheKey, '""', "vision_llm_processing", Date.now(), expiresAt], + ); + return Array.isArray(rows) && rows.length > 0; + } catch (error) { + logger.error( + { error: error instanceof Error ? error.message : String(error) }, + "Failed to acquire media analysis lock", + ); + return false; + } +} + +export async function deleteCachedMediaAnalysis( + cacheKey: string, +): Promise { + try { + await executeAll( + `DELETE FROM text_analysis_cache + WHERE text = $1 AND source = 'vision_llm_processing'`, + [cacheKey], + ); + } catch (error) { + logger.error( + { error: error instanceof Error ? error.message : String(error) }, + "Failed to delete cached media analysis lock", + ); + } +} + // --------------------------------------------------------------------------- // Per-user moderation result cache (for spammer deduplication) // --------------------------------------------------------------------------- @@ -251,19 +295,16 @@ export async function upsertCachedMediaAnalysis( * Two users sending the same text get separate cache entries so that * per-user action history (e.g. repeated spam) can be tracked later. */ -export function makeUserModerationCacheKey( - userId: string, - content: string, -): string { +export function makeTextModerationCacheKey(content: string): string { const hash = createHash("sha256").update(content).digest("hex").slice(0, 16); - return `user_mod:${userId}:${hash}`; + return `text_mod:${hash}`; } /** - * Lookup a cached moderation result for a (user, content) pair. + * Lookup a cached moderation result for a text content. * Returns the stored result fields or null. */ -export async function getCachedUserModeration(cacheKey: string): Promise<{ +export async function getCachedTextModeration(cacheKey: string): Promise<{ status: "clean" | "flagged"; flags: string[]; score: number; @@ -317,7 +358,7 @@ export async function getCachedUserModeration(cacheKey: string): Promise<{ * Store a moderation result for a (user, content) pair. * The `flags` field stores the full result object as JSON. */ -export async function setCachedUserModeration( +export async function setCachedTextModeration( cacheKey: string, result: { flags: string[]; @@ -327,7 +368,7 @@ export async function setCachedUserModeration( severity: string; confidence: number; recommendedAction: string; - status?: "clean" | "warn" | "flagged"; + status?: "clean" | "warn" | "flagged" | "processing"; }, ): Promise { const now = Date.now(); diff --git a/services/discord-gateway/src/modules/message-capture/messageStore.ts b/services/discord-gateway/src/modules/message-capture/messageStore.ts index b8c78b1..0305559 100644 --- a/services/discord-gateway/src/modules/message-capture/messageStore.ts +++ b/services/discord-gateway/src/modules/message-capture/messageStore.ts @@ -41,6 +41,8 @@ interface QueryBuilder extends PromiseLike { onConflictDoNothing(...args: unknown[]): QueryBuilder; returning(...args: unknown[]): QueryBuilder; set(...args: unknown[]): QueryBuilder; + for(mode: string, options?: { skipLocked?: boolean }): QueryBuilder; + toSQL(): { sql: string; params: unknown[] }; } interface MessageDatabase { @@ -49,6 +51,7 @@ interface MessageDatabase { insert(...args: unknown[]): QueryBuilder; update(...args: unknown[]): QueryBuilder; transaction(callback: (tx: MessageDatabase) => Promise): Promise; + execute(sql: unknown): Promise; } function db(): MessageDatabase { @@ -408,7 +411,7 @@ export async function updateAttachmentAsFailedUpload( } interface AIAnalysisUpdate { - status: "pending" | "clean" | "warn" | "flagged" | "error"; + status: "pending" | "processing" | "clean" | "warn" | "flagged" | "error"; flags?: string | null; score?: number | null; analysis?: string | null; @@ -651,28 +654,39 @@ export async function getPendingMessagesByConversation( // conversationKey is either thread_id or channel_id // Query both to safely handle the key - const sq = database - .select({ id: messagesTable.id }) - .from(messagesTable) - .where( - and( - or( - eq(messagesTable.thread_id, conversationKey), - eq(messagesTable.channel_id, conversationKey), + const rows = await database.transaction(async (tx) => { + const pendingIdsQuery = tx + .select({ id: messagesTable.id }) + .from(messagesTable) + .where( + and( + or( + eq(messagesTable.thread_id, conversationKey), + eq(messagesTable.channel_id, conversationKey), + ), + eq(messagesTable.ai_status, "pending"), + isNull(messagesTable.deleted_at), ), - eq(messagesTable.ai_status, "pending"), - isNull(messagesTable.deleted_at), - ), - ) - .orderBy(asc(messagesTable.created_at)) - .limit(limit) - .for("update", { skipLocked: true }); + ) + .orderBy(asc(messagesTable.created_at)) + .limit(limit) + .for("update", { skipLocked: true }); - const rows = await database - .update(messagesTable) - .set({ ai_status: "processing", ai_analyzed_at: Date.now() }) - .where(inArray(messagesTable.id, sq)) - .returning(); + const pendingIds = await pendingIdsQuery; + + if (pendingIds.length === 0) return []; + + return await tx + .update(messagesTable) + .set({ ai_status: "processing", ai_analyzed_at: Date.now() }) + .where( + inArray( + messagesTable.id, + (pendingIds as any[]).map((r) => r.id as string), + ), + ) + .returning(); + }); return rows as MessageRecord[]; } catch (error) { @@ -863,32 +877,43 @@ export async function getIncompleteMessagesByConversation( ): Promise { try { const database = db(); - const sq = database - .select({ id: messagesTable.id }) - .from(messagesTable) - .where( - and( - or( - eq(messagesTable.thread_id, conversationKey), - eq(messagesTable.channel_id, conversationKey), + const rows = await database.transaction(async (tx) => { + const pendingIdsQuery = tx + .select({ id: messagesTable.id }) + .from(messagesTable) + .where( + and( + or( + eq(messagesTable.thread_id, conversationKey), + eq(messagesTable.channel_id, conversationKey), + ), + 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), ), - 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), - ), - ) - .orderBy(asc(messagesTable.created_at)) - .limit(limit) - .for("update", { skipLocked: true }); + ) + .orderBy(asc(messagesTable.created_at)) + .limit(limit) + .for("update", { skipLocked: true }); - const rows = await database - .update(messagesTable) - .set({ ai_status: "processing", ai_analyzed_at: Date.now() }) - .where(inArray(messagesTable.id, sq)) - .returning(); + const pendingIds = await pendingIdsQuery; + + if (pendingIds.length === 0) return []; + + return await tx + .update(messagesTable) + .set({ ai_status: "processing", ai_analyzed_at: Date.now() }) + .where( + inArray( + messagesTable.id, + (pendingIds as any[]).map((r) => r.id as string), + ), + ) + .returning(); + }); return rows as MessageRecord[]; } catch (error) { @@ -1280,14 +1305,14 @@ export async function revertStuckProcessingMessages( ) .returning({ id: messagesTable.id }); - if (rows.length > 0) { - logger.warn( - { count: rows.length, messageIds: rows.map((r) => r.id) }, - "Reverted stuck processing messages to pending", + if (Array.isArray(rows) && rows.length > 0) { + logger.info( + { count: rows.length, messageIds: rows.map((r: { id: string }) => r.id) }, + "Reverted stuck processing messages back to pending", ); } - - return rows.length; + + return Array.isArray(rows) ? rows.length : 0; } catch (error) { logger.error( { error: error instanceof Error ? error.message : String(error) },