Revert "feat: migrate frontend to Astro + expand AI moderation + backend admin/runtime config"

This reverts commit d59b59a7a7.
This commit is contained in:
asepharyana
2026-07-02 03:54:44 +07:00
parent 9636087e99
commit 7efaf00c93
91 changed files with 670 additions and 11161 deletions
@@ -1,86 +0,0 @@
/**
* abortHelper.ts — Centralized AbortController with automatic timeout cleanup.
*
* All `new AbortController()` + `setTimeout(abort, ms)` patterns across the
* moderation subsystem are replaced by this module so that:
* 1. Every timer calls `.unref()` so it cannot keep Node alive during shutdown.
* 2. Cleanup is guaranteed via `cleanup()` or `withAbortTimeout()`.
* 3. The AbortError is distinguishable via `isAbortError()`.
*/
/**
* Create an AbortController that auto-aborts after `ms` milliseconds.
* Returns the signal and a `cleanup()` function that MUST be called
* (typically in a `finally` block) to cancel the timer.
*
* @example
* ```
* const { signal, cleanup } = createAbortTimeout(8000);
* try {
* await fetch(url, { signal });
* } finally {
* cleanup();
* }
* ```
*/
export function createAbortTimeout(ms: number): {
signal: AbortSignal;
cleanup: () => void;
} {
const controller = new AbortController();
const timer = setTimeout(() => controller.abort(), ms);
// Prevent the timer from keeping the Node.js event loop alive during shutdown
timer.unref();
return {
signal: controller.signal,
cleanup: () => clearTimeout(timer),
};
}
/**
* Higher-order wrapper that runs an async function with an abort timeout.
* The cleanup is handled automatically — callers never forget `clearTimeout`.
*
* If the operation is aborted by the timeout, the resulting error is re-thrown
* with a descriptive message.
*
* @example
* ```
* const result = await withAbortTimeout(8000, async (signal) => {
* return fetch(url, { signal }).then(r => r.json());
* }, "SearXNG search");
* ```
*/
export async function withAbortTimeout<T>(
ms: number,
fn: (signal: AbortSignal) => Promise<T>,
label = "operation",
): Promise<T> {
const { signal, cleanup } = createAbortTimeout(ms);
try {
return await fn(signal);
} catch (err) {
if (signal.aborted) {
throw new Error(`${label} timed out after ${ms}ms`);
}
throw err;
} finally {
cleanup();
}
}
/**
* Check whether an error was caused by an AbortController signal firing
* (either our explicit abort or the timeout).
*/
export function isAbortError(err: unknown): boolean {
if (err instanceof DOMException) return err.name === "AbortError";
if (err instanceof Error) {
return (
err.name === "AbortError" ||
err.message.includes("timed out after") ||
err.message.includes("aborted")
);
}
return false;
}
@@ -186,14 +186,14 @@ async function processBatch(job: {
}
}
const allRows: MessageRecord[] = [];
// ── Parallel: text-only + media analysis run concurrently ──────────
// Text-only → fast LLM call. Media → download + vision + LLM.
// Running both in parallel means media downloads overlap with text LLM call.
// Each path saves to DB as soon as its own results are ready.
// Each promise resolves to its own MessageRecord[] — combined via
// destructured Promise.all to avoid race conditions on a shared array.
// ────────────────────────────────────────────────────────────────────
const textPromise: Promise<MessageRecord[]> = textOnly.length > 0
const textPromise = textOnly.length > 0
? runModerationAnalysis({
targets: textOnly,
contextText: contextLines.join("\n"),
@@ -216,18 +216,17 @@ async function processBatch(job: {
}));
if (updates.length > 0) {
return updateMessagesAIAnalysisBulk(updates).then((rows) => {
allRows.push(...rows);
logger.info(
{ count: updates.length, conversationKey },
"Text-only batch saved — media analysis still in progress",
);
return rows;
});
}
return [];
})
: Promise.resolve([]);
: Promise.resolve();
const mediaPromise: Promise<MessageRecord[]> = media.length > 0
const mediaPromise = media.length > 0
? runModerationAnalysis({
targets: media,
contextText: contextLines.join("\n"),
@@ -249,15 +248,15 @@ async function processBatch(job: {
},
}));
if (updates.length > 0) {
return updateMessagesAIAnalysisBulk(updates);
return updateMessagesAIAnalysisBulk(updates).then((rows) => {
allRows.push(...rows);
});
}
return [];
})
: Promise.resolve([]);
: Promise.resolve();
// Wait for both to complete and destructure results — no shared mutable array
const [textRows, mediaRows] = await Promise.all([textPromise, mediaPromise]);
const allRows = [...textRows, ...mediaRows];
// Wait for both to complete
await Promise.all([textPromise, mediaPromise]);
logger.info(
{ total: messages.length, textOnly: textOnly.length, media: media.length, saved: allRows.length },
@@ -178,6 +178,22 @@ export async function processBatch(
messages,
})) as AnalysisWorkerResponse;
// Do not broadcast or auto-delete if it's an API failure that will be reverted.
for (const row of result.rows) {
let isApiFailure = false;
if (row.ai_status === "error") {
try {
const flags = JSON.parse(row.ai_moderation_flags ?? "[]") as string[];
isApiFailure = flags.includes("analysis_api_failed");
} catch {}
}
if (!isApiFailure) {
broadcastAnalysisCompleted(row);
scheduleAutoDelete(row);
}
}
// Post-batch reputation updates (fire-and-forget)
postBatchReputationUpdate(
result.rows.filter((r) => {
@@ -36,10 +36,6 @@ export const workerPool = new Piscina({
filename: fileURLToPath(getAnalysisWorkerUrl()),
execArgv: process.execArgv,
maxThreads: config.PISCINA_MAX_THREADS ?? availableParallelism(),
// Each worker processes at most 1 task at a time so the pool itself
// acts as the concurrency governor. Combined with per-worker p-limit
// inside concurrencyLimiter.ts, this prevents LLM API overload.
concurrentTasksPerWorker: 1,
});
/**
@@ -232,7 +228,7 @@ export function scheduleAutoDelete(row: MessageRecord): void {
};
if (config.AUTO_DELETE_FLAGGED_DELAY_MS > 0) {
setTimeout(run, config.AUTO_DELETE_FLAGGED_DELAY_MS).unref();
setTimeout(run, config.AUTO_DELETE_FLAGGED_DELAY_MS);
return;
}
setImmediate(run);
@@ -22,6 +22,7 @@ function updateCounts(): void {
}
export async function withLlmConcurrency<T>(fn: () => Promise<T>): Promise<T> {
const queuedAt = activeCount + pendingCount;
pendingCount++;
logger.debug(
{ activeCount, pendingCount, maxConcurrent: config.AI_LLM_MAX_CONCURRENT },
@@ -29,17 +30,17 @@ export async function withLlmConcurrency<T>(fn: () => Promise<T>): Promise<T> {
);
return llmSemaphore(async () => {
pendingCount--;
activeCount++;
if (activeCount >= (config.AI_LLM_MAX_CONCURRENT ?? 5)) {
logger.warn(
{ activeCount, maxConcurrent: config.AI_LLM_MAX_CONCURRENT },
"LLM concurrency limit reached",
);
}
try {
pendingCount--;
activeCount++;
if (activeCount >= (config.AI_LLM_MAX_CONCURRENT ?? 5)) {
logger.warn(
{ activeCount, maxConcurrent: config.AI_LLM_MAX_CONCURRENT },
"LLM concurrency limit reached",
);
}
return await fn();
} finally {
activeCount--;
@@ -277,50 +277,6 @@ async function processIndividualFallback(
}
}
// ---------------------------------------------------------------------------
// Individual fallback retry queue (used when circuit breaker is active)
// ---------------------------------------------------------------------------
/** Messages awaiting retry when individual CB cools down. */
const individualRetryQueue = new LRUCache<string, MessageRecord>({ max: 10000 });
let individualRetryTimer: ReturnType<typeof setTimeout> | null = null;
const INDIVIDUAL_RETRY_CHECK_MS = 30000;
/**
* Schedule a retry for messages that were skipped because the individual
* circuit breaker was active. Retries once after cooldown expires.
*/
function scheduleIndividualRetry(messages: MessageRecord[]): void {
for (const msg of messages) {
if (!individualRetryQueue.has(msg.id)) {
individualRetryQueue.set(msg.id, msg);
}
}
if (individualRetryTimer === null) {
individualRetryTimer = setTimeout(() => {
individualRetryTimer = null;
if (Date.now() < individualCooldownUntil) {
// Still in cooldown — reschedule
scheduleIndividualRetry([]);
return;
}
const ids = [...individualRetryQueue.keys()];
const msgs: MessageRecord[] = [];
for (const id of ids) {
const m = individualRetryQueue.get(id);
if (m) {
individualRetryQueue.delete(id);
msgs.push(m);
}
}
if (msgs.length > 0) {
logger.info({ count: msgs.length }, "Retrying individual fallback messages after circuit breaker cooldown");
enqueueIndividualFallbacks(msgs);
}
}, INDIVIDUAL_RETRY_CHECK_MS);
}
}
// ---------------------------------------------------------------------------
// Enqueue individual fallbacks
// ---------------------------------------------------------------------------
@@ -339,9 +295,8 @@ export function enqueueIndividualFallbacks(messages: MessageRecord[]): void {
until: new Date(individualCooldownUntil).toISOString(),
skipped: messages.length,
},
"Individual fallback circuit breaker active messages queued for retry",
"Individual fallback circuit breaker active -- messages will be recovered later",
);
scheduleIndividualRetry(messages);
return;
}
@@ -23,7 +23,6 @@ import type {
import { llmVision } from "./llmClient.js";
import { sanitizeAiContent } from "./moderationPrompt.js";
import {
buildAttachmentTextOnlyWarning,
buildCustomEmojiVisionPrompt,
buildGeneralImageVisionPrompt,
buildStickerTextOnlyWarning,
@@ -52,7 +51,6 @@ import { searchSearxng, extractSearchQueries, formatSearchResults } from "./sear
import { getUserProfile } from "./userProfileStore.js";
import { initializeUserReputation } from "./userReputationStore.js";
import { escapeXml, getAnalysisContent, buildReferenceXml } from "./moderationBuilders.js";
import { createAbortTimeout, isAbortError } from "./abortHelper.js";
// ---------------------------------------------------------------------------
// Types
@@ -294,97 +292,50 @@ async function downloadSingleAttachment(
targetId: string,
maxDimension: number,
imageMap: Map<string, MessageImagePart[]>,
mediaAnalysisMap?: Map<string, string[]>,
): Promise<void> {
const log = createChildLogger("mediaAnalysis");
const urlToUse = att.uploaded_url ?? att.discord_url ?? null;
if (!urlToUse) return;
// Collect all available URLs — uploaded_url first (Telegram CDN, faster),
// then discord_url as fallback (may have expired CDN signature)
const urlsToTry = [
att.uploaded_url,
att.discord_url,
].filter((url): url is string => url !== null && url !== undefined);
const controller = new AbortController();
const timeoutId = setTimeout(() => controller.abort(), 15000);
try {
const res = await fetch(urlToUse, { signal: controller.signal });
if (!res.ok || !res.body) return;
// No URL at all — record a neutral fallback so downstream knows the
// attachment existed but was unreachable
if (urlsToTry.length === 0) {
if (mediaAnalysisMap) {
const existing = mediaAnalysisMap.get(targetId) ?? [];
existing.push(`[attachment: "${att.filename}" dari pesan id=${targetId} — tidak ada URL untuk diunduh]`);
mediaAnalysisMap.set(targetId, existing);
}
return;
}
let lastError: Error | null = null;
// Try each URL, with 3 retries per URL (exponential backoff: 1s, 2s)
for (const url of urlsToTry) {
for (let attempt = 0; attempt < 3; attempt++) {
try {
const { signal, cleanup } = createAbortTimeout(15000);
const res = await fetch(url, { signal });
if (!res.ok || !res.body) {
cleanup();
throw new Error(`HTTP ${res.status}`);
}
let totalBytes = 0;
const chunks: Uint8Array[] = [];
const reader = res.body.getReader();
while (true) {
const { done, value } = await reader.read();
if (done) break;
if (value) {
totalBytes += value.length;
if (totalBytes > 10 * 1024 * 1024) {
reader.cancel().catch(() => {});
res.body?.cancel().catch(() => {});
return;
}
chunks.push(value);
}
}
const imageBytes = Buffer.concat(chunks);
const sniffedMime = sniffImageMimeType(imageBytes);
if (!sniffedMime && att.type.startsWith("video/")) {
await extractVideoFrames(att, imageBytes, targetId, maxDimension, imageMap, signal);
return;
}
if (!sniffedMime) return;
const { data: resizedBuffer, mimeType: resizedMime } = await resizeImageForVision(imageBytes, maxDimension);
const dataUrl = `data:${resizedMime};base64,${resizedBuffer.toString("base64")}`;
addImageToMap(imageMap, targetId, {
type: "image_url",
image_url: { url: dataUrl },
sourceLabel: `[gambar di atas adalah attachment ${att.filename} dari pesan id=${att.message_id}]`,
});
cleanup();
return; // success!
} catch (err) {
lastError = err instanceof Error ? err : new Error(String(err));
log.warn(
{ attachmentId: att.id, url: url.slice(0, 80), attempt, error: lastError.message },
`Download attempt ${attempt + 1}/3 failed`,
);
if (attempt < 2) await delay(1000 * (attempt + 1)); // backoff: 1s, 2s
let totalBytes = 0;
const chunks: Uint8Array[] = [];
const reader = res.body.getReader();
while (true) {
const { done, value } = await reader.read();
if (done) break;
if (value) {
totalBytes += value.length;
if (totalBytes > 10 * 1024 * 1024) { reader.cancel(); return; }
chunks.push(value);
}
}
}
const imageBytes = Buffer.concat(chunks);
const sniffedMime = sniffImageMimeType(imageBytes);
// All URLs + all retries exhausted — record a neutral fallback
if (mediaAnalysisMap) {
const existing = mediaAnalysisMap.get(targetId) ?? [];
existing.push(buildAttachmentTextOnlyWarning(att.filename, targetId));
mediaAnalysisMap.set(targetId, existing);
if (!sniffedMime && att.type.startsWith("video/")) {
await extractVideoFrames(att, imageBytes, targetId, maxDimension, imageMap);
return;
}
if (!sniffedMime) return;
const { data: resizedBuffer, mimeType: resizedMime } = await resizeImageForVision(imageBytes, maxDimension);
const dataUrl = `data:${resizedMime};base64,${resizedBuffer.toString("base64")}`;
addImageToMap(imageMap, targetId, {
type: "image_url",
image_url: { url: dataUrl },
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 {
clearTimeout(timeoutId);
}
log.warn(
{ attachmentId: att.id, filename: att.filename, error: lastError?.message },
"All download attempts exhausted for attachment",
);
}
async function extractVideoFrames(
@@ -393,7 +344,6 @@ async function extractVideoFrames(
targetId: string,
maxDimension: number,
imageMap: Map<string, MessageImagePart[]>,
signal?: AbortSignal,
): Promise<void> {
const log = createChildLogger("mediaAnalysis");
const execFileAsync = promisify(execFile);
@@ -401,19 +351,15 @@ async function extractVideoFrames(
const inputPath = path.join(tmpDir, att.filename || "video.mp4");
const outputPattern = path.join(tmpDir, "frame-%03d.jpg");
try {
if (signal?.aborted) return;
await writeFile(inputPath, videoBytes);
if (signal?.aborted) return;
const { stdout: durationStr } = await execFileAsync("/usr/bin/ffprobe", [
"-v", "error", "-show_entries", "format=duration", "-of", "csv=p=0", inputPath,
], { timeout: 10000, signal });
if (signal?.aborted) return;
], { timeout: 10000 });
const duration = parseFloat(durationStr.trim()) || 1;
const fps = (3 / duration).toFixed(6);
if (signal?.aborted) return;
await execFileAsync("/usr/bin/ffmpeg", [
"-i", inputPath, "-vf", `fps=${fps}`, "-frames:v", "4", "-vsync", "vfr", "-q:v", "2", outputPattern,
], { timeout: 30000, signal });
], { timeout: 30000 });
for (let i = 1; i <= 4; i++) {
try {
const framePath = path.join(tmpDir, `frame-${String(i).padStart(3, "0")}.jpg`);
@@ -431,8 +377,10 @@ async function extractVideoFrames(
} catch (ffmpegErr) {
log.warn({ attachmentId: att.id, error: ffmpegErr instanceof Error ? ffmpegErr.message : String(ffmpegErr) }, "ffmpeg failed");
} finally {
// rm(tmpDir, { recursive: true }) already removes all files inside,
// so individual unlink calls are redundant. Just clean up the whole dir.
try { await unlink(inputPath); } catch { /* ignore */ }
for (let i = 1; i <= 4; i++) {
try { await unlink(path.join(tmpDir, `frame-${String(i).padStart(3, "0")}.jpg`)); } catch { /* ignore */ }
}
try { await rm(tmpDir, { recursive: true, force: true }); } catch { /* ignore */ }
}
}
@@ -538,7 +486,7 @@ export async function prepareMediaMessage(
.filter((a) => a.message_id === targetId && (a.uploaded_url ?? a.discord_url ?? null) && (a.type.startsWith("image/") || a.type.startsWith("video/")))
.slice(0, 8);
for (const att of msgAttachments) {
downloadPromises.push(downloadSingleAttachment(att, targetId, maxDimension, imageMap, mediaAnalysisMap));
downloadPromises.push(downloadSingleAttachment(att, targetId, maxDimension, imageMap));
}
// URLs
@@ -557,8 +505,8 @@ export async function prepareMediaMessage(
await Promise.all(downloadPromises);
if (urlWebTexts.length > 0) webTextMap.set(targetId, urlWebTexts);
// Vision analysis — use allSettled so one failure doesn't cascade
const visionResults = await Promise.allSettled(
// Vision analysis
await Promise.all(
Array.from(imageMap.entries()).flatMap(([msgId, images]) =>
images.map(async (image) => {
const summary = await analyzeSingleMediaImage(msgId, image);
@@ -568,12 +516,6 @@ export async function prepareMediaMessage(
}),
),
);
// Log any vision failures without aborting the batch
for (const r of visionResults) {
if (r.status === "rejected") {
log.warn({ error: r.reason instanceof Error ? r.reason.message : String(r.reason) }, "Individual vision analysis failed (batched)");
}
}
// SearXNG
let searxngXml = "";
@@ -29,7 +29,6 @@ import {
import { extractUrlsFromText, fetchUrlSafely } from "./urlFetcher.js";
import { getUserProfile } from "./userProfileStore.js";
import { initializeUserReputation } from "./userReputationStore.js";
import { createAbortTimeout, isAbortError } from "./abortHelper.js";
const log = createChildLogger("moderationOrchestrator");
@@ -41,8 +40,10 @@ interface RetryState {
lastInvalidContent: string | null;
}
// ─── Few-shot correction builder ────────────────────────────────────────────
const _buildCorrectedFewShotExamples = async (): Promise<string> => {
// ---------------------------------------------------------------------------
// Few-shot correction builder
// ---------------------------------------------------------------------------
async function buildCorrectedFewShotExamples(): Promise<string> {
try {
const corrections = await getRecentCorrectedModerations(5);
if (corrections.length === 0) return "";
@@ -61,22 +62,6 @@ const _buildCorrectedFewShotExamples = async (): Promise<string> => {
} catch {
return "";
}
};
// ─── In-memory cache untuk correctedFewShotExamples ──────────────────────
// getCachedFewShotExamples() dipanggil di banyak tempat (setiap sub-batch
// dan retry), padahal datanya jarang berubah. Cache sederhana TTL 60 detik
// mengurangi redundant DB queries dari O(retries × subBatches) ke O(1).
let _fewShotCache: { result: string; expiresAt: number } | null = null;
const FEW_SHOT_CACHE_TTL = 60_000; // 60 detik
async function getCachedFewShotExamples(): Promise<string> {
if (_fewShotCache && Date.now() < _fewShotCache.expiresAt) {
return _fewShotCache.result;
}
const result = await _buildCorrectedFewShotExamples();
_fewShotCache = { result, expiresAt: Date.now() + FEW_SHOT_CACHE_TTL };
return result;
}
// ---------------------------------------------------------------------------
@@ -223,25 +208,7 @@ async function runTextOnlyBatch(
}
const urlArr = Array.from(allUrls).slice(0, 10);
if (urlArr.length === 0) return new Map<string, string>();
// Per-host rate limiting: add a small delay between requests to the same
// domain to avoid overwhelming third-party servers with concurrent fetches.
const hostGroups = new Map<string, string[]>();
for (const url of urlArr) {
try {
const host = new URL(url).hostname;
const group = hostGroups.get(host) ?? [];
group.push(url);
hostGroups.set(host, group);
} catch { /* invalid URL, skip */ }
}
const results = await Promise.allSettled(
Array.from(hostGroups.values()).flatMap((group) =>
group.map((url, idx) => () =>
idx > 0 ? delay(200 * idx).then(() => fetchUrlSafely(url)) : fetchUrlSafely(url),
),
).map((fn) => fn()),
);
const results = await Promise.allSettled(urlArr.map((url) => fetchUrlSafely(url)));
const map = new Map<string, string>();
for (let i = 0; i < urlArr.length; i++) {
const r = results[i];
@@ -324,7 +291,7 @@ async function runTextOnlyBatch(
const buildContent = async (state: RetryState): Promise<string> => {
const correction = state.lastParseError ? { error: state.lastParseError, preview: state.lastInvalidContent?.slice(0, 800) ?? "<empty>" } : undefined;
const correctedExamples = await getCachedFewShotExamples();
const correctedExamples = await buildCorrectedFewShotExamples();
const systemText = buildSystemPromptModular({ contextText, mode: "text", correction, correctedExamples, channelCulture });
const messagesBlock = (await Promise.all(batch.map(async (msg) => {
@@ -347,18 +314,20 @@ async function runTextOnlyBatch(
return `${systemText}${searxngBlock}\n\n<messages_to_analyze>\n${messagesBlock}\n</messages_to_analyze>`;
};
const { signal, cleanup } = createAbortTimeout(timeoutMs);
const abortController = new AbortController();
const timeoutId = setTimeout(() => abortController.abort(), timeoutMs);
timeoutId.unref();
let batchResult: { results: AnalysisResult[]; raw: unknown };
try {
batchResult = await callModerationLLM(buildContent, targetIds, `text-batch-${i + 1}`, signal);
batchResult = await callModerationLLM(buildContent, targetIds, `text-batch-${i + 1}`, abortController.signal);
} catch (err: any) {
if (isAbortError(err)) {
if (err.name === "AbortError" || abortController.signal.aborted) {
throw new Error(`Text-only batch sub-batch ${i + 1} timed out for messages ${targetIds.join(", ")}`);
}
throw err;
} finally {
cleanup();
clearTimeout(timeoutId);
}
// Fan-out results for deduplicated messages
@@ -403,7 +372,7 @@ async function runMediaBatch(
const channelId = targets[0].channel_id;
const channelCultureObj = channelId ? await getChannelCulture(channelId) : null;
const channelCulture = channelCultureObj?.culture_summary;
const correctedExamples = await getCachedFewShotExamples();
const correctedExamples = await buildCorrectedFewShotExamples();
const systemText = buildSystemPromptModular({ contextText, mode: "mixed", correctedExamples, channelCulture });
const messagesBlock = prepared.map((p) => p.messageBlock).join("\n");
@@ -412,24 +381,26 @@ async function runMediaBatch(
const perMsgTimeout = config.AI_LLM_MEDIA_ANALYSIS_TIMEOUT_MS ?? 60000;
const batchTimeout = Math.min(Math.max(perMsgTimeout, perMsgTimeout * targets.length), 300_000);
const { signal, cleanup } = createAbortTimeout(batchTimeout);
const abortController = new AbortController();
const timeoutId = setTimeout(() => abortController.abort(), batchTimeout);
timeoutId.unref();
try {
const result = await callModerationLLM(
async (_state: RetryState) => userContent,
targetIds,
`media-batch:${targetIds.length}msgs`,
signal,
abortController.signal,
);
log.info({ mediaCount: targets.length, resultCount: result.results.length }, "Media batch analysis complete");
return result;
} catch (err: any) {
if (isAbortError(err)) {
if (err.name === "AbortError" || abortController.signal.aborted) {
throw new Error(`Media batch analysis timed out after ${batchTimeout}ms for ${targets.length} messages`);
}
throw err;
} finally {
cleanup();
clearTimeout(timeoutId);
}
}
@@ -475,7 +446,7 @@ export async function runModerationAnalysis(
const rawContent = target.edited_content ?? target.content;
if (!rawContent.trim()) { uncachedTargets.push(target); continue; }
const cacheKey = makeTextModerationCacheKey(rawContent, target.user_id);
const cacheKey = makeTextModerationCacheKey(rawContent);
if (seenCacheKeys.has(cacheKey)) {
const previousHit = cacheHits.find((h) => h.messageId !== target.id);
if (previousHit) {
@@ -563,7 +534,7 @@ export async function runModerationAnalysis(
if (evidence.attachments.length > 0 || evidence.stickers.length > 0 || evidence.embeds.length > 0) continue;
}
const cacheKey = makeTextModerationCacheKey(rawContent, target.user_id);
const cacheKey = makeTextModerationCacheKey(rawContent);
setCachedTextModeration(cacheKey, {
flags: result.flags ?? [],
score: result.score ?? 0,
@@ -573,7 +544,7 @@ export async function runModerationAnalysis(
confidence: result.confidence ?? result.score ?? 0,
recommendedAction: result.recommendedAction ?? "none",
status: result.status,
}).catch((e) => log.error({ error: e instanceof Error ? e.message : String(e) }, "Failed to cache text moderation result"));
}).catch(() => {});
}
const allResults = [...cacheHits, ...textBatchResult.results, ...mediaBatchResult.results];
@@ -1,6 +1,5 @@
import Redis from "ioredis";
import { createChildLogger } from "@bete/shared/logger";
import { createAbortTimeout } from "./abortHelper.js";
const log = createChildLogger("searxng-search");
@@ -9,97 +8,15 @@ const MAX_RESULTS = 3;
const TIMEOUT_MS = 8000;
const CACHE_TTL = 86400; // 24 hours
const CACHE_PREFIX = "searxng:";
/** How often to attempt reconnection when Redis is down (ms). */
const RECONNECT_INTERVAL_MS = 60_000;
let redis: Redis | null = null;
let _initialized = false;
let _redisUrl = "";
/** Tracks whether Redis is currently healthy (connected + responding). */
let _redisHealthy = false;
let _lastLogAt = 0; // throttle repeated warn logs to once per 60s
// ---------------------------------------------------------------------------
// Health tracking
// ---------------------------------------------------------------------------
/**
* Returns true if the SearXNG Redis cache is connected and healthy.
* Use this for health-check endpoints or status dashboards.
*/
export function isSearxngCacheAvailable(): boolean {
return _redisHealthy;
}
/**
* Returns a human-readable status string for logging / health endpoints.
*/
export function getSearxngCacheStatus(): string {
if (!_initialized) return "not-initialized";
if (!redis) return "no-url";
return _redisHealthy ? "healthy" : "disconnected";
}
// ---------------------------------------------------------------------------
// Reconnection helper
// ---------------------------------------------------------------------------
/**
* Schedule a one-shot reconnection attempt after RECONNECT_INTERVAL_MS.
* Only one reconnection timer runs at a time.
*/
let _reconnectTimer: ReturnType<typeof setTimeout> | null = null;
function scheduleReconnect(): void {
if (_reconnectTimer) return; // already scheduled
_reconnectTimer = setTimeout(async () => {
_reconnectTimer = null;
if (!redis || _redisHealthy) return; // nothing to do
log.info("SearXNG Redis: attempting reconnection...");
try {
// ioredis reconnects automatically if `lazyConnect` is false,
// but we set it to true, so we need to manually call connect().
await redis.connect();
// If we get here, connection succeeded
_redisHealthy = true;
log.info("SearXNG Redis: reconnected successfully ✅");
} catch {
_redisHealthy = false;
const now = Date.now();
if (now - _lastLogAt > 60_000) {
log.warn(
{ nextRetryMs: RECONNECT_INTERVAL_MS },
"SearXNG Redis: reconnection failed — will retry",
);
_lastLogAt = now;
}
// Schedule another attempt
scheduleReconnect();
}
}, RECONNECT_INTERVAL_MS);
_reconnectTimer.unref?.();
}
// ---------------------------------------------------------------------------
// Initialization
// ---------------------------------------------------------------------------
/**
* Initialize Redis connection for SearXNG cache.
* Safe to call multiple times — only creates one connection.
* Returns true if Redis cache is available, false if falling back to no-cache.
*/
export function initSearxngCache(redisUrl: string): boolean {
if (_initialized) return _redisHealthy;
_initialized = true;
_redisUrl = redisUrl;
if (!redisUrl) {
log.warn("No REDIS_URL provided — SearXNG cache disabled");
return false;
}
export function initSearxngCache(redisUrl: string): void {
if (redis) return;
redis = new Redis(redisUrl, {
maxRetriesPerRequest: 3,
retryStrategy(times) {
@@ -109,92 +26,16 @@ export function initSearxngCache(redisUrl: string): boolean {
lazyConnect: true,
enableReadyCheck: false,
});
redis.on("error", (err) => {
const wasHealthy = _redisHealthy;
_redisHealthy = false;
if (wasHealthy) {
// State transition: healthy → unhealthy — always log
log.warn(
{ error: err.message },
"SearXNG Redis: connection lost — falling back to no-cache",
);
} else {
// Already unhealthy — throttle repeated error logs
const now = Date.now();
if (now - _lastLogAt > 60_000) {
log.warn(
{ error: err.message },
"SearXNG Redis: still disconnected",
);
_lastLogAt = now;
}
}
scheduleReconnect();
log.warn({ err: err.message }, "SearXNG Redis cache error");
});
redis.on("ready", () => {
if (!_redisHealthy) {
_redisHealthy = true;
log.info("SearXNG Redis: connected and healthy ✅");
}
});
redis.connect().catch(() => {
_redisHealthy = false;
log.warn("SearXNG Redis: initial connection failed — running without cache");
scheduleReconnect();
log.warn("SearXNG Redis cache unavailable — falling back to no-cache");
redis = null;
});
log.info("SearXNG Redis cache initialized");
return true;
}
// ---------------------------------------------------------------------------
// Cache operations with visible failure logging
// ---------------------------------------------------------------------------
async function cacheGet(key: string): Promise<string | null> {
if (!redis || !_redisHealthy) return null;
try {
const result = await redis.get(key);
return result;
} catch (err) {
// Log once per minute to avoid log spam
const now = Date.now();
if (now - _lastLogAt > 60_000) {
log.warn(
{ error: err instanceof Error ? err.message : String(err) },
"SearXNG Redis: cache read failed",
);
_lastLogAt = now;
}
_redisHealthy = false;
scheduleReconnect();
return null;
}
}
function cacheSet(key: string, value: string, ttlSeconds: number): void {
if (!redis || !_redisHealthy) return;
redis.setex(key, ttlSeconds, value).catch((err) => {
const now = Date.now();
if (now - _lastLogAt > 60_000) {
log.warn(
{ error: err instanceof Error ? err.message : String(err) },
"SearXNG Redis: cache write failed",
);
_lastLogAt = now;
}
_redisHealthy = false;
scheduleReconnect();
});
}
// ---------------------------------------------------------------------------
// Search
// ---------------------------------------------------------------------------
export interface SearxngResult {
title: string;
url: string;
@@ -212,30 +53,33 @@ export async function searchSearxng(
const cacheKey = `${CACHE_PREFIX}${category}:${query.toLowerCase().trim()}`;
// Try cache first
const cached = await cacheGet(cacheKey);
if (cached) {
log.debug({ query, category }, "SearXNG cache HIT");
if (redis) {
try {
return JSON.parse(cached) as SearxngResult[];
const cached = await redis.get(cacheKey);
if (cached) {
log.debug({ query, category }, "SearXNG cache HIT");
return JSON.parse(cached) as SearxngResult[];
}
} catch {
// Corrupted cache entry — continue to API
// Cache read failed, continue to API
}
}
// Cache miss — hit SearXNG API
try {
const url = `${SEARXNG_BASE_URL}/search?q=${encodeURIComponent(query)}&format=json&language=id&categories=${category}`;
const { signal, cleanup } = createAbortTimeout(TIMEOUT_MS);
const controller = new AbortController();
const timeoutId = setTimeout(() => controller.abort(), TIMEOUT_MS);
const response = await fetch(url, {
signal,
signal: controller.signal,
headers: {
Accept: "application/json",
"User-Agent":
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
},
});
cleanup();
clearTimeout(timeoutId);
if (!response.ok) {
log.warn({ status: response.status, query }, "SearXNG search failed");
@@ -253,7 +97,11 @@ export async function searchSearxng(
}));
// Store in cache (fire and forget — don't block on write)
cacheSet(cacheKey, JSON.stringify(mapped), CACHE_TTL);
if (redis) {
redis.setex(cacheKey, CACHE_TTL, JSON.stringify(mapped)).catch(() => {
// Cache write failed silently
});
}
log.debug({ query, category, resultCount: mapped.length }, "SearXNG search OK");
return mapped;
@@ -266,10 +114,6 @@ export async function searchSearxng(
}
}
// ---------------------------------------------------------------------------
// Query extraction
// ---------------------------------------------------------------------------
/**
* Extract meaningful search queries from message content.
* Uses multiple strategies to find terms worth searching.
@@ -335,10 +179,6 @@ export function extractSearchQueries(content: string): string[] {
return Array.from(queries).slice(0, 3);
}
// ---------------------------------------------------------------------------
// Formatting
// ---------------------------------------------------------------------------
/**
* Format SearXNG results as XML for LLM context.
*/
@@ -67,24 +67,6 @@ export function buildStickerTextOnlyWarning(
);
}
/**
* Build a neutral fallback text when an image/video attachment cannot be
* downloaded or analyzed. Unlike the sticker warning, this is deliberately
* neutral — it only records *that* an attachment existed, without any
* instruction to the LLM about how to treat it. The absence of visual
* data already means the LLM must rely on message content alone.
*
* Returns a formatted string for inclusion in the media context block.
*/
export function buildAttachmentTextOnlyWarning(
filename: string,
messageId: string,
): string {
const fallback = `[attachment: "${filename}" dari pesan id=${messageId} — tidak tersedia untuk analisis visual]`;
logger.debug({ filename, messageId }, "Built attachment text-only fallback");
return fallback;
}
/**
* Prompt used when a custom emoji image was successfully downloaded
* and is being sent to the vision LLM as a base64 image.
@@ -49,11 +49,6 @@ export async function getCachedText(
/**
* Insert or update a text analysis cache entry.
*
* NOTE: The INSERT ... ON CONFLICT pattern is intentional — for new cache
* keys we always INSERT rather than checking existence first, so there is
* no TOCTOU race. The ON CONFLICT DO UPDATE handles the case where another
* worker inserted the same key between our check and our insert.
*/
export async function upsertCachedText(
text: string,
@@ -296,16 +291,12 @@ export async function deleteCachedMediaAnalysis(
/**
* Generate a deterministic cache key for a per-user moderation result.
*
* Format: text_mod:<userId>:<sha256(content).slice(0,16)>
* By including userId, two users sending the same text get separate
* cache entries so per-user context (reputation, username flags, etc.)
* is respected.
* Format: user_mod:<userId>:<sha256(content).slice(0,16)>
* 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 makeTextModerationCacheKey(content: string, userId?: string): string {
export function makeTextModerationCacheKey(content: string): string {
const hash = createHash("sha256").update(content).digest("hex").slice(0, 16);
if (userId) {
return `text_mod:${userId}:${hash}`;
}
return `text_mod:${hash}`;
}
@@ -1,7 +1,6 @@
import { resolve } from "node:dns/promises";
import { isIP } from "node:net";
import { createChildLogger } from "@bete/shared/logger";
import { createAbortTimeout } from "./abortHelper.js";
const log = createChildLogger("urlFetcher");
@@ -16,103 +15,52 @@ export interface FetchedUrlContext {
const MAX_FETCH_SIZE = 5 * 1024 * 1024; // 5 MB
const FETCH_TIMEOUT_MS = 8000;
const URL_REGEX = /https?:\/\/[^\s<]+[^<.,:;"')?\]\s]/gi;
// ═══════════════════════════════════════════════════════════════════════════════
// SSRF Protection with DNS Rebinding Defense
// ═══════════════════════════════════════════════════════════════════════════════
// Strategy: Resolve the hostname to IP addresses BEFORE fetching, then fetch
// directly from a pinned IP (using a Host header for virtual hosting).
// This prevents DNS rebinding where a domain alternates between a public IP
// and an internal IP (127.0.0.1, 10.x.x.x) between the check and the fetch.
//
// Edge cases handled:
// - No DNS records → reject (cannot fetch)
// - Multiple IPs (round-robin DNS) → pick first public one
// - All IPs are internal → reject
// - Direct IP literal → validate and pass through
// ═══════════════════════════════════════════════════════════════════════════════
interface PinnedAddress {
/** The original hostname from the URL (used in Host header) */
hostname: string;
/** The pinned, validated IP address to connect to (already vetted as safe) */
ip: string;
/** The port from the original URL */
port: string;
/** The protocol (http: or https:) */
protocol: string;
/** The pathname + search + hash (everything after host:port) */
path: string;
}
function isPrivateIP(ip: string): boolean {
return (
ip === "127.0.0.1" ||
ip === "::1" ||
ip === "0.0.0.0" ||
ip.startsWith("192.168.") ||
ip.startsWith("10.") ||
/^172\.(1[6-9]|2[0-9]|3[0-1])\./.test(ip) ||
ip.startsWith("169.254.") || // link-local
ip.startsWith("fc") || // IPv6 unique local (fc00::/7)
ip.startsWith("fd") // IPv6 unique local
);
}
const URL_REGEX = /https?:\/\/[^\s<]+[^<.,:;"')\]\s]/gi;
/**
* Resolve a hostname to a pinned IP address.
* Returns the first public IP found, or null if all resolved IPs are private.
* Also returns null if the host is a private IP literal.
*
* This function is the sole gate — once a safe IP is returned, the caller
* MUST use it directly without re-resolving the hostname.
* Basic SSRF protection.
* Note: A sophisticated attacker could still use DNS rebinding.
*/
async function resolveAndPinAddress(urlStr: string): Promise<PinnedAddress | null> {
async function isSafeUrl(urlStr: string): Promise<boolean> {
try {
const parsed = new URL(urlStr);
const host = parsed.hostname;
const protocol = parsed.protocol; // "http:" or "https:"
const port = parsed.port || (protocol === "https:" ? "443" : "80");
const path = parsed.pathname + parsed.search + parsed.hash;
// Block private IP literals immediately
if (isIP(host)) {
if (isPrivateIP(host)) return null;
// Direct public IP literal — can fetch directly
return { hostname: host, ip: host, port, protocol, path };
}
// Block obvious private hostnames
// Block obvious local IPs/hostnames
if (
host === "localhost" ||
host === "localhost.localdomain" ||
host.endsWith(".local") ||
host.endsWith(".internal")
host === "127.0.0.1" ||
host === "::1" ||
host.startsWith("192.168.") ||
host.startsWith("10.") ||
/^172\.(1[6-9]|2[0-9]|3[0-1])\./.test(host)
) {
return null;
return false;
}
// Resolve hostname to IP addresses
let addresses: string[];
try {
addresses = await resolve(host);
} catch {
// DNS resolution failed — can't verify safety
return null;
// Try resolving to check if it resolves to a local IP
if (!isIP(host)) {
try {
const addresses = await resolve(host);
for (const ip of addresses) {
if (
ip === "127.0.0.1" ||
ip.startsWith("192.168.") ||
ip.startsWith("10.") ||
/^172\.(1[6-9]|2[0-9]|3[0-1])\./.test(ip)
) {
return false;
}
}
} catch (err) {
// If DNS fails, we can't fetch it anyway
return false;
}
}
if (addresses.length === 0) return null;
// Pick the first non-private IP
const publicIp = addresses.find((ip) => !isPrivateIP(ip));
if (!publicIp) return null;
// We now have a pinned, verified safe IP.
// The caller MUST use this IP directly for the fetch.
return { hostname: host, ip: publicIp, port, protocol, path };
} catch {
return null;
return true;
} catch (err) {
return false;
}
}
@@ -160,27 +108,20 @@ export async function fetchUrlSafely(
return { url, type: "error", error: "Max redirect/meta depth reached" };
}
// Resolve + pin IP address FIRST (defence against DNS rebinding).
// The pinned IP is used directly — we never re-resolve the hostname.
const pinned = await resolveAndPinAddress(url);
if (!pinned) {
if (!(await isSafeUrl(url))) {
return { url, type: "error", error: "Unsafe URL blocked" };
}
// Reconstruct the URL using the pinned IP directly, keeping original Host
const pinnedUrl = `${pinned.protocol}//${pinned.ip}:${pinned.port}${pinned.path}`;
const { signal, cleanup } = createAbortTimeout(FETCH_TIMEOUT_MS);
const controller = new AbortController();
const timeoutId = setTimeout(() => controller.abort(), FETCH_TIMEOUT_MS);
try {
const response = await fetch(pinnedUrl, {
signal,
const response = await fetch(url, {
signal: controller.signal,
headers: {
"User-Agent":
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 DiscordBot/2.0",
Accept: "image/webp,image/apng,image/*,*/*;q=0.8",
// Use original hostname so virtual hosting still works
Host: pinned.hostname,
},
// Do not follow more than a few redirects natively, fetch handles up to 20 by default
});
@@ -249,7 +190,7 @@ export async function fetchUrlSafely(
error: err instanceof Error ? err.message : String(err),
};
} finally {
cleanup();
clearTimeout(timeoutId);
}
}
@@ -63,10 +63,11 @@ async function learnUserProfile(
.join("\n");
const prompt = `Anda adalah AI ahli psikologi, analisis perilaku online, dan pembaca karakter.
Tugas Anda adalah merangkum profil kepribadian SEORANG PRIBADI berdasarkan riwayat pesan-pesan mereka di server Discord.
Tugas Anda adalah merangkum profil kepribadian SEORANG PRIBADI — bukan sekadar statistik
gaya bicara — berdasarkan riwayat pesan-pesan mereka di server Discord.
Buatlah ringkasan yang KAYA AKAN PERSONALITAS sehingga pembaca merasa "mengenal" orang ini.
Pesan-pesan terakhir dari user (hanya pesan bersih/clean):
Pesan-pesan terakhir dari user "${userId}" (hanya pesan bersih/clean):
<messages>
${messagesText}
</messages>
@@ -121,18 +122,7 @@ atau konten SARA, itu akan SANGAT tidak sesuai dengan karakternya dan patut dicu
const text = completion.choices[0]?.message?.content?.trim();
if (!text) throw new Error("Empty response from LLM");
// Sanitize the AI-generated profile before saving to prevent
// prompt injection when the profile is later injected into prompts.
// Strip markdown code fences and XML special chars.
const sanitized = text
.replace(/```[\s\S]*?```/g, "")
.replace(/[<>&"']/g, (ch) => {
const entities: Record<string, string> = { "<": "&lt;", ">": "&gt;", "&": "&amp;", '"': "&quot;", "'": "&#39;" };
return entities[ch] || ch;
})
.trim();
await updateUserProfile(userId, guildId, sanitized);
await updateUserProfile(userId, guildId, text);
log.info(
{ userId, guildId },
"Successfully learned and updated user profile",