diff --git a/scripts/migrate-text-analysis-cache.sql b/scripts/migrate-text-analysis-cache.sql new file mode 100644 index 0000000..1ab9f18 --- /dev/null +++ b/scripts/migrate-text-analysis-cache.sql @@ -0,0 +1,23 @@ +-- Create text_analysis_cache table for DB-backed moderation analysis caching. +-- Full-text key preserves context so "kau" (clean) ≠ "awas kau" (harassment). + +CREATE TABLE IF NOT EXISTS text_analysis_cache ( + text TEXT PRIMARY KEY, + flags TEXT NOT NULL DEFAULT '[]', + source TEXT NOT NULL DEFAULT 'local' + CHECK (source IN ('local', 'nvidia', 'primary_ai', 'groq')), + analyzed_at BIGINT NOT NULL, + expires_at BIGINT NOT NULL, + hit_count INTEGER NOT NULL DEFAULT 0 +); + +CREATE INDEX IF NOT EXISTS idx_text_analysis_cache_expires_at + ON text_analysis_cache (expires_at); + +CREATE INDEX IF NOT EXISTS idx_text_analysis_cache_source + ON text_analysis_cache (source); + +-- Periodic cleanup: run every hour to remove expired entries. +-- This is optional; prune_expired_texts() in textCacheStore.ts also works. +-- COMMENT: Consider using pg_cron or an external scheduler for: +-- DELETE FROM text_analysis_cache WHERE expires_at < extract(epoch from now()) * 1000; diff --git a/src/database/schema.ts b/src/database/schema.ts index f2653df..8cda333 100644 --- a/src/database/schema.ts +++ b/src/database/schema.ts @@ -354,17 +354,20 @@ 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. + * Text Analysis Cache Table (PostgreSQL) + * Caches per-normalized-text moderation analysis results so repeated + * phrases reuse previously computed API / fallback results instead of + * re-calling expensive LLM or external moderation APIs. + * + * Uses the FULL normalized text (not per-word) because context matters: + * "kau" alone is clean, but "awas kau" can be a threat. */ -export const pgWordAnalysisCacheTable = pgTable( - "word_analysis_cache", +export const pgTextAnalysisCacheTable = pgTable( + "text_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"]). */ + /** Normalized text (lowercase, whitespace-collapsed) — primary key. */ + text: pgText("text").primaryKey(), + /** JSON array of moderation flags detected for this text (e.g. ["vulgar_language","harassment"]). */ flags: pgText("flags").notNull().default("[]"), /** Which source produced this result: "local" | "nvidia" | "primary_ai" | "groq". */ source: pgText("source", { @@ -376,14 +379,14 @@ export const pgWordAnalysisCacheTable = pgTable( 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. */ + /** How many times this cached text has been reused. */ hit_count: pgInteger("hit_count").notNull().default(0), }, (table) => ({ - expiresAtIdx: pgIndex("idx_word_analysis_cache_expires_at").on( + expiresAtIdx: pgIndex("idx_text_analysis_cache_expires_at").on( table.expires_at, ), - sourceIdx: pgIndex("idx_word_analysis_cache_source").on(table.source), + sourceIdx: pgIndex("idx_text_analysis_cache_source").on(table.source), }), ); @@ -399,7 +402,7 @@ export const voiceRecordingsTable = pgVoiceRecordingsTable; export const messageReviewsTable = pgMessageReviewsTable; export const moderationActionsTable = pgModerationActionsTable; export const retentionPoliciesTable = pgRetentionPoliciesTable; -export const wordAnalysisCacheTable = pgWordAnalysisCacheTable; +export const textAnalysisCacheTable = pgTextAnalysisCacheTable; // 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 155cc29..ccce45f 100644 --- a/src/moderation/indonesianTextNormalizer.ts +++ b/src/moderation/indonesianTextNormalizer.ts @@ -3,15 +3,10 @@ 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"; +import { getCachedText, upsertCachedText } from "./textCacheStore.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. */ @@ -67,9 +62,20 @@ const VALID_PRIMARY_AI_FLAGS = new Set([ "self_promo", ]); +/** + * In-memory cache TTL (10 min) — fastest path for repeated identical texts. + */ const BADWORD_CACHE_TTL_MS = 10 * 60 * 1000; + +/** + * DB cache TTL (24 hours) — survives restarts, stores full-text results + * so context is preserved (e.g. "kaus" is clean, "kau" alone is clean, + * but "awas kau" is harassment). + */ +const DB_CACHE_TTL_MS = 24 * 60 * 60 * 1000; + const NEMOTRON_RATE_LIMIT_COOLDOWN_MS = 60 * 1000; -const PRIMARY_AI_RATE_LIMIT_COOLDOWN_MS = 30 * 1000; +const PRIMARY_AI_RATE_LIMIT_COOLDOWN_MS = 30_000; const GROQ_RATE_LIMIT_COOLDOWN_MS = 60 * 1000; interface BadwordCacheEntry { @@ -459,8 +465,6 @@ async function callNemotronContentSafety(text: string): Promise { if (content) { const lowerContent = content.toLowerCase(); for (const category of NVIDIA_BAD_CATEGORIES) { - // Check if the category appears as a key in the response - // The Nemotron content safety model returns structured data with category scores if (lowerContent.includes(category)) { categories.push(CATEGORY_TO_BADWORD_LABEL[category] ?? category); } @@ -487,188 +491,149 @@ async function callNemotronContentSafety(text: string): Promise { return Array.from(new Set(categories)); } -/** - * 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(), - ); -} +// --------------------------------------------------------------------------- +// Three-tier cache pipeline +// --------------------------------------------------------------------------- /** - * Detect badwords in text using a two-tier cache strategy: + * Detect badwords in text using a **three-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. + * 2. **DB cache** (DB_CACHE_TTL_MS, 24 h) — same full-text key, persisted + * across restarts. Uses the FULL normalized text (not per-word) because + * context matters: "kau" alone is clean, but "awas kau" can be a threat. + * 3. **API/fallback pipeline** (NVIDIA → Primary AI → Groq → local lexical) + * only runs when both cache layers miss. */ export async function detectIndonesianBadwords( text: string, ): Promise { const cacheKey = normalizeBadwordCacheKey(text); + + // ── Tier 1: In-memory cache (fastest) ── const cached = getCachedBadwords(cacheKey); if (cached) { return cached; } + // De-duplicate concurrent lookups const inFlight = inFlightBadwordLookups.get(cacheKey); if (inFlight) { return inFlight; } const lookupPromise = (async () => { - const words = tokenizeToWords(text); - const uniqueWords = Array.from(new Set(words)); - - // ── 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); - } + // ── Tier 2: DB cache (survives restarts, preserves context) ── + const dbEntry = await getCachedText(cacheKey); + if (dbEntry) { + const flags = [...dbEntry.flags]; + setCachedBadwords(cacheKey, flags); // populate in-memory too + return flags; } - // ── Step 3: If all words are cached, return immediately ── - if (uncachedWords.length === 0) { - const finalHits = Array.from(cachedFlags); - setCachedBadwords(cacheKey, finalHits); - return finalHits; - } + // ── Tier 3: API / fallback pipeline ── - // ── 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); + // 3a. Local lexical check (instant, no network) + const localHits = detectLocalBadwords(text); if (localHits.length > 0) { - for (const hit of localHits) { - uncachedFlags.add(hit); - } + setCachedBadwords(cacheKey, localHits); + await upsertCachedText( + cacheKey, + localHits, + "local", + Date.now() + DB_CACHE_TTL_MS, + ); + return localHits; } - // 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. + const hits = new Set(); let sourceUsed: "local" | "nvidia" | "primary_ai" | "groq" = "local"; - 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) { + // 3b. 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 apiCategories = await callNemotronContentSafety(text); + for (const hit of apiCategories) { + hits.add(hit); + } + if (apiCategories.length > 0) sourceUsed = "nvidia"; + } 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", + ); + } + } + + // 3c. Try the main AI model next. + if (hits.size === 0 && Date.now() >= primaryAiUnavailableUntil) { + try { + const primaryHits = await callPrimaryAiModeration(text); + for (const hit of primaryHits) { + hits.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", + ); + } + } + + // 3d. Try Groq Llama Prompt Guard as final API fallback. + if (hits.size === 0 && Date.now() >= groqUnavailableUntil) { + const groqKey = config.GROQ_API_KEY; + if (groqKey) { try { - const apiCategories = await callNemotronContentSafety(uncachedText); - for (const hit of apiCategories) { - uncachedFlags.add(hit); + const groqHits = await callGrokModeration(text); + for (const hit of groqHits) { + hits.add(hit); } - sourceUsed = "nvidia"; + if (groqHits.length > 0) sourceUsed = "groq"; } catch (error) { const status = axios.isAxiosError(error) ? error.response?.status : null; if (status === 429) { - nemotronUnavailableUntil = - Date.now() + NEMOTRON_RATE_LIMIT_COOLDOWN_MS; + groqUnavailableUntil = Date.now() + GROQ_RATE_LIMIT_COOLDOWN_MS; } log.warn( { error }, - "NVIDIA Nemotron API call failed, falling back to primary AI then local detection", + "Groq Llama Prompt Guard moderation failed, falling back to 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", - ); - } - } - } } - // ── 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, - }); - } + const finalHits = Array.from(hits); - if (newWordEntries.length > 0) { - await upsertCachedWords(newWordEntries); - } - - // ── Step 6: Merge cached + uncached flags ── - const finalHits = Array.from(new Set([...cachedFlags, ...uncachedFlags])); + // Populate all cache tiers so the same text never triggers another API call + // within the TTL window. setCachedBadwords(cacheKey, finalHits); + await upsertCachedText( + cacheKey, + finalHits, + sourceUsed, + Date.now() + DB_CACHE_TTL_MS, + ); + return finalHits; })(); diff --git a/src/moderation/textCacheStore.ts b/src/moderation/textCacheStore.ts new file mode 100644 index 0000000..fe99802 --- /dev/null +++ b/src/moderation/textCacheStore.ts @@ -0,0 +1,151 @@ +import { executeAll, executeGet } from "../database/drizzle.js"; +import { createChildLogger } from "../logger.js"; + +const logger = createChildLogger("text-cache-store"); + +export interface TextCacheEntry { + text: string; + flags: string[]; + source: "local" | "nvidia" | "primary_ai" | "groq"; + analyzed_at: number; + expires_at: number; + hit_count: number; +} + +/** + * Lookup cached analysis result for a normalized text string. + * Returns null if not found or expired. + */ +export async function getCachedText( + text: string, +): Promise { + try { + const row = await executeGet( + `SELECT text, flags, source, analyzed_at, expires_at, hit_count + FROM text_analysis_cache + WHERE text = $1 AND expires_at > $2`, + [text, Date.now()], + ); + + if (!row) return null; + + return { + text: row.text, + flags: JSON.parse(row.flags), + source: row.source, + analyzed_at: row.analyzed_at, + expires_at: row.expires_at, + hit_count: row.hit_count, + }; + } catch (error) { + logger.error( + { error: error instanceof Error ? error.message : String(error) }, + "Failed to get cached text", + ); + return null; + } +} + +/** + * Insert or update a text analysis cache entry. + */ +export async function upsertCachedText( + text: string, + flags: string[], + source: "local" | "nvidia" | "primary_ai" | "groq", + expiresAt: number, +): Promise { + const now = Date.now(); + + try { + await executeAll( + `INSERT INTO text_analysis_cache (text, flags, source, analyzed_at, expires_at, hit_count) + VALUES ($1, $2, $3, $4, $5, 0) + ON CONFLICT (text) DO UPDATE SET + flags = EXCLUDED.flags, + source = EXCLUDED.source, + analyzed_at = EXCLUDED.analyzed_at, + expires_at = EXCLUDED.expires_at`, + [text, JSON.stringify(flags), source, now, expiresAt], + ); + } catch (error) { + logger.error( + { error: error instanceof Error ? error.message : String(error) }, + "Failed to upsert cached text", + ); + } +} + +/** + * Increment hit count for a cached text entry (called on cache hit). + */ +export async function incrementTextCacheHit(text: string): Promise { + try { + await executeAll( + `UPDATE text_analysis_cache SET hit_count = hit_count + 1 WHERE text = $1`, + [text], + ); + } catch (error) { + // Silent fail — this is just a counter, not critical + } +} + +/** + * Delete expired cache entries. Run periodically to keep the table clean. + */ +export async function pruneExpiredTexts(): Promise { + try { + const result = await executeAll( + `DELETE FROM text_analysis_cache WHERE expires_at < $1`, + [Date.now()], + ); + return (result as any).rowCount ?? 0; + } catch (error) { + logger.error( + { error: error instanceof Error ? error.message : String(error) }, + "Failed to prune expired texts", + ); + return 0; + } +} + +/** + * Get cache statistics for observability. + */ +export async function getTextCacheStats(): 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 text_analysis_cache`), + executeAll( + `SELECT count(*) as cnt FROM text_analysis_cache WHERE expires_at < $1`, + [now], + ), + executeAll( + `SELECT source, count(*) as cnt FROM text_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 text cache stats", + ); + return { total: 0, expired: 0, bySource: {} }; + } +} diff --git a/src/moderation/wordCacheStore.ts b/src/moderation/wordCacheStore.ts deleted file mode 100644 index cd9eed1..0000000 --- a/src/moderation/wordCacheStore.ts +++ /dev/null @@ -1,182 +0,0 @@ -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: {} }; - } -}