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:
@@ -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();
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
|
|||||||
Reference in New Issue
Block a user