MythEclipse/GMW
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(moderation): enhance attachment handling and AI analysis integration

Browse Source
This commit is contained in:
MythEclipse
2026-05-21 02:39:28 +07:00
parent 3d64228d6a
commit 7eb01606b7
13 changed files with 447 additions and 223 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/moderation/aiAnalyzer.ts
+17 -5
View File
@@ -25,10 +25,12 @@ function getModerationBroadcaster(): ModerationBroadcaster | undefined {
// Debounce state per conversation key
const conversationDebounceTimers = new Map<string, NodeJS.Timeout>();
// Track conversations currently being processed
const conversationProcessing = new Set<string>();
const conversationProcessing = new Map<string, number>();
// Track conversations in error cooldown (failed recently)
const conversationErrorCooldown = new Map<string, number>();
const AI_PROCESSING_OVERLAP_MS = 30000;
let activeRequests = 0;
let lastError: string | null = null;
@@ -72,6 +74,13 @@ export function pickBatchWithinBudget(
return batch;
}
function isConversationProcessingLocked(conversationKey: string): boolean {
const startedAt = conversationProcessing.get(conversationKey);
return Boolean(
startedAt && Date.now() - startedAt < AI_PROCESSING_OVERLAP_MS,
);
}
/**
* Processes a batch of messages for a conversation
*/
@@ -82,7 +91,8 @@ async function processBatch(
if (messages.length === 0) return;
activeRequests++;
conversationProcessing.add(conversationKey);
const processingStartedAt = Date.now();
conversationProcessing.set(conversationKey, processingStartedAt);
try {
const result = await runAnalysisInWorker(conversationKey, messages);
@@ -136,7 +146,9 @@ async function processBatch(
);
} finally {
activeRequests--;
conversationProcessing.delete(conversationKey);
if (conversationProcessing.get(conversationKey) === processingStartedAt) {
conversationProcessing.delete(conversationKey);
}
}
}
@@ -171,7 +183,7 @@ async function runAnalysisInWorker(
*/
function scheduleConversationAnalysis(conversationKey: string): void {
// Skip if already processing
if (conversationProcessing.has(conversationKey)) {
if (isConversationProcessingLocked(conversationKey)) {
return;
}
@@ -275,7 +287,7 @@ export function startPendingAIAnalysisWorker(): void {
}
// Skip if currently processing
if (conversationProcessing.has(key)) {
if (isConversationProcessingLocked(key)) {
continue;
}
src/moderation/attachmentUploader.ts
+40 -2
View File
@@ -4,14 +4,34 @@ import { uploadToTele } from "../uploader/teleUpload";
import {
updateAttachmentAsFailedUpload,
updateAttachmentAsUploaded,
updateAttachmentDiscordUrl,
} from "./messageStore";
const logger = createChildLogger("attachment-uploader");
class AttachmentDownloadError extends Error {
constructor(
message: string,
readonly status: number,
) {
super(message);
this.name = "AttachmentDownloadError";
}
}
export type RefreshDiscordAttachmentUrl = () => Promise<string | null>;
function toErrorMessage(error: unknown): string {
return error instanceof Error ? error.message : String(error);
}
function shouldRefreshDiscordUrl(error: unknown): boolean {
return (
error instanceof AttachmentDownloadError &&
(error.status === 403 || error.status === 404)
);
}
export async function uploadAttachmentToTele(
fileBuffer: Buffer,
filename: string,
@@ -47,7 +67,10 @@ export async function downloadDiscordAttachment(url: string): Promise<Buffer> {
});
if (!response.ok) {
throw new Error(`Download failed with status ${response.status}`);
throw new AttachmentDownloadError(
`Download failed with status ${response.status}`,
response.status,
);
}
const buffer = await response.arrayBuffer();
@@ -65,9 +88,24 @@ export async function processAttachmentUpload(
attachmentId: string,
discordUrl: string,
filename: string,
options: { refreshDiscordUrl?: RefreshDiscordAttachmentUrl } = {},
): Promise<void> {
try {
const buffer = await downloadDiscordAttachment(discordUrl);
let currentDiscordUrl = discordUrl;
let buffer: Buffer;
try {
buffer = await downloadDiscordAttachment(currentDiscordUrl);
} catch (error) {
if (!options.refreshDiscordUrl || !shouldRefreshDiscordUrl(error)) {
throw error;
}
const freshUrl = await options.refreshDiscordUrl();
if (!freshUrl) throw error;
currentDiscordUrl = freshUrl;
await updateAttachmentDiscordUrl(attachmentId, freshUrl);
buffer = await downloadDiscordAttachment(currentDiscordUrl);
}
const sizeMb = buffer.length / (1024 * 1024);
if (sizeMb > config.ATTACHMENT_MAX_SIZE_MB) {
src/moderation/llmModerationClient.ts
+137 -138
View File
@@ -1,9 +1,30 @@
import OpenAI from "openai";
import { config } from "../config.ts";
import { createChildLogger } from "../logger.ts";
import { retryWithBackoff } from "../retry.ts";
import type { AnalysisResult, AttachmentRecord, MessageRecord } from "./types";
const log = createChildLogger("llmModerationClient");
const openai = new OpenAI({
apiKey: config.AI_LLM_API_KEY,
baseURL: config.AI_LLM_BASE_URL,
maxRetries: 0,
timeout: 2_147_483_647,
fetch: async (url, init) => {
const response = await globalThis.fetch(url, init);
if (response.headers) return response;
const body =
typeof response.text === "function"
? await response.text()
: JSON.stringify(await response.json());
return new Response(body, {
status: response.status ?? 200,
headers: { "Content-Type": "application/json" },
});
},
});
interface RawModerationResult {
message_id: string;
@@ -17,48 +38,6 @@ interface RawModerationResponse {
results: RawModerationResult[];
}
function parseFirstJsonObject(content: string): unknown {
for (
let start = content.indexOf("{");
start !== -1;
start = content.indexOf("{", start + 1)
) {
let depth = 0;
let inString = false;
let escaped = false;
for (let index = start; index < content.length; index++) {
const char = content[index];
if (escaped) {
escaped = false;
continue;
}
if (char === "\\") {
escaped = inString;
continue;
}
if (char === '"') {
inString = !inString;
continue;
}
if (inString) continue;
if (char === "{") depth++;
if (char === "}") depth--;
if (depth === 0) {
return JSON.parse(content.slice(start, index + 1));
}
}
}
throw new Error("No JSON object found in response body");
}
/**
* Helper to extract a JSON object from a potentially conversational or markdown-wrapped string.
* It first scans for markdown json code blocks, then falls back to trying all start/end brace pairs from largest to smallest.
@@ -113,6 +92,49 @@ export function extractJson(content: string): any {
* Extracts JSON from surrounding text, validates structure, and transforms to AnalysisResult[].
* Scans from first '{' and attempts JSON.parse at each candidate closing brace.
*/
function salvageMalformedModerationResponse(
content: string,
targetIds: string[],
): AnalysisResult[] | null {
const idMatches = content.match(/\d{10,22}/g) ?? [];
let matchedId: string | null = null;
for (const targetId of targetIds) {
if (content.includes(targetId)) {
matchedId = targetId;
break;
}
}
if (!matchedId) {
for (const candidate of idMatches) {
matchedId =
targetIds.find(
(targetId) =>
targetId.startsWith(candidate) || candidate.startsWith(targetId),
) ?? null;
if (matchedId) break;
}
}
if (!matchedId) return null;
const statusMatch = content.match(/"status"\s*:\s*"(clean|warn|flagged)"/);
const scoreMatch = content.match(/"score"\s*:\s*(\d+(?:\.\d+)?)/);
const analysisMatch = content.match(/"analysis"\s*:\s*"([^"]*)"/);
return [
{
messageId: matchedId,
status: (statusMatch?.[1] as "clean" | "warn" | "flagged") ?? "clean",
flags: [],
score: scoreMatch ? Math.max(0, Math.min(1, Number(scoreMatch[1]))) : 0,
analysis:
analysisMatch?.[1] ?? "Recovered from malformed moderation response",
},
];
}
export function parseModerationResponse(
content: string,
targetIds: string[],
@@ -260,9 +282,11 @@ export function parseModerationResponse(
}
if (!targetIdSet.has(finalId)) {
throw new Error(
`Unknown message_id: ${finalId} (original: ${message_id})`,
log.warn(
{ unknownId: finalId, originalId: message_id, targetIds },
"Skipping moderation result for non-target message_id",
);
return null;
}
if (foundIds.has(finalId)) {
@@ -322,12 +346,11 @@ export function parseModerationResponse(
{ missingIds, foundCount: foundIds.size, totalCount: targetIds.length },
"Some target IDs missing in response - marking as incomplete",
);
// Add clean results for missing IDs instead of failing the batch
for (const missingId of missingIds) {
filteredResults.push({
messageId: missingId,
status: "clean",
flags: [],
status: "error",
flags: ["analysis_incomplete"],
score: 0,
analysis: "Analysis incomplete - LLM did not process this message",
});
@@ -391,7 +414,6 @@ ${messagesText}`;
const targetIdSet = new Set(targets.map((t) => t.id));
const getAttachmentImageUrl = (att: AttachmentRecord): string | null => {
if (att.uploaded_url) return att.uploaded_url;
if (targetIdSet.has(att.message_id)) return att.discord_url;
return null;
};
const imageAttachments = (attachments || [])
@@ -471,79 +493,27 @@ ${messagesText}`;
}
const result = await retryWithBackoff(
async () => {
const controller = new AbortController();
const timeoutId = setTimeout(
() => controller.abort(),
config.AI_ANALYSIS_TIMEOUT_MS,
);
try {
const response = await fetch(
`${config.AI_LLM_BASE_URL}/chat/completions`,
() =>
openai.chat.completions.create({
model: config.AI_LLM_MODEL,
messages: [
{
method: "POST",
headers: {
"Content-Type": "application/json",
Authorization: `Bearer ${config.AI_LLM_API_KEY}`,
},
signal: controller.signal,
body: JSON.stringify({
model: config.AI_LLM_MODEL,
messages: [
{
role: "system",
content: systemPrompt,
},
{
role: "user",
content: messageContent,
},
],
temperature: 0,
top_p: 1,
max_tokens: 8192,
response_format: { type: "json_object" },
chat_template_kwargs: { enable_thinking: false },
}),
role: "system",
content: systemPrompt,
},
);
// Read the response body once (either text() or json()), then reuse it.
let rawBody: string | undefined = undefined;
if (typeof response.text === "function") {
try {
rawBody = await response.text();
} catch {
rawBody = undefined;
}
} else if (typeof response.json === "function") {
try {
const j = await response.json();
rawBody = JSON.stringify(j);
} catch {
rawBody = undefined;
}
}
if (!response.ok) {
throw new Error(
`LLM API error ${response.status}: ${rawBody ?? "(no body)"}`,
);
}
if (!rawBody) {
throw new Error("Empty LLM response");
}
try {
return JSON.parse(rawBody);
} catch {
return parseFirstJsonObject(rawBody);
}
} finally {
clearTimeout(timeoutId);
}
},
{
role: "user",
content: messageContent,
},
],
temperature: 0.2,
top_p: 0.95,
max_tokens: 65536,
response_format: { type: "json_object" },
stream: false,
chat_template_kwargs: { enable_thinking: false },
reasoning_budget: 0,
} as OpenAI.Chat.Completions.ChatCompletionCreateParamsNonStreaming),
{
retries: 3,
minTimeout: 1000,
@@ -569,25 +539,54 @@ ${messagesText}`;
} catch (parseError) {
const errorMsg =
parseError instanceof Error ? parseError.message : String(parseError);
log.error(
{
error: errorMsg,
contentLength: content.length,
contentPreview: content.substring(0, 500),
fullContent: content,
targetIds,
model: config.AI_LLM_MODEL,
timestamp: new Date().toISOString(),
},
"Robust Fallback: Failed to parse moderation response. Defaulting all targets to clean.",
);
parsed = targetIds.map((id) => ({
messageId: id,
status: "clean",
flags: [],
score: 0.1,
analysis: `Parsing failed: ${errorMsg}. Defaulted to clean.`,
}));
const salvaged = salvageMalformedModerationResponse(content, targetIds);
if (salvaged) {
log.warn(
{
error: errorMsg,
contentLength: content.length,
contentPreview: content.substring(0, 500),
targetIds,
recoveredIds: salvaged.map((result) => result.messageId),
model: config.AI_LLM_MODEL,
timestamp: new Date().toISOString(),
},
"Recovered moderation response from malformed JSON",
);
const recoveredIds = new Set(salvaged.map((result) => result.messageId));
parsed = [
...salvaged,
...targetIds
.filter((id) => !recoveredIds.has(id))
.map((id) => ({
messageId: id,
status: "error" as const,
flags: ["analysis_incomplete"],
score: 0,
analysis: "Analysis incomplete - malformed LLM response",
})),
];
} else {
log.error(
{
error: errorMsg,
contentLength: content.length,
contentPreview: content.substring(0, 500),
fullContent: content,
targetIds,
model: config.AI_LLM_MODEL,
timestamp: new Date().toISOString(),
},
"Robust Fallback: Failed to parse moderation response. Defaulting all targets to clean.",
);
parsed = targetIds.map((id) => ({
messageId: id,
status: "error",
flags: ["analysis_parse_failed"],
score: 0,
analysis: `Parsing failed: ${errorMsg}.`,
}));
}
}
log.info(
src/moderation/messageCapture.ts
+38 -12
View File
@@ -121,6 +121,8 @@ export async function captureMessage(
broadcaster.messageCreated(messageRecord);
}
const attachmentUploadTasks: Promise<void>[] = [];
// Insert attachments before queuing analysis to avoid race condition
if (message.attachments.size > 0) {
for (const [, attachment] of message.attachments) {
@@ -136,16 +138,29 @@ export async function captureMessage(
// Initiate async upload (non-blocking, fire-and-forget)
if (!isBacklog) {
processAttachmentUpload(
attachment.id,
attachment.url,
attachment.name || "unknown",
).catch((err) => {
logger.error(
{ attachmentId: attachment.id, error: err },
"Failed to initiate attachment upload",
);
});
attachmentUploadTasks.push(
processAttachmentUpload(
attachment.id,
attachment.url,
attachment.name || "unknown",
{
refreshDiscordUrl: async () => {
const freshMessage = await message.channel.messages.fetch(
message.id,
);
const freshAttachment = freshMessage.attachments.get(
attachment.id,
);
return freshAttachment?.url ?? null;
},
},
).catch((err) => {
logger.error(
{ attachmentId: attachment.id, error: err },
"Failed to initiate attachment upload",
);
}),
);
}
if (broadcaster) {
@@ -154,9 +169,20 @@ export async function captureMessage(
}
}
// Queue analysis after attachments are inserted
// Queue analysis after attachment uploads settle so AI uses stable tele URLs.
if (!isBacklog) {
queueMessageAnalysis(message.id);
if (attachmentUploadTasks.length > 0) {
Promise.allSettled(attachmentUploadTasks)
.then(() => queueMessageAnalysis(message.id))
.catch((err) => {
logger.error(
{ messageId: message.id, error: err },
"Failed to queue message analysis after attachment upload",
);
});
} else {
queueMessageAnalysis(message.id);
}
}
}
src/moderation/messageStore.ts
+22
View File
@@ -325,6 +325,28 @@ export async function updateAttachmentAsUploaded(
}
}
export async function updateAttachmentDiscordUrl(
attachmentId: string,
discordUrl: string,
): Promise<void> {
try {
const database = db();
await database
.update(attachmentsTable)
.set({ discord_url: discordUrl })
.where(eq(attachmentsTable.id, attachmentId));
} catch (error) {
logger.error(
{
attachmentId,
error: error instanceof Error ? error.message : String(error),
},
"Failed to update attachment Discord URL",
);
throw error;
}
}
export async function updateAttachmentAsFailedUpload(
attachmentId: string,
error: string,
src/moderation/types.ts
+1 -1
View File
@@ -86,7 +86,7 @@ export interface PageResult<T> {
export interface AnalysisResult {
messageId: string;
status: Exclude<AIStatus, "pending" | "error">;
status: Exclude<AIStatus, "pending">;
flags: string[];
score: number;
analysis: string;