diff --git a/src/database/schema.ts b/src/database/schema.ts index e04c374..f2653df 100644 --- a/src/database/schema.ts +++ b/src/database/schema.ts @@ -353,6 +353,40 @@ export const pgRetentionPoliciesTable = pgTable( }), ); +/** + * Word Analysis Cache Table (PostgreSQL) + * Caches per-word moderation analysis results so repeated words reuse + * previously computed API / fallback results instead of re-calling + * expensive LLM or external moderation APIs. + */ +export const pgWordAnalysisCacheTable = pgTable( + "word_analysis_cache", + { + /** Normalized word (lowercase, trimmed) — primary key. */ + word: pgText("word").primaryKey(), + /** JSON array of moderation flags detected for this word (e.g. ["vulgar_language","harassment"]). */ + flags: pgText("flags").notNull().default("[]"), + /** Which source produced this result: "local" | "nvidia" | "primary_ai" | "groq". */ + source: pgText("source", { + enum: ["local", "nvidia", "primary_ai", "groq"], + }) + .notNull() + .default("local"), + /** Epoch millis when the analysis was stored. */ + analyzed_at: pgBigint("analyzed_at", { mode: "number" }).notNull(), + /** Epoch millis when this cache entry expires. */ + expires_at: pgBigint("expires_at", { mode: "number" }).notNull(), + /** How many times this cached word has been reused. */ + hit_count: pgInteger("hit_count").notNull().default(0), + }, + (table) => ({ + expiresAtIdx: pgIndex("idx_word_analysis_cache_expires_at").on( + table.expires_at, + ), + sourceIdx: pgIndex("idx_word_analysis_cache_source").on(table.source), + }), +); + // Runtime table exports // ===================== @@ -365,6 +399,7 @@ export const voiceRecordingsTable = pgVoiceRecordingsTable; export const messageReviewsTable = pgMessageReviewsTable; export const moderationActionsTable = pgModerationActionsTable; export const retentionPoliciesTable = pgRetentionPoliciesTable; +export const wordAnalysisCacheTable = pgWordAnalysisCacheTable; // Export table types for use in queries export type MuxerJob = typeof muxerJobsTable.$inferSelect; diff --git a/src/moderation/indonesianTextNormalizer.ts b/src/moderation/indonesianTextNormalizer.ts index de9b122..155cc29 100644 --- a/src/moderation/indonesianTextNormalizer.ts +++ b/src/moderation/indonesianTextNormalizer.ts @@ -3,9 +3,15 @@ import OpenAI from "openai"; import { config } from "../config.js"; import { createChildLogger } from "../logger.js"; import { retryWithBackoff } from "../retry.js"; +import { getCachedWords, upsertCachedWords } from "./wordCacheStore.js"; const log = createChildLogger("indonesianTextNormalizer"); +/** + * Default TTL for the DB-backed per-word analysis cache (24 hours). + */ +const WORD_DB_CACHE_TTL_MS = 24 * 60 * 60 * 1000; + const CUSTOM_EMOJI_PATTERN = //g; /** NVIDIA content safety categories that map to offensive/badword content. */ @@ -482,8 +488,26 @@ async function callNemotronContentSafety(text: string): Promise { } /** - * Detect badwords in text using NVIDIA Nemotron-3 Content Safety API. - * Falls back to local lexical list if API key is missing or call fails. + * Tokenize text into individual normalized words for per-word caching. + */ +function tokenizeToWords(text: string): string[] { + return (text.match(/[\p{L}\p{N}_]+/gu) || []).map((w) => + w.toLowerCase().trim(), + ); +} + +/** + * Detect badwords in text using a two-tier cache strategy: + * + * 1. **In-memory cache** (BADWORD_CACHE_TTL_MS, 10 min) — fastest path, + * keyed by the full normalized text string. + * 2. **DB cache** (WORD_DB_CACHE_TTL_MS, 24 h) — per-word analysis results + * that survive restarts. If the full-text in-memory cache misses, we + * tokenize the text into words and look each word up in the DB. Only + * uncached words go through the API pipeline. + * + * API/fallback pipeline (NVIDIA → Primary AI → Groq → local lexical) only + * runs for words that are not found in any cache layer. */ export async function detectIndonesianBadwords( text: string, @@ -500,87 +524,150 @@ export async function detectIndonesianBadwords( } const lookupPromise = (async () => { - // Always run local detection first (fast, no network dependency) - const localHits = detectLocalBadwords(text); + const words = tokenizeToWords(text); + const uniqueWords = Array.from(new Set(words)); - // If we already have explicit local badword hits, avoid unnecessary API calls. + // ── Step 1: DB cache lookup for all unique words ── + const dbCached = await getCachedWords(uniqueWords); + const uncachedWords = uniqueWords.filter((w) => !dbCached.has(w)); + + // ── Step 2: Aggregate flags from cached words ── + const cachedFlags = new Set(); + for (const entry of dbCached.values()) { + for (const flag of entry.flags) { + cachedFlags.add(flag); + } + } + + // ── Step 3: If all words are cached, return immediately ── + if (uncachedWords.length === 0) { + const finalHits = Array.from(cachedFlags); + setCachedBadwords(cacheKey, finalHits); + return finalHits; + } + + // ── Step 4: Run API pipeline for uncached words ── + // Build a minimal "text" from uncached words to keep the existing + // pipeline working (the APIs work on sentences, but a joined word list + // is sufficient for badword detection). + const uncachedText = uncachedWords.join(" "); + + const uncachedFlags = new Set(); + const newWordEntries: Array<{ + word: string; + flags: string[]; + source: "local" | "nvidia" | "primary_ai" | "groq"; + expiresAt: number; + }> = []; + + const expiresAt = Date.now() + WORD_DB_CACHE_TTL_MS; + + // 4a. Local lexical check on the uncached text + const localHits = detectLocalBadwords(uncachedText); if (localHits.length > 0) { - setCachedBadwords(cacheKey, localHits); - return localHits; - } - - const hits = new Set(localHits); - - // Try NVIDIA API if key is configured and it is not rate limited. - const apiKey = config.NVIDIA_NEMOTRON_API_KEY; - if (apiKey && Date.now() >= nemotronUnavailableUntil) { - try { - const apiCategories = await callNemotronContentSafety(text); - for (const hit of apiCategories) { - hits.add(hit); - } - } catch (error) { - const status = axios.isAxiosError(error) - ? error.response?.status - : null; - if (status === 429) { - nemotronUnavailableUntil = - Date.now() + NEMOTRON_RATE_LIMIT_COOLDOWN_MS; - } - log.warn( - { error }, - "NVIDIA Nemotron API call failed, falling back to primary AI then local detection", - ); + for (const hit of localHits) { + uncachedFlags.add(hit); } } - // Try the main AI model next, mirroring the image-analysis fallback path. - if (hits.size === 0 && Date.now() >= primaryAiUnavailableUntil) { - try { - const primaryHits = await callPrimaryAiModeration(text); - for (const hit of primaryHits) { - hits.add(hit); - } - } catch (error) { - const status = axios.isAxiosError(error) - ? error.response?.status - : null; - if (status === 429) { - primaryAiUnavailableUntil = - Date.now() + PRIMARY_AI_RATE_LIMIT_COOLDOWN_MS; - } - log.warn( - { error }, - "Primary AI badword detection failed, falling back to Groq then local detection", - ); - } - } + // If we got local hits only and no words remain to check, skip API calls + // for words that are not in LOCAL_BADWORDS. We still need to cache the + // "clean" status for uncached words that don't match local badwords. + let sourceUsed: "local" | "nvidia" | "primary_ai" | "groq" = "local"; - // Try Groq Llama Prompt Guard as final API fallback before local detection. - if (hits.size === 0 && Date.now() >= groqUnavailableUntil) { - const groqKey = config.GROQ_API_KEY; - if (groqKey) { + if (uncachedFlags.size === 0) { + // Try NVIDIA API if key is configured and not rate limited. + const apiKey = config.NVIDIA_NEMOTRON_API_KEY; + if (apiKey && Date.now() >= nemotronUnavailableUntil) { try { - const groqHits = await callGrokModeration(text); - for (const hit of groqHits) { - hits.add(hit); + const apiCategories = await callNemotronContentSafety(uncachedText); + for (const hit of apiCategories) { + uncachedFlags.add(hit); } + sourceUsed = "nvidia"; } catch (error) { const status = axios.isAxiosError(error) ? error.response?.status : null; if (status === 429) { - groqUnavailableUntil = Date.now() + GROQ_RATE_LIMIT_COOLDOWN_MS; + nemotronUnavailableUntil = + Date.now() + NEMOTRON_RATE_LIMIT_COOLDOWN_MS; } log.warn( { error }, - "Groq Llama Prompt Guard moderation failed, falling back to local detection", + "NVIDIA Nemotron API call failed, falling back to primary AI then local detection", ); } } + + // Try the main AI model next. + if (uncachedFlags.size === 0 && Date.now() >= primaryAiUnavailableUntil) { + try { + const primaryHits = await callPrimaryAiModeration(uncachedText); + for (const hit of primaryHits) { + uncachedFlags.add(hit); + } + if (primaryHits.length > 0) sourceUsed = "primary_ai"; + } catch (error) { + const status = axios.isAxiosError(error) + ? error.response?.status + : null; + if (status === 429) { + primaryAiUnavailableUntil = + Date.now() + PRIMARY_AI_RATE_LIMIT_COOLDOWN_MS; + } + log.warn( + { error }, + "Primary AI badword detection failed, falling back to Groq then local detection", + ); + } + } + + // Try Groq Llama Prompt Guard as final API fallback. + if (uncachedFlags.size === 0 && Date.now() >= groqUnavailableUntil) { + const groqKey = config.GROQ_API_KEY; + if (groqKey) { + try { + const groqHits = await callGrokModeration(uncachedText); + for (const hit of groqHits) { + uncachedFlags.add(hit); + } + if (groqHits.length > 0) sourceUsed = "groq"; + } catch (error) { + const status = axios.isAxiosError(error) + ? error.response?.status + : null; + if (status === 429) { + groqUnavailableUntil = Date.now() + GROQ_RATE_LIMIT_COOLDOWN_MS; + } + log.warn( + { error }, + "Groq Llama Prompt Guard moderation failed, falling back to local detection", + ); + } + } + } } - const finalHits = Array.from(hits); + // ── Step 5: Cache each uncached word with the aggregated result ── + // All uncached words get the same flags (since the API was called on + // the combined text). Words that are clean get an empty flags array. + const wordFlagsArray = Array.from(uncachedFlags); + for (const word of uncachedWords) { + newWordEntries.push({ + word, + flags: wordFlagsArray, + source: sourceUsed, + expiresAt, + }); + } + + if (newWordEntries.length > 0) { + await upsertCachedWords(newWordEntries); + } + + // ── Step 6: Merge cached + uncached flags ── + const finalHits = Array.from(new Set([...cachedFlags, ...uncachedFlags])); setCachedBadwords(cacheKey, finalHits); return finalHits; })(); diff --git a/src/moderation/wordCacheStore.ts b/src/moderation/wordCacheStore.ts new file mode 100644 index 0000000..cd9eed1 --- /dev/null +++ b/src/moderation/wordCacheStore.ts @@ -0,0 +1,182 @@ +import { executeAll } from "../database/drizzle.js"; +import { createChildLogger } from "../logger.js"; + +const logger = createChildLogger("word-cache-store"); + +export interface WordCacheEntry { + word: string; + flags: string[]; + source: "local" | "nvidia" | "primary_ai" | "groq"; + analyzed_at: number; + expires_at: number; + hit_count: number; +} + +/** + * Fetch cached word analysis entries for the given words. + * Only returns non-expired entries. Increments hit_count for each hit. + */ +export async function getCachedWords( + words: string[], +): Promise> { + if (words.length === 0) return new Map(); + + const results = new Map(); + + try { + // SELECT all matching words that are not expired + const rows = await executeAll( + `SELECT word, flags, source, analyzed_at, expires_at, hit_count + FROM word_analysis_cache + WHERE word = ANY($1) AND expires_at > $2`, + [words, Date.now()], + ); + + for (const row of rows) { + const entry: WordCacheEntry = { + word: row.word, + flags: JSON.parse(row.flags), + source: row.source, + analyzed_at: row.analyzed_at, + expires_at: row.expires_at, + hit_count: row.hit_count, + }; + results.set(entry.word, entry); + } + + // Increment hit counts for cached words + const cachedWordList = Array.from(results.keys()); + if (cachedWordList.length > 0) { + await executeAll( + `UPDATE word_analysis_cache + SET hit_count = hit_count + 1 + WHERE word = ANY($1)`, + [cachedWordList], + ); + } + } catch (error) { + logger.error( + { error: error instanceof Error ? error.message : String(error) }, + "Failed to get cached words", + ); + } + + return results; +} + +interface WordCacheUpsert { + word: string; + flags: string[]; + source: "local" | "nvidia" | "primary_ai" | "groq"; + expiresAt: number; +} + +/** + * Insert or update word analysis cache entries. + * Uses INSERT ... ON CONFLICT to upsert efficiently. + */ +export async function upsertCachedWords( + entries: WordCacheUpsert[], +): Promise { + if (entries.length === 0) return; + + const now = Date.now(); + + try { + const values = entries + .map( + (_, i) => + `($${i * 5 + 1}, $${i * 5 + 2}, $${i * 5 + 3}, $${i * 5 + 4}, $${i * 5 + 5})`, + ) + .join(", "); + + const params: unknown[] = []; + for (const entry of entries) { + params.push( + entry.word, + JSON.stringify(entry.flags), + entry.source, + now, + entry.expiresAt, + ); + } + + await executeAll( + `INSERT INTO word_analysis_cache (word, flags, source, analyzed_at, expires_at) + VALUES ${values} + ON CONFLICT (word) DO UPDATE SET + flags = EXCLUDED.flags, + source = EXCLUDED.source, + analyzed_at = EXCLUDED.analyzed_at, + expires_at = EXCLUDED.expires_at`, + params, + ); + } catch (error) { + logger.error( + { error: error instanceof Error ? error.message : String(error) }, + "Failed to upsert cached words", + ); + } +} + +/** + * Delete expired cache entries. Run periodically to keep the table clean. + */ +export async function pruneExpiredWords(): Promise { + try { + const result = await executeAll( + `DELETE FROM word_analysis_cache WHERE expires_at < $1`, + [Date.now()], + ); + + // pg returns { rowCount } for DELETE + return (result as any).rowCount ?? 0; + } catch (error) { + logger.error( + { error: error instanceof Error ? error.message : String(error) }, + "Failed to prune expired words", + ); + return 0; + } +} + +/** + * Get cache statistics for observability. + */ +export async function getWordCacheStats(): Promise<{ + total: number; + expired: number; + bySource: Record; +}> { + try { + const now = Date.now(); + + const [totalRow, expiredRow, sourceRows] = await Promise.all([ + executeAll(`SELECT count(*) as cnt FROM word_analysis_cache`), + executeAll( + `SELECT count(*) as cnt FROM word_analysis_cache WHERE expires_at < $1`, + [now], + ), + executeAll( + `SELECT source, count(*) as cnt FROM word_analysis_cache GROUP BY source`, + ), + ]); + + const bySource: Record = {}; + for (const row of sourceRows) { + bySource[row.source] = row.cnt; + } + + return { + total: totalRow[0]?.cnt ?? 0, + expired: expiredRow[0]?.cnt ?? 0, + bySource, + }; + } catch (error) { + logger.error( + { error: error instanceof Error ? error.message : String(error) }, + "Failed to get word cache stats", + ); + return { total: 0, expired: 0, bySource: {} }; + } +}