diff --git a/package.json b/package.json index 2e83cec..378c066 100644 --- a/package.json +++ b/package.json @@ -48,6 +48,7 @@ "opusscript": "^0.0.8", "p-retry": "^8.0.0", "pg": "^8.21.0", + "piscina": "^5.1.4", "play-dl": "^1.9.7", "prism-media": "2.0.0-alpha.0", "prom-client": "^15.1.3", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 105e06e..f813180 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -74,6 +74,9 @@ importers: pg: specifier: ^8.21.0 version: 8.21.0 + piscina: + specifier: ^5.1.4 + version: 5.1.4 play-dl: specifier: ^1.9.7 version: 1.9.7 @@ -1130,6 +1133,119 @@ packages: '@minhducsun2002/leb128@1.0.0': resolution: {integrity: sha512-eFrYUPDVHeuwWHluTG1kwNQUEUcFjVKYwPkU8z9DR1JH3AW7JtJsG9cRVGmwz809kKtGfwGJj58juCZxEvnI/g==} + '@napi-rs/nice-android-arm-eabi@1.1.1': + resolution: {integrity: sha512-kjirL3N6TnRPv5iuHw36wnucNqXAO46dzK9oPb0wj076R5Xm8PfUVA9nAFB5ZNMmfJQJVKACAPd/Z2KYMppthw==} + engines: {node: '>= 10'} + cpu: [arm] + os: [android] + + '@napi-rs/nice-android-arm64@1.1.1': + resolution: {integrity: sha512-blG0i7dXgbInN5urONoUCNf+DUEAavRffrO7fZSeoRMJc5qD+BJeNcpr54msPF6qfDD6kzs9AQJogZvT2KD5nw==} + engines: {node: '>= 10'} + cpu: [arm64] + os: [android] + + '@napi-rs/nice-darwin-arm64@1.1.1': + resolution: {integrity: sha512-s/E7w45NaLqTGuOjC2p96pct4jRfo61xb9bU1unM/MJ/RFkKlJyJDx7OJI/O0ll/hrfpqKopuAFDV8yo0hfT7A==} + engines: {node: '>= 10'} + cpu: [arm64] + os: [darwin] + + '@napi-rs/nice-darwin-x64@1.1.1': + resolution: {integrity: sha512-dGoEBnVpsdcC+oHHmW1LRK5eiyzLwdgNQq3BmZIav+9/5WTZwBYX7r5ZkQC07Nxd3KHOCkgbHSh4wPkH1N1LiQ==} + engines: {node: '>= 10'} + cpu: [x64] + os: [darwin] + + '@napi-rs/nice-freebsd-x64@1.1.1': + resolution: {integrity: sha512-kHv4kEHAylMYmlNwcQcDtXjklYp4FCf0b05E+0h6nDHsZ+F0bDe04U/tXNOqrx5CmIAth4vwfkjjUmp4c4JktQ==} + engines: {node: '>= 10'} + cpu: [x64] + os: [freebsd] + + '@napi-rs/nice-linux-arm-gnueabihf@1.1.1': + resolution: {integrity: sha512-E1t7K0efyKXZDoZg1LzCOLxgolxV58HCkaEkEvIYQx12ht2pa8hoBo+4OB3qh7e+QiBlp1SRf+voWUZFxyhyqg==} + engines: {node: '>= 10'} + cpu: [arm] + os: [linux] + + '@napi-rs/nice-linux-arm64-gnu@1.1.1': + resolution: {integrity: sha512-CIKLA12DTIZlmTaaKhQP88R3Xao+gyJxNWEn04wZwC2wmRapNnxCUZkVwggInMJvtVElA+D4ZzOU5sX4jV+SmQ==} + engines: {node: '>= 10'} + cpu: [arm64] + os: [linux] + libc: [glibc] + + '@napi-rs/nice-linux-arm64-musl@1.1.1': + resolution: {integrity: sha512-+2Rzdb3nTIYZ0YJF43qf2twhqOCkiSrHx2Pg6DJaCPYhhaxbLcdlV8hCRMHghQ+EtZQWGNcS2xF4KxBhSGeutg==} + engines: {node: '>= 10'} + cpu: [arm64] + os: [linux] + libc: [musl] + + '@napi-rs/nice-linux-ppc64-gnu@1.1.1': + resolution: {integrity: sha512-4FS8oc0GeHpwvv4tKciKkw3Y4jKsL7FRhaOeiPei0X9T4Jd619wHNe4xCLmN2EMgZoeGg+Q7GY7BsvwKpL22Tg==} + engines: {node: '>= 10'} + cpu: [ppc64] + os: [linux] + libc: [glibc] + + '@napi-rs/nice-linux-riscv64-gnu@1.1.1': + resolution: {integrity: sha512-HU0nw9uD4FO/oGCCk409tCi5IzIZpH2agE6nN4fqpwVlCn5BOq0MS1dXGjXaG17JaAvrlpV5ZeyZwSon10XOXw==} + engines: {node: '>= 10'} + cpu: [riscv64] + os: [linux] + libc: [glibc] + + '@napi-rs/nice-linux-s390x-gnu@1.1.1': + resolution: {integrity: sha512-2YqKJWWl24EwrX0DzCQgPLKQBxYDdBxOHot1KWEq7aY2uYeX+Uvtv4I8xFVVygJDgf6/92h9N3Y43WPx8+PAgQ==} + engines: {node: '>= 10'} + cpu: [s390x] + os: [linux] + libc: [glibc] + + '@napi-rs/nice-linux-x64-gnu@1.1.1': + resolution: {integrity: sha512-/gaNz3R92t+dcrfCw/96pDopcmec7oCcAQ3l/M+Zxr82KT4DljD37CpgrnXV+pJC263JkW572pdbP3hP+KjcIg==} + engines: {node: '>= 10'} + cpu: [x64] + os: [linux] + libc: [glibc] + + '@napi-rs/nice-linux-x64-musl@1.1.1': + resolution: {integrity: sha512-xScCGnyj/oppsNPMnevsBe3pvNaoK7FGvMjT35riz9YdhB2WtTG47ZlbxtOLpjeO9SqqQ2J2igCmz6IJOD5JYw==} + engines: {node: '>= 10'} + cpu: [x64] + os: [linux] + libc: [musl] + + '@napi-rs/nice-openharmony-arm64@1.1.1': + resolution: {integrity: sha512-6uJPRVwVCLDeoOaNyeiW0gp2kFIM4r7PL2MczdZQHkFi9gVlgm+Vn+V6nTWRcu856mJ2WjYJiumEajfSm7arPQ==} + engines: {node: '>= 10'} + cpu: [arm64] + os: [openharmony] + + '@napi-rs/nice-win32-arm64-msvc@1.1.1': + resolution: {integrity: sha512-uoTb4eAvM5B2aj/z8j+Nv8OttPf2m+HVx3UjA5jcFxASvNhQriyCQF1OB1lHL43ZhW+VwZlgvjmP5qF3+59atA==} + engines: {node: '>= 10'} + cpu: [arm64] + os: [win32] + + '@napi-rs/nice-win32-ia32-msvc@1.1.1': + resolution: {integrity: sha512-CNQqlQT9MwuCsg1Vd/oKXiuH+TcsSPJmlAFc5frFyX/KkOh0UpBLEj7aoY656d5UKZQMQFP7vJNa1DNUNORvug==} + engines: {node: '>= 10'} + cpu: [ia32] + os: [win32] + + '@napi-rs/nice-win32-x64-msvc@1.1.1': + resolution: {integrity: sha512-vB+4G/jBQCAh0jelMTY3+kgFy00Hlx2f2/1zjMoH821IbplbWZOkLiTYXQkygNTzQJTq5cvwBDgn2ppHD+bglQ==} + engines: {node: '>= 10'} + cpu: [x64] + os: [win32] + + '@napi-rs/nice@1.1.1': + resolution: {integrity: sha512-xJIPs+bYuc9ASBl+cvGsKbGrJmS6fAKaSZCnT0lhahT5rhA2VVy9/EcIgd2JhtEuFOJNx7UHNn/qiTPTY4nrQw==} + engines: {node: '>= 10'} + '@napi-rs/wasm-runtime@1.1.4': resolution: {integrity: sha512-3NQNNgA1YSlJb/kMH1ildASP9HW7/7kYnRI2szWJaofaS1hWmbGI4H+d3+22aGzXXN9IJ+n+GiFVcGipJP18ow==} peerDependencies: @@ -3713,6 +3829,10 @@ packages: resolution: {integrity: sha512-QP88BAKvMam/3NxH6vj2o21R6MjxZUAd6nlwAS/pnGvN9IVLocLHxGYIzFhg6fUQ+5th6P4dv4eW9jX3DSIj7A==} engines: {node: '>=12'} + piscina@5.1.4: + resolution: {integrity: sha512-7uU4ZnKeQq22t9AsmHGD2w4OYQGonwFnTypDypaWi7Qr2EvQIFVtG8J5D/3bE7W123Wdc9+v4CZDu5hJXVCtBg==} + engines: {node: '>=20.x'} + pkg-pr-new@0.0.62: resolution: {integrity: sha512-K2jtf1PLCJJFDimQIpasPXDdnRTZSYPy36Ldz+QhMpLz2YN1Wi0ZQkomVt5Wi3NdBZcYuYGXekyFWfJ6fAHYjg==} hasBin: true @@ -5346,6 +5466,78 @@ snapshots: '@minhducsun2002/leb128@1.0.0': {} + '@napi-rs/nice-android-arm-eabi@1.1.1': + optional: true + + '@napi-rs/nice-android-arm64@1.1.1': + optional: true + + '@napi-rs/nice-darwin-arm64@1.1.1': + optional: true + + '@napi-rs/nice-darwin-x64@1.1.1': + optional: true + + '@napi-rs/nice-freebsd-x64@1.1.1': + optional: true + + '@napi-rs/nice-linux-arm-gnueabihf@1.1.1': + optional: true + + '@napi-rs/nice-linux-arm64-gnu@1.1.1': + optional: true + + '@napi-rs/nice-linux-arm64-musl@1.1.1': + optional: true + + '@napi-rs/nice-linux-ppc64-gnu@1.1.1': + optional: true + + '@napi-rs/nice-linux-riscv64-gnu@1.1.1': + optional: true + + '@napi-rs/nice-linux-s390x-gnu@1.1.1': + optional: true + + '@napi-rs/nice-linux-x64-gnu@1.1.1': + optional: true + + '@napi-rs/nice-linux-x64-musl@1.1.1': + optional: true + + '@napi-rs/nice-openharmony-arm64@1.1.1': + optional: true + + '@napi-rs/nice-win32-arm64-msvc@1.1.1': + optional: true + + '@napi-rs/nice-win32-ia32-msvc@1.1.1': + optional: true + + '@napi-rs/nice-win32-x64-msvc@1.1.1': + optional: true + + '@napi-rs/nice@1.1.1': + optionalDependencies: + '@napi-rs/nice-android-arm-eabi': 1.1.1 + '@napi-rs/nice-android-arm64': 1.1.1 + '@napi-rs/nice-darwin-arm64': 1.1.1 + '@napi-rs/nice-darwin-x64': 1.1.1 + '@napi-rs/nice-freebsd-x64': 1.1.1 + '@napi-rs/nice-linux-arm-gnueabihf': 1.1.1 + '@napi-rs/nice-linux-arm64-gnu': 1.1.1 + '@napi-rs/nice-linux-arm64-musl': 1.1.1 + '@napi-rs/nice-linux-ppc64-gnu': 1.1.1 + '@napi-rs/nice-linux-riscv64-gnu': 1.1.1 + '@napi-rs/nice-linux-s390x-gnu': 1.1.1 + '@napi-rs/nice-linux-x64-gnu': 1.1.1 + '@napi-rs/nice-linux-x64-musl': 1.1.1 + '@napi-rs/nice-openharmony-arm64': 1.1.1 + '@napi-rs/nice-win32-arm64-msvc': 1.1.1 + '@napi-rs/nice-win32-ia32-msvc': 1.1.1 + '@napi-rs/nice-win32-x64-msvc': 1.1.1 + optional: true + '@napi-rs/wasm-runtime@1.1.4(@emnapi/core@1.10.0)(@emnapi/runtime@1.10.0)': dependencies: '@emnapi/core': 1.10.0 @@ -7851,6 +8043,10 @@ snapshots: picomatch@4.0.4: {} + piscina@5.1.4: + optionalDependencies: + '@napi-rs/nice': 1.1.1 + pkg-pr-new@0.0.62: dependencies: '@actions/core': 1.11.1 diff --git a/src/moderation/aiAnalysisWorker.ts b/src/moderation/aiAnalysisWorker.ts index fb806c2..543ebc4 100644 --- a/src/moderation/aiAnalysisWorker.ts +++ b/src/moderation/aiAnalysisWorker.ts @@ -1,4 +1,3 @@ -import { parentPort } from "node:worker_threads"; import { config } from "../config.js"; import { initializeDatabase } from "../database/drizzle.js"; import { buildConversationPromptMessages } from "./conversationContext.js"; @@ -6,18 +5,29 @@ import { runModerationAnalysis } from "./llmModerationClient.js"; import { getAttachmentsForMessages, getConversationContextBefore, - updateMessageAIAnalysis, + updateMessagesAIAnalysisBulk, } from "./messageStore.js"; import type { MessageRecord } from "./types.js"; let dbInitialized = false; +let dbInitPromise: Promise | null = null; -interface AnalysisWorkerRequest { +async function ensureDb() { + if (dbInitialized) return; + if (!dbInitPromise) { + dbInitPromise = initializeDatabase().then(() => { + dbInitialized = true; + }); + } + await dbInitPromise; +} + +export interface AnalysisWorkerRequest { conversationKey: string; messages: MessageRecord[]; } -type AnalysisWorkerResponse = +export type AnalysisWorkerResponse = | { ok: true; conversationKey: string; @@ -30,16 +40,13 @@ type AnalysisWorkerResponse = error: string; }; -async function processAnalysisRequest({ +export default async function processAnalysisRequest({ conversationKey, messages, }: AnalysisWorkerRequest): Promise { try { try { - if (!dbInitialized) { - await initializeDatabase(); - dbInitialized = true; - } + await ensureDb(); } catch (dbError) { const msg = dbError instanceof Error ? dbError.message : String(dbError); return { @@ -77,9 +84,9 @@ async function processAnalysisRequest({ attachments, }); - const rows: MessageRecord[] = []; - for (const analysisResult of result.results) { - const row = await updateMessageAIAnalysis(analysisResult.messageId, { + const updates = result.results.map((analysisResult) => ({ + messageId: analysisResult.messageId, + result: { status: analysisResult.status, flags: JSON.stringify(analysisResult.flags), score: analysisResult.score, @@ -87,9 +94,10 @@ async function processAnalysisRequest({ analysis: analysisResult.analysis, analyzedAt: Date.now(), error: null, - }); - if (row) rows.push(row); - } + } + })); + + const rows = await updateMessagesAIAnalysisBulk(updates); return { ok: true, conversationKey, rows }; } catch (error) { @@ -112,7 +120,3 @@ async function processAnalysisRequest({ return { ok: false, conversationKey, rows, error: errorMessage }; } } - -parentPort?.on("message", async (request: AnalysisWorkerRequest) => { - parentPort?.postMessage(await processAnalysisRequest(request)); -}); diff --git a/src/moderation/aiAnalyzer.ts b/src/moderation/aiAnalyzer.ts index 44bd476..74497ba 100644 --- a/src/moderation/aiAnalyzer.ts +++ b/src/moderation/aiAnalyzer.ts @@ -1,6 +1,6 @@ import { existsSync } from "node:fs"; import { fileURLToPath } from "node:url"; -import { Worker } from "node:worker_threads"; +import { Piscina } from "piscina"; import { config } from "../config.js"; import { createChildLogger } from "../logger.js"; import { @@ -36,6 +36,32 @@ const AI_PROCESSING_OVERLAP_MS = 30000; let activeRequests = 0; let lastError: string | null = null; +// Global circuit breaker state +let consecutiveErrors = 0; +const MAX_CONSECUTIVE_ERRORS = 5; +let globalCooldownUntil = 0; + +function getAnalysisWorkerUrl(): URL { + const candidates = [ + new URL("./aiAnalysisWorker.js", import.meta.url), + new URL("../aiAnalysisWorker.js", import.meta.url), + new URL("./aiAnalysisWorker.ts", import.meta.url), + ]; + + for (const candidate of candidates) { + if (existsSync(fileURLToPath(candidate))) { + return candidate; + } + } + + return candidates[2]; +} + +const workerPool = new Piscina({ + filename: fileURLToPath(getAnalysisWorkerUrl()), + execArgv: process.execArgv, +}); + interface AnalysisWorkerResponse { ok: boolean; conversationKey: string; @@ -62,9 +88,9 @@ export function pickBatchWithinBudget( let usedTokens = 0; for (const msg of messages) { - // Estimate tokens based on actual content length + // Estimate tokens based on actual content length (conservative: 3 chars/token) const content = msg.edited_content ?? msg.content; - const contentTokens = Math.ceil(content.length / 4); + const contentTokens = Math.ceil(content.length / 3); const msgTokens = contentTokens + tokensPerMessage; if (usedTokens + msgTokens <= maxTokens) { @@ -91,19 +117,28 @@ async function processBatch( messages: MessageRecord[], ): Promise { if (messages.length === 0) return; + if (Date.now() < globalCooldownUntil) { + return; // Circuit breaker is open + } activeRequests++; let shouldScheduleNext = false; const processingStartedAt = Date.now(); conversationProcessing.set(conversationKey, processingStartedAt); try { - const result = await runAnalysisInWorker(conversationKey, messages); + const result = (await workerPool.run({ conversationKey, messages })) as AnalysisWorkerResponse; for (const row of result.rows) { getModerationBroadcaster()?.messageAnalyzed(row); } if (!result.ok) { + consecutiveErrors++; + if (consecutiveErrors >= MAX_CONSECUTIVE_ERRORS) { + globalCooldownUntil = Date.now() + 60000; + logger.warn("Global circuit breaker triggered due to consecutive errors"); + } + lastError = result.error ?? "Analysis worker failed"; conversationErrorCooldown.set( conversationKey, @@ -125,9 +160,16 @@ async function processBatch( return; } + consecutiveErrors = 0; // Reset circuit breaker conversationErrorCooldown.delete(conversationKey); shouldScheduleNext = true; } catch (error) { + consecutiveErrors++; + if (consecutiveErrors >= MAX_CONSECUTIVE_ERRORS) { + globalCooldownUntil = Date.now() + 60000; + logger.warn("Global circuit breaker triggered due to consecutive errors"); + } + lastError = error instanceof Error ? error.message : String(error); const errorStack = error instanceof Error ? error.stack : undefined; conversationErrorCooldown.set( @@ -159,46 +201,6 @@ async function processBatch( } } -function getAnalysisWorkerUrl(): URL { - const candidates = [ - new URL("./aiAnalysisWorker.js", import.meta.url), - new URL("../aiAnalysisWorker.js", import.meta.url), - new URL("./aiAnalysisWorker.ts", import.meta.url), - ]; - - for (const candidate of candidates) { - if (existsSync(fileURLToPath(candidate))) { - return candidate; - } - } - - return candidates[2]; -} - -async function runAnalysisInWorker( - conversationKey: string, - messages: MessageRecord[], -): Promise { - return new Promise((resolve, reject) => { - const worker = new Worker(getAnalysisWorkerUrl(), { - execArgv: process.execArgv, - }); - - worker.once("message", (response: AnalysisWorkerResponse) => { - worker.terminate().catch((error) => { - logger.warn({ error }, "Failed to terminate analysis worker"); - }); - resolve(response); - }); - worker.once("error", reject); - worker.once("exit", (code) => { - if (code !== 0) { - reject(new Error(`Analysis worker exited with code ${code}`)); - } - }); - worker.postMessage({ conversationKey, messages }); - }); -} /** * Debounced analysis trigger for a conversation diff --git a/src/moderation/llmModerationClient.ts b/src/moderation/llmModerationClient.ts index ad40698..d63eac3 100644 --- a/src/moderation/llmModerationClient.ts +++ b/src/moderation/llmModerationClient.ts @@ -130,199 +130,49 @@ export function extractJson(content: string): any { throw new Error("No JSON object found in response"); } -/** - * Parses LLM moderation response and validates against target IDs. - * 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[], ): AnalysisResult[] { - // Extract and parse JSON object - let parsed = extractJson(content); + let parsed: any; + try { + parsed = JSON.parse(content); + } catch (e) { + parsed = extractJson(content); + } - // If parsed is a direct array, wrap it in a results object to handle LLM variations if (Array.isArray(parsed)) { parsed = { results: parsed }; } else if (parsed && typeof parsed === "object" && !("results" in parsed)) { - // Handle single result object (has message_id or status) - if ("message_id" in parsed || "status" in parsed) { - const msgId = (parsed as any).message_id || (parsed as any).id; - parsed = { - results: [ - { - message_id: msgId, - status: (parsed as any).status || "clean", - flags: (parsed as any).flags || [], - score: - (parsed as any).score !== undefined ? (parsed as any).score : 0.1, - analysis: (parsed as any).analysis || "", - }, - ], - }; + const arrayKey = Object.keys(parsed).find((key) => + Array.isArray((parsed as any)[key]), + ); + if (arrayKey) { + parsed.results = (parsed as any)[arrayKey]; } else { - // Look for any array property (result, data, messages, moderation, etc.) - const arrayKey = Object.keys(parsed).find((key) => - Array.isArray((parsed as any)[key]), - ); - if (arrayKey) { - parsed.results = (parsed as any)[arrayKey]; - } + parsed = { results: [parsed] }; } } - // Validate structure - if (!parsed || typeof parsed !== "object" || !("results" in parsed)) { + if (!parsed || typeof parsed !== "object" || !Array.isArray(parsed.results)) { throw new Error("Response missing 'results' array"); } const response = parsed as RawModerationResponse; - if (!Array.isArray(response.results)) { - throw new Error("'results' must be an array"); - } - - // Track which target IDs were found const foundIds = new Set(); const targetIdSet = new Set(targetIds); - // Parse and validate each result const results: (AnalysisResult | null)[] = response.results.map( (result, index) => { const { message_id, status, flags, score, analysis } = result; - // Validate message_id exists and is in target list if (!message_id) { throw new Error("Result missing 'message_id'"); } - let finalId = String(message_id).trim(); - // Remove wrapping double quotes if any (common in some LLM outputs) - if (finalId.startsWith('"') && finalId.endsWith('"')) { - finalId = finalId.slice(1, -1).trim(); - } - if (finalId.startsWith("'") && finalId.endsWith("'")) { - finalId = finalId.slice(1, -1).trim(); - } - if (finalId.startsWith("[") && finalId.endsWith("]")) { - finalId = finalId.slice(1, -1).trim(); - } - - // Advanced Precision Loss & Alignment Fix - if (!targetIdSet.has(finalId)) { - const isSnowflake = (id: string) => - /^\d{15,22}$/.test(id) || id.includes("e+"); - - // 1. If there's only one target, map it directly if both are Snowflake-like - if ( - targetIds.length === 1 && - isSnowflake(finalId) && - isSnowflake(targetIds[0]) - ) { - log.warn( - { roundedId: finalId, matchedId: targetIds[0] }, - "Mapped single target ID directly to handle precision loss", - ); - finalId = targetIds[0]; - } else { - // 2. Try matching by long prefix similarity (e.g. 12+ digits) - let cleanLlmId = finalId; - if (finalId.includes("e+")) { - // Convert scientific notation back to string of digits if possible - try { - cleanLlmId = BigInt(Number(finalId)).toString(); - } catch (_) {} - } - - let bestMatch: string | null = null; - let maxCommonPrefixLen = 0; - - for (const targetId of targetIds) { - let commonLen = 0; - const minLen = Math.min(targetId.length, cleanLlmId.length); - for (let i = 0; i < minLen; i++) { - if (targetId[i] === cleanLlmId[i]) { - commonLen++; - } else { - break; - } - } - if (commonLen >= 12 && commonLen > maxCommonPrefixLen) { - maxCommonPrefixLen = commonLen; - bestMatch = targetId; - } - } - - if (bestMatch) { - log.warn( - { - roundedId: finalId, - cleanLlmId, - matchedId: bestMatch, - commonLength: maxCommonPrefixLen, - }, - "Fixed precision loss in message ID using prefix similarity", - ); - finalId = bestMatch; - } else if ( - response.results.length === targetIds.length && - targetIds[index] && - isSnowflake(finalId) && - isSnowflake(targetIds[index]) - ) { - // 3. Fallback: if the number of results matches the number of targets, - // map them 1:1 chronologically (by index) only if they are Snowflake-like - log.warn( - { roundedId: finalId, index, matchedId: targetIds[index] }, - "Aligned message ID using chronological index fallback", - ); - finalId = targetIds[index]; - } - } - } + const finalId = String(message_id).trim(); if (!targetIdSet.has(finalId)) { log.warn( @@ -339,25 +189,15 @@ export function parseModerationResponse( foundIds.add(finalId); - // Validate status const validStatuses = ["clean", "warn", "flagged"] as const; - if (!validStatuses.includes(status as (typeof validStatuses)[number])) { - throw new Error( - `Invalid status: ${status}. Must be one of: ${validStatuses.join(", ")}`, - ); - } + const safeStatus = validStatuses.includes(status as any) ? status : "clean"; - // Validate score: reject null/undefined/non-finite before coercion - if (score === null || score === undefined) { - throw new Error("Invalid score: must not be null or undefined"); - } let numScore = Number(score); if (!Number.isFinite(numScore)) { - throw new Error(`Invalid score: ${score}. Must be a finite number`); + numScore = 0; } numScore = Math.max(0, Math.min(1, numScore)); - // Coerce flags to string array let flagsArray: string[] = []; if (Array.isArray(flags)) { flagsArray = flags.map((f) => String(f)); @@ -365,12 +205,11 @@ export function parseModerationResponse( flagsArray = [String(flags)]; } - // Fallback analysis const analysisStr = analysis ? String(analysis) : ""; return { messageId: finalId, - status: status as "clean" | "warn" | "flagged", + status: safeStatus as "clean" | "warn" | "flagged", flags: flagsArray, score: numScore, analysis: analysisStr, @@ -382,7 +221,6 @@ export function parseModerationResponse( (r): r is AnalysisResult => r !== null, ); - // Check that all target IDs were found const missingIds = targetIds.filter((id) => !foundIds.has(id)); if (missingIds.length > 0) { log.warn( @@ -556,6 +394,12 @@ export async function runModerationAnalysis( return; } + const contentLength = Number(res.headers.get("content-length") || 0); + if (contentLength > 10 * 1024 * 1024) { + log.warn({ attachmentId: att.id, contentLength }, "Attachment too large, skipping"); + return; + } + const imageBytes = Buffer.from(await res.arrayBuffer()); const sniffedMime = sniffImageMimeType(imageBytes); if (!sniffedMime) { @@ -740,8 +584,36 @@ CRITICAL: "message_id" HARUS berupa STRING (dibungkus tanda kutip ganda). Jangan ], temperature: 0.2, top_p: 0.95, - max_tokens: 65536, - response_format: { type: "json_object" }, + max_tokens: 16384, + response_format: { + type: "json_schema", + json_schema: { + name: "moderation", + strict: true, + schema: { + type: "object", + properties: { + results: { + type: "array", + items: { + type: "object", + properties: { + message_id: { type: "string" }, + status: { type: "string", enum: ["clean", "warn", "flagged"] }, + flags: { type: "array", items: { type: "string" } }, + score: { type: "number" }, + analysis: { type: "string" } + }, + required: ["message_id", "status", "flags", "score", "analysis"], + additionalProperties: false + } + } + }, + required: ["results"], + additionalProperties: false + } + } + }, stream: false, chat_template_kwargs: { enable_thinking: false }, reasoning_budget: 0, @@ -802,54 +674,26 @@ CRITICAL: "message_id" HARUS berupa STRING (dibungkus tanda kutip ganda). Jangan const errorMsg = parseError instanceof Error ? parseError.message : String(parseError); const content: string = lastInvalidContent; - 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. Marking all targets as analysis errors.", - ); - parsed = targetIds.map((id) => ({ - messageId: id, - status: "error", - flags: ["analysis_parse_failed"], - score: 0, - analysis: `Parsing failed: ${errorMsg}.`, - })); - } + + 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. Marking all targets as analysis errors.", + ); + parsed = targetIds.map((id) => ({ + messageId: id, + status: "error", + flags: ["analysis_parse_failed"], + score: 0, + analysis: `Parsing failed: ${errorMsg}.`, + })); } log.info( diff --git a/src/moderation/messageStore.ts b/src/moderation/messageStore.ts index fbc1308..460c3fe 100644 --- a/src/moderation/messageStore.ts +++ b/src/moderation/messageStore.ts @@ -419,6 +419,26 @@ export async function updateMessageAIAnalysis( } } +export async function updateMessagesAIAnalysisBulk( + updates: Array<{ messageId: string; result: AIAnalysisUpdate }> +): Promise { + if (updates.length === 0) return []; + try { + const results = await Promise.all( + updates.map(({ messageId, result }) => updateMessageAIAnalysis(messageId, result)) + ); + return results.filter((r): r is MessageRecord => r !== null); + } catch (error) { + logger.error( + { + error: error instanceof Error ? error.message : String(error), + }, + "Failed to bulk update messages AI analysis", + ); + throw error; + } +} + export async function getPendingAIAnalysisMessages( limit: number = 25, ): Promise {