feat(moderation): refactor word analysis caching to text analysis caching with improved context handling
This commit is contained in:
MythEclipse
@@ -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;
|
||||
+16
-13
@@ -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;
|
||||
|
||||
@@ -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 = /<a?:([a-zA-Z0-9_]+):(\d+)>/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<string[]> {
|
||||
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<string[]> {
|
||||
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<string[]> {
|
||||
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<string>();
|
||||
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<string>();
|
||||
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<string>();
|
||||
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;
|
||||
})();
|
||||
|
||||
|
||||
@@ -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<TextCacheEntry | null> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<number> {
|
||||
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<string, number>;
|
||||
}> {
|
||||
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<string, number> = {};
|
||||
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: {} };
|
||||
}
|
||||
}
|
||||
@@ -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<Map<string, WordCacheEntry>> {
|
||||
if (words.length === 0) return new Map();
|
||||
|
||||
const results = new Map<string, WordCacheEntry>();
|
||||
|
||||
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<void> {
|
||||
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<number> {
|
||||
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<string, number>;
|
||||
}> {
|
||||
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<string, number> = {};
|
||||
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: {} };
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user