fix(ai-moderation): attachment-upload race dropped images before vision

Root cause (2nd layer after 50371bd): the analysis worker could pick up an
image message while its attachment upload was still in flight
(upload_status='pending'). downloadAndExtractFrame then fell back to the
Discord CDN URL (cdn.discordapp.com), which often 404s for old/purged links,
and 'if (!res.ok) return' silently dropped the image — no log, no vision
call, empty image map, and the LLM produced a text-only verdict like
'lampiran yang gagal terbaca oleh sistem'.

Fixes:
- ai-analysis-worker: skip targets whose attachment upload is still pending
  (both batch + individual paths) — they stay ai_status='pending' and the
  next 15s cycle analyzes them after the upload lands.
- mediaDownloader.downloadAndExtractFrame: try uploaded_url first, then
  discord_url as fallback; log non-OK responses (status + host) instead of
  silently returning; log when all candidate URLs fail.
This commit is contained in:
asepharyana
2026-08-11 09:44:34 +07:00
parent 50371bd2d1
commit 4f4c43555f
2 changed files with 152 additions and 85 deletions
@@ -287,18 +287,37 @@ async function processBatch(job: {
lines: contextLines.lines, lines: contextLines.lines,
}); });
const targetIds = messages.map((m) => m.id); const allTargetIds = messages.map((m) => m.id);
const contextIds = contextBefore.map((m) => m.id); const contextIds = contextBefore.map((m) => m.id);
const attachments = await messageStore.getAttachmentsForMessages([ const attachments = await messageStore.getAttachmentsForMessages([
...targetIds, ...allTargetIds,
...contextIds, ...contextIds,
]); ]);
// Attachment-upload race guard: a message whose attachment is still being
// uploaded (upload_status='pending') must not be analyzed yet. Its
// uploaded_url is not ready, and falling back to the Discord CDN link often
// 404s (expired/purged) — which used to silently produce a text-only
// verdict ("lampiran yang gagal terbaca"). Leave those targets pending; the
// next worker cycle picks them up after the upload lands.
const pendingUploadTargetIds = new Set(
(attachments ?? [])
.filter((a) => a.upload_status === "pending")
.map((a) => a.message_id),
);
const readyMessages =
pendingUploadTargetIds.size === 0
? messages
: messages.filter((m) => !pendingUploadTargetIds.has(m.id));
if (readyMessages.length === 0) {
return { ok: true, conversationKey, rows: [] };
}
// The orchestrator handles text/media split + caching + parallel paths // The orchestrator handles text/media split + caching + parallel paths
// internally, so a 20-message batch = 1 text LLM call (+1 media call // internally, so a 20-message batch = 1 text LLM call (+1 media call
// when media is present), not N per-message calls. // when media is present), not N per-message calls.
const moderationResult = await runModerationAnalysis({ const moderationResult = await runModerationAnalysis({
targets: messages, targets: readyMessages,
contextBlock, contextBlock,
attachments, attachments,
}); });
@@ -306,7 +325,7 @@ async function processBatch(job: {
const results = moderationResult.results.map((r) => const results = moderationResult.results.map((r) =>
normalizeResult( normalizeResult(
r as unknown as AnalysisResult, r as unknown as AnalysisResult,
messages.find((m) => m.id === r.messageId), readyMessages.find((m) => m.id === r.messageId),
), ),
); );
@@ -334,9 +353,10 @@ async function processBatch(job: {
logger.info( logger.info(
{ {
total: messages.length, total: readyMessages.length,
saved: allRows.length, saved: allRows.length,
conversationKey, conversationKey,
skippedPendingUpload: messages.length - readyMessages.length,
}, },
"LLM batch analysis complete", "LLM batch analysis complete",
); );
@@ -384,6 +404,17 @@ async function processIndividual(job: {
...contextIds, ...contextIds,
]); ]);
// Same attachment-upload race guard as the batch path: while the upload is
// still in-flight the uploaded_url is not ready and the Discord CDN fallback
// often 404s — analyzing now would silently produce a text-only verdict.
// Return no results so the message stays pending for the next cycle.
const uploadStillPending = (attachments ?? []).some(
(a) => a.message_id === message.id && a.upload_status === "pending",
);
if (uploadStillPending) {
return { ok: true, results: [] };
}
try { try {
const moderationResult = await runModerationAnalysis({ const moderationResult = await runModerationAnalysis({
targets: [message], targets: [message],
@@ -296,13 +296,37 @@ export async function downloadAndExtractFrame(
imageMap: Map<string, MessageImagePart[]>, imageMap: Map<string, MessageImagePart[]>,
): Promise<void> { ): Promise<void> {
const log = createChildLogger("mediaAnalysis"); const log = createChildLogger("mediaAnalysis");
const urlToUse = att.uploaded_url ?? att.discord_url ?? null; // Prefer the upload proxy (uploaded_url); the Discord CDN link can expire
if (!urlToUse) return; // or be purged (404), and a non-OK response used to silently drop the image
// from vision analysis (no log, empty image map → text-only verdict). Try
// each candidate URL in order and surface failures.
const urlCandidates = [
att.uploaded_url,
att.discord_url && att.discord_url !== att.uploaded_url
? att.discord_url
: null,
].filter((u): u is string => Boolean(u));
if (urlCandidates.length === 0) return;
let imageBytes: Buffer | null = null;
let lastStatus = 0;
let lastError: string | null = null;
for (const urlToUse of urlCandidates) {
const { controller, clear } = createAbortControllerWithTimeout(15000); const { controller, clear } = createAbortControllerWithTimeout(15000);
try { try {
const res = await fetch(urlToUse, { signal: controller.signal }); const res = await fetch(urlToUse, { signal: controller.signal });
if (!res.ok || !res.body) return; if (!res.ok || !res.body) {
lastStatus = res.status;
log.warn(
{
attachmentId: att.id,
urlHost: new URL(urlToUse).host,
status: res.status,
},
"Attachment fetch non-OK — trying next URL",
);
continue;
}
let totalBytes = 0; let totalBytes = 0;
const chunks: Uint8Array[] = []; const chunks: Uint8Array[] = [];
@@ -319,17 +343,40 @@ export async function downloadAndExtractFrame(
chunks.push(value); chunks.push(value);
} }
} }
const imageBytes = Buffer.concat(chunks); imageBytes = Buffer.concat(chunks);
break;
} catch (err) {
lastError = err instanceof Error ? err.message : String(err);
log.warn(
{
attachmentId: att.id,
urlHost: new URL(urlToUse).host,
error: lastError,
},
"Attachment download failed — trying next URL",
);
} finally {
clear();
}
}
if (!imageBytes) {
log.warn(
{
attachmentId: att.id,
filename: att.filename,
lastStatus,
lastError,
},
"All attachment URLs failed — skipping media analysis",
);
return;
}
const sniffedMime = sniffImageMimeType(imageBytes); const sniffedMime = sniffImageMimeType(imageBytes);
if (!sniffedMime && att.type.startsWith("video/")) { if (!sniffedMime && att.type.startsWith("video/")) {
await extractVideoFrames( await extractVideoFrames(att, imageBytes, targetId, maxDimension, imageMap);
att,
imageBytes,
targetId,
maxDimension,
imageMap,
);
return; return;
} }
@@ -380,17 +427,6 @@ export async function downloadAndExtractFrame(
image_url: { url: dataUrl }, image_url: { url: dataUrl },
sourceLabel: `[gambar di atas adalah attachment ${att.filename} dari pesan id=${att.message_id}]`, sourceLabel: `[gambar di atas adalah attachment ${att.filename} dari pesan id=${att.message_id}]`,
}); });
} catch (err) {
log.warn(
{
attachmentId: att.id,
error: err instanceof Error ? err.message : String(err),
},
"Download failed",
);
} finally {
clear();
}
} }
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------