refactor(ai-moderation): implement distributed locking and content-based caching

Refactors the AI moderation pipeline to improve concurrency control and
cache efficiency by moving from user-centric to content-centric caching.

- Implements a distributed locking mechanism for media analysis using
  `acquireMediaAnalysisLock` to prevent redundant LLM vision calls across
  multiple pods.
- Transitions text moderation caching from `user_mod:userId:hash` to a
  purely content-based `text_mod:hash` approach to increase hit rates.
- Enhances `getPendingMessagesByConversation` with atomic transactions
  and `FOR UPDATE SKIP LOCKED` to safely transition messages from
  `pending` to `processing` state.
- Adds `processing` status to the `AIStatus` type and database schema to
  track active analysis lifecycles.
- Implements polling logic in `llmModerationClient.ts` to wait for
  in-progress media analyses.

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,
+  makeTextModerationCacheKey,
-  setCachedUserModeration,
+  setCachedTextModeration,
   upsertCachedMediaAnalysis,
   upsertCachedMediaByPhash,
 } from "./textCacheStore.js";
@@ -583,6 +585,25 @@ const analyzeSingleMediaImage = async (
       : buildGeneralImageVisionPrompt(image.sourceLabel, messageId);
 
   const visionPromise = (async (): Promise<string> => {
+    // 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);
+    const cacheKey = makeTextModerationCacheKey(rawContent);
-    setCachedUserModeration(cacheKey, {
+    setCachedTextModeration(cacheKey, {
       flags: result.flags ?? [],
       score: result.score ?? 0,
       analysis: result.analysis ?? "",

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 ®©
+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."}]}
+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 ®©
+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."}]}
+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 ®©
+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."}]}
+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

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<boolean> {
+  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<void> {
+  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(
+export function makeTextModerationCacheKey(content: string): string {
-  userId: string,
-  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<void> {
   const now = Date.now();

services/discord-gateway/src/modules/message-capture/messageStore.ts

@@ -41,6 +41,8 @@ interface QueryBuilder<T = unknown> extends PromiseLike<T> {
   onConflictDoNothing(...args: unknown[]): QueryBuilder<T>;
   returning(...args: unknown[]): QueryBuilder<T>;
   set(...args: unknown[]): QueryBuilder<T>;
+  for(mode: string, options?: { skipLocked?: boolean }): QueryBuilder<T>;
+  toSQL(): { sql: string; params: unknown[] };
 }
 
 interface MessageDatabase {
@@ -49,6 +51,7 @@ interface MessageDatabase {
   insert<T = unknown>(...args: unknown[]): QueryBuilder<T>;
   update(...args: unknown[]): QueryBuilder<unknown>;
   transaction<T>(callback: (tx: MessageDatabase) => Promise<T>): Promise<T>;
+  execute(sql: unknown): Promise<any>;
 }
 
 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
+    const rows = await database.transaction(async (tx) => {
-      .select({ id: messagesTable.id })
+      const pendingIdsQuery = tx
-      .from(messagesTable)
+        .select({ id: messagesTable.id })
-      .where(
+        .from(messagesTable)
-        and(
+        .where(
-          or(
+          and(
-            eq(messagesTable.thread_id, conversationKey),
+            or(
-            eq(messagesTable.channel_id, conversationKey),
+              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
+      const pendingIds = await pendingIdsQuery;
-      .update(messagesTable)
+
-      .set({ ai_status: "processing", ai_analyzed_at: Date.now() })
+      if (pendingIds.length === 0) return [];
-      .where(inArray(messagesTable.id, sq))
+
-      .returning();
+      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<MessageRecord[]> {
   try {
     const database = db();
-    const sq = database
+    const rows = await database.transaction(async (tx) => {
-      .select({ id: messagesTable.id })
+      const pendingIdsQuery = tx
-      .from(messagesTable)
+        .select({ id: messagesTable.id })
-      .where(
+        .from(messagesTable)
-        and(
+        .where(
-          or(
+          and(
-            eq(messagesTable.thread_id, conversationKey),
+            or(
-            eq(messagesTable.channel_id, conversationKey),
+              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%"}`,
+        .orderBy(asc(messagesTable.created_at))
-          // Same guard as getConversationKeysWithIncompleteAnalysis: exclude
+        .limit(limit)
-          // rows that are already exhausted to prevent re-entry to recovery.
+        .for("update", { skipLocked: true });
-          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 });
 
-    const rows = await database
+      const pendingIds = await pendingIdsQuery;
-      .update(messagesTable)
+
-      .set({ ai_status: "processing", ai_analyzed_at: Date.now() })
+      if (pendingIds.length === 0) return [];
-      .where(inArray(messagesTable.id, sq))
+
-      .returning();
+      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) {
+    if (Array.isArray(rows) && rows.length > 0) {
-      logger.warn(
+      logger.info(
-        { count: rows.length, messageIds: rows.map((r) => r.id) },
+        { count: rows.length, messageIds: rows.map((r: { id: string }) => r.id) },
-        "Reverted stuck processing messages to pending",
+        "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) },