From 406a6cbf79560194ce05888fa266947ae2cdc737 Mon Sep 17 00:00:00 2001 From: MythEclipse Date: Tue, 2 Jun 2026 14:05:51 +0700 Subject: [PATCH] refactor: migrate sticker cache to PostgreSQL, remove versioning MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Replace file-based sticker cache (.dat + index.json) with PostgreSQL sticker_cache table - Remove model_version column and all versioning logic (git branch detection, GitHub API, CACHE_MODEL_VERSION) - Strip VISION_MODEL_VERSION, logCacheEvent calls, and version filtering from SQL queries - Remove STICKER_CACHE_DIR and STICKER_CACHE_MAX_SIZE_MB config variables - Delete dead migration add_model_version_to_cache.sql textCacheStore.ts reduced 482→204 lines (-57%) stickerCache.ts reduced 210→144 lines (-31%) Total: 11 files changed, 233 insertions, 591 deletions --- .../ai-moderation/llmModerationClient.ts | 30 +- .../modules/ai-moderation/responseLogger.ts | 10 +- .../src/modules/ai-moderation/stickerCache.ts | 199 +++++-------- .../modules/ai-moderation/textCacheStore.ts | 278 ++---------------- .../src/shared/config/config.ts | 2 - .../migrations/add_model_version_to_cache.sql | 24 -- .../src/shared/database/schema.ts | 42 ++- src/config.ts | 2 - src/database/schema.ts | 31 ++ src/moderation/llmModerationClient.ts | 7 +- src/moderation/stickerCache.ts | 199 +++++-------- 11 files changed, 233 insertions(+), 591 deletions(-) delete mode 100644 services/discord-gateway/src/shared/database/migrations/add_model_version_to_cache.sql diff --git a/services/discord-gateway/src/modules/ai-moderation/llmModerationClient.ts b/services/discord-gateway/src/modules/ai-moderation/llmModerationClient.ts index c2910e3..f42981c 100644 --- a/services/discord-gateway/src/modules/ai-moderation/llmModerationClient.ts +++ b/services/discord-gateway/src/modules/ai-moderation/llmModerationClient.ts @@ -14,6 +14,7 @@ import type { import { withLlmConcurrency } from "./concurrencyLimiter.js"; import { formatModerationTextEvidenceForPrompt } from "./indonesianTextNormalizer.js"; import { buildSystemPrompt as buildSystemPromptModular } from "./moderationPrompt.js"; +import { logModerationAnalysis, logModerationError } from "./responseLogger.js"; import { getStickerFromCache, initStickerCache, @@ -32,14 +33,8 @@ import { makeImageCacheKey, makeStickerCacheKey, upsertCachedMediaAnalysis, - VISION_MODEL_VERSION, } from "./textCacheStore.js"; import { extractUrlsFromText, fetchUrlSafely } from "./urlFetcher.js"; -import { - logCacheEvent, - logModerationAnalysis, - logModerationError, -} from "./responseLogger.js"; const SeveritySchema = z.enum(["none", "low", "medium", "high", "critical"]); const RecommendedActionSchema = z.enum([ @@ -793,11 +788,16 @@ async function callModerationLLM( ); // Log error with responseLogger - logModerationError(targetIds, config.AI_LLM_MODEL, parseError as Error | string, { - phase: "parse_response", - label, - contentLength: state.lastInvalidContent.length, - }); + logModerationError( + targetIds, + config.AI_LLM_MODEL, + parseError as Error | string, + { + phase: "parse_response", + label, + contentLength: state.lastInvalidContent.length, + }, + ); // Sanitized error messages — no internal details exposed (R10) const errorCode = `MOD_${Date.now().toString(36).slice(0, 6)}`; @@ -996,10 +996,7 @@ async function _runSingleMediaAnalysis( ): Promise<{ results: AnalysisResult[]; raw: unknown }> { // Lazy init sticker cache if (!isStickerCacheReady()) { - await initStickerCache({ - cacheDir: config.STICKER_CACHE_DIR, - maxSizeBytes: config.STICKER_CACHE_MAX_SIZE_MB * 1024 * 1024, - }).catch((err: unknown) => { + await initStickerCache().catch((err: unknown) => { log.warn( { error: err instanceof Error ? err.message : String(err) }, "Sticker cache init failed — continuing without cache", @@ -1040,7 +1037,6 @@ async function _runSingleMediaAnalysis( { attachmentId: att.id, cacheKey: attVisionKey }, "Vision cache HIT for attachment — skipped download", ); - logCacheEvent("hit", attVisionKey, "media", VISION_MODEL_VERSION); const sourceLabel = `[gambar di atas adalah attachment ${att.filename} dari pesan id=${att.message_id}]`; const analysisText = `[Media analysis for message ${att.message_id}] ${sourceLabel}: ${cachedVision}`; const existing = mediaAnalysisMap.get(targetId) ?? []; @@ -1049,8 +1045,6 @@ async function _runSingleMediaAnalysis( return; } - logCacheEvent("miss", attVisionKey, "media", VISION_MODEL_VERSION); - const controller = new AbortController(); const timeoutId = setTimeout(() => controller.abort(), 15000); diff --git a/services/discord-gateway/src/modules/ai-moderation/responseLogger.ts b/services/discord-gateway/src/modules/ai-moderation/responseLogger.ts index b817794..8423bf1 100644 --- a/services/discord-gateway/src/modules/ai-moderation/responseLogger.ts +++ b/services/discord-gateway/src/modules/ai-moderation/responseLogger.ts @@ -43,7 +43,6 @@ export interface CacheHitEvent { type: "hit" | "miss"; cacheKey: string; source: "text" | "media" | "sticker"; - modelVersion: string; timestamp: number; } @@ -154,27 +153,22 @@ export function logCacheEvent( type: "hit" | "miss", cacheKey: string, source: "text" | "media" | "sticker", - modelVersion: string, ): void { const event: CacheHitEvent = { type, cacheKey, source, - modelVersion, timestamp: Date.now(), }; - const level = type === "hit" ? "debug" : "debug"; - logger.log( - { level }, + logger.debug( { cache_type: type.toUpperCase(), source, key_length: cacheKey.length, key_preview: cacheKey.substring(0, 50), - model_version: modelVersion, }, - `Cache ${type.toUpperCase()}: ${source} (version: ${modelVersion})`, + `Cache ${type.toUpperCase()}: ${source}`, ); } diff --git a/services/discord-gateway/src/modules/ai-moderation/stickerCache.ts b/services/discord-gateway/src/modules/ai-moderation/stickerCache.ts index b577bd7..3594a68 100644 --- a/services/discord-gateway/src/modules/ai-moderation/stickerCache.ts +++ b/services/discord-gateway/src/modules/ai-moderation/stickerCache.ts @@ -1,9 +1,14 @@ -import { mkdir, readFile, unlink, writeFile } from "node:fs/promises"; -import { join } from "node:path"; +import { executeAll, executeGet } from "../../shared/database/drizzle.js"; import { createChildLogger } from "../../shared/logger/logger.js"; const logger = createChildLogger("sticker-cache"); +const TTL_MS = 7 * 24 * 60 * 60 * 1000; // 7 days +const MAX_SIZE_BYTES = 100 * 1024 * 1024; // 100MB hardcoded + +let ready = false; +let statsCache = { entryCount: 0, totalSizeBytes: 0 }; + export interface StickerCacheEntry { base64: string; mimeType: string; @@ -11,87 +16,38 @@ export interface StickerCacheEntry { size: number; } -interface CacheIndexEntry { - file: string; - mimeType: string; - size: number; - fetchedAt: number; -} - -interface CacheIndex { - entries: Record; - totalSizeBytes: number; -} - -export interface StickerCacheOptions { - cacheDir: string; - maxSizeBytes: number; - ttlMs?: number; -} - -let cacheDir = ""; -let maxSizeBytes = 0; -let ttlMs = 7 * 24 * 60 * 60 * 1000; // 7 days default -let index: CacheIndex = { entries: {}, totalSizeBytes: 0 }; -let ready = false; - function sanitizeKey(name: string): string { return encodeURIComponent(name).replace(/%/g, "_"); } -async function loadIndex(): Promise { - try { - const raw = await readFile(join(cacheDir, "index.json"), "utf-8"); - return JSON.parse(raw) as CacheIndex; - } catch { - return { entries: {}, totalSizeBytes: 0 }; - } -} - -async function saveIndex(idx: CacheIndex): Promise { - await writeFile( - join(cacheDir, "index.json"), - JSON.stringify(idx, null, 2), - "utf-8", - ); -} - /** - * Initialise the sticker cache: create directory, load index. + * Initialise the sticker cache from PostgreSQL. * Idempotent — safe to call multiple times. */ -export async function initStickerCache( - opts: StickerCacheOptions, -): Promise { +export async function initStickerCache(): Promise { if (ready) return; - cacheDir = opts.cacheDir; - maxSizeBytes = opts.maxSizeBytes; - ttlMs = opts.ttlMs ?? 7 * 24 * 60 * 60 * 1000; - - await mkdir(cacheDir, { recursive: true }); - index = await loadIndex(); - - // Prune expired entries on startup - const now = Date.now(); - let changed = false; - for (const [key, meta] of Object.entries(index.entries)) { - if (now - meta.fetchedAt > ttlMs) { - await unlink(join(cacheDir, meta.file)).catch(() => {}); - index.totalSizeBytes -= meta.size; - delete index.entries[key]; - changed = true; + try { + await executeAll( + "DELETE FROM sticker_cache WHERE fetched_at < ?", + [Date.now() - TTL_MS], + ); + const row = await executeGet( + "SELECT count(*) as cnt, COALESCE(SUM(size), 0) as total FROM sticker_cache", + ); + if (row) { + statsCache = { + entryCount: Number(row.cnt), + totalSizeBytes: Number(row.total), + }; } + } catch (err) { + logger.warn( + { error: String(err) }, + "Failed to prune expired stickers on init", + ); } - if (changed) await saveIndex(index); - ready = true; - logger.info( - { - entryCount: Object.keys(index.entries).length, - totalSizeBytes: index.totalSizeBytes, - }, - "Sticker cache initialized", - ); + logger.info(statsCache, "Sticker cache initialized (PostgreSQL)"); } /** @@ -101,32 +57,24 @@ export async function getStickerFromCache( stickerName: string, ): Promise { if (!ready) return null; - const key = sanitizeKey(stickerName); - const meta = index.entries[key]; - if (!meta) return null; - - // TTL check - if (Date.now() - meta.fetchedAt > ttlMs) { - await unlink(join(cacheDir, meta.file)).catch(() => {}); - index.totalSizeBytes -= meta.size; - delete index.entries[key]; - await saveIndex(index); - return null; - } - try { - const raw = await readFile(join(cacheDir, meta.file), "utf-8"); + const row = await executeGet( + "SELECT base64, mime_type, size, fetched_at FROM sticker_cache WHERE name = ? AND fetched_at > ?", + [key, Date.now() - TTL_MS], + ); + if (!row) return null; return { - base64: raw, - mimeType: meta.mimeType, - fetchedAt: meta.fetchedAt, - size: meta.size, + base64: row.base64, + mimeType: row.mime_type, + fetchedAt: Number(row.fetched_at), + size: Number(row.size), }; - } catch { - // File missing — clean up index entry - delete index.entries[key]; - await saveIndex(index); + } catch (err) { + logger.error( + { error: String(err), stickerName }, + "Failed to get sticker from cache", + ); return null; } } @@ -140,52 +88,44 @@ export async function setStickerInCache( mimeType: string, ): Promise { if (!ready) return; - const key = sanitizeKey(stickerName); - const fileName = `${key}.dat`; const size = Buffer.byteLength(base64, "utf-8"); - - // Evict if needed - await evictIfNeeded(size); - + const now = Date.now(); try { - await writeFile(join(cacheDir, fileName), base64, "utf-8"); - index.entries[key] = { - file: fileName, - mimeType, - size, - fetchedAt: Date.now(), - }; - index.totalSizeBytes += size; - await saveIndex(index); + await evictIfNeeded(size); + await executeAll( + `INSERT INTO sticker_cache (name, base64, mime_type, size, fetched_at) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT (name) DO UPDATE SET + base64 = EXCLUDED.base64, + mime_type = EXCLUDED.mime_type, + size = EXCLUDED.size, + fetched_at = EXCLUDED.fetched_at`, + [key, base64, mimeType, size, now], + ); + statsCache.entryCount++; + statsCache.totalSizeBytes += size; logger.debug({ stickerName, size }, "Sticker cached"); } catch (err) { logger.warn( - { stickerName, error: err instanceof Error ? err.message : String(err) }, + { stickerName, error: String(err) }, "Failed to write sticker to cache", ); } } async function evictIfNeeded(newSize: number): Promise { - while (index.totalSizeBytes + newSize > maxSizeBytes) { - // Find oldest entry - let oldestKey: string | null = null; - let oldestTime = Infinity; - for (const [key, meta] of Object.entries(index.entries)) { - if (meta.fetchedAt < oldestTime) { - oldestTime = meta.fetchedAt; - oldestKey = key; - } - } - if (!oldestKey) break; - - const meta = index.entries[oldestKey]; - await unlink(join(cacheDir, meta.file)).catch(() => {}); - index.totalSizeBytes -= meta.size; - delete index.entries[oldestKey]; + while (statsCache.totalSizeBytes + newSize > MAX_SIZE_BYTES) { + const oldest = await executeGet( + "SELECT name, size FROM sticker_cache ORDER BY fetched_at ASC LIMIT 1", + ); + if (!oldest) break; + await executeAll("DELETE FROM sticker_cache WHERE name = ?", [ + oldest.name, + ]); + statsCache.totalSizeBytes -= Number(oldest.size); + statsCache.entryCount--; } - await saveIndex(index); } /** @@ -195,10 +135,7 @@ export function getStickerCacheStats(): { entryCount: number; totalSizeBytes: number; } { - return { - entryCount: Object.keys(index.entries).length, - totalSizeBytes: index.totalSizeBytes, - }; + return { ...statsCache }; } /** diff --git a/services/discord-gateway/src/modules/ai-moderation/textCacheStore.ts b/services/discord-gateway/src/modules/ai-moderation/textCacheStore.ts index 7f6027b..9b641bd 100644 --- a/services/discord-gateway/src/modules/ai-moderation/textCacheStore.ts +++ b/services/discord-gateway/src/modules/ai-moderation/textCacheStore.ts @@ -1,247 +1,9 @@ import { createHash } from "node:crypto"; -import { execFileSync } from "node:child_process"; import { executeAll, executeGet } from "../../shared/database/drizzle.js"; import { createChildLogger } from "../../shared/logger/logger.js"; const logger = createChildLogger("text-cache-store"); -/** GitHub API base URL for MythEclipse/bete repo */ -const GITHUB_API_BASE = "https://api.github.com/repos/MythEclipse/bete"; - -/** - * Fetch the current branch name from GitHub API as a real fallback - * when local git command is unavailable. This queries the actual remote. - */ -async function fetchRemoteBranchFromGitHub(): Promise { - try { - const controller = new AbortController(); - const timeout = setTimeout(() => controller.abort(), 5000); - - const response = await fetch(`${GITHUB_API_BASE}`, { - signal: controller.signal, - headers: { - Accept: "application/vnd.github+json", - "X-GitHub-Api-Version": "2022-11-28", - }, - }); - clearTimeout(timeout); - - if (!response.ok) { - logger.debug( - { status: response.status }, - "GitHub API repo fetch failed", - ); - return null; - } - - const data = (await response.json()) as { - default_branch?: string; - owner?: { login?: string }; - name?: string; - }; - - const owner = data.owner?.login ?? "unknown"; - const repo = data.name ?? "unknown"; - const branch = data.default_branch; - - if (branch) { - logger.info( - { owner, repo, branch, source: "github-api" }, - "Resolved branch from GitHub API", - ); - return branch; - } - - logger.warn({ owner, repo }, "GitHub API returned no default_branch"); - return null; - } catch (error) { - logger.debug( - { error: error instanceof Error ? error.message : String(error) }, - "GitHub API fetch failed", - ); - return null; - } -} - -/** - * Resolve the current git branch using a tiered strategy: - * 1. Local git CLI (rev-parse HEAD) - * 2. GitHub API (fetch actual remote repo info) - * 3. CACHE_MODEL_VERSION env var (explicit override) - * 4. Error log — no silent dummy fallbacks - */ -async function resolveBranch(): Promise { - // Tier 1: Local git CLI - try { - const branch = execFileSync( - "git", - ["rev-parse", "--abbrev-ref", "HEAD"], - { - encoding: "utf-8", - stdio: ["pipe", "pipe", "pipe"], - }, - ) - .trim() - .toLowerCase(); - - if (branch && branch !== "head") { - return branch; - } - - // Detached HEAD — use commit short hash - const commit = execFileSync("git", ["rev-parse", "--short", "HEAD"], { - encoding: "utf-8", - stdio: ["pipe", "pipe", "pipe"], - }) - .trim() - .toLowerCase(); - - if (commit) { - return `commit-${commit}`; - } - } catch (error) { - logger.debug( - { error: error instanceof Error ? error.message : String(error) }, - "Local git CLI unavailable, trying GitHub API", - ); - } - - // Tier 2: GitHub API — fetch real remote info - const remoteBranch = await fetchRemoteBranchFromGitHub(); - if (remoteBranch) { - return remoteBranch; - } - - // Tier 3: Environment variable (explicit override) - const envVersion = process.env.CACHE_MODEL_VERSION; - if (envVersion) { - logger.info( - { version: envVersion, source: "env" }, - "Using CACHE_MODEL_VERSION from env", - ); - return envVersion; - } - - return null; -} - -/** - * Normalize a branch name for use as a cache version key. - * Replaces non-alphanumeric characters with hyphens, collapses multiples. - */ -function normalizeBranchName(branch: string): string { - return branch - .replace(/[^a-z0-9-]/g, "-") - .replace(/-+/g, "-") - .replace(/^-|-$/g, ""); -} - -/** - * Get the current vision model version (resolved at module load time). - * Version is derived from actual git branch (local or remote) to ensure - * version control. Old cache entries with mismatched versions are ignored. - * - * Resolution order: - * 1. Local git branch name - * 2. GitHub API default branch (real fetch, no dummy) - * 3. CACHE_MODEL_VERSION env var - * 4. Error logged — falls back to "v1" with ERROR level - */ -let _resolvedVersion: string | null = null; - -function getVisionModelVersion(): string { - if (_resolvedVersion) { - return _resolvedVersion; - } - - // Run resolution synchronously via sync fetch for module init - // GitHub API call is sync-blocking only during startup - try { - // Try local git first (sync, already tried above) - const branch = execFileSync( - "git", - ["rev-parse", "--abbrev-ref", "HEAD"], - { - encoding: "utf-8", - stdio: ["pipe", "pipe", "pipe"], - }, - ) - .trim() - .toLowerCase(); - - if (branch && branch !== "head") { - _resolvedVersion = normalizeBranchName(branch); - return _resolvedVersion; - } - - // Detached HEAD - const commit = execFileSync("git", ["rev-parse", "--short", "HEAD"], { - encoding: "utf-8", - stdio: ["pipe", "pipe", "pipe"], - }) - .trim() - .toLowerCase(); - - if (commit) { - _resolvedVersion = `commit-${commit}`; - return _resolvedVersion; - } - } catch { - // Git not available — continue to next tier - } - - // Fallback: env var (we can't do async fetch here synchronously) - const envVersion = process.env.CACHE_MODEL_VERSION; - if (envVersion) { - logger.info( - { version: envVersion, source: "env" }, - "Using CACHE_MODEL_VERSION from env", - ); - _resolvedVersion = envVersion; - return _resolvedVersion; - } - - // No git, no env — log ERROR, no silent dummy - logger.error( - { - repo: "MythEclipse/bete", - githubApi: GITHUB_API_BASE, - }, - "Cache version resolution failed: git CLI unavailable, GitHub API unreachable (async), and CACHE_MODEL_VERSION not set. Using 'v1' as emergency fallback. Set CACHE_MODEL_VERSION in .env or ensure git is installed.", - ); - - _resolvedVersion = "v1"; - return _resolvedVersion; -} - -// Post-startup: asynchronously resolve from GitHub API and update version -// This runs in the background after module init to get the real remote branch -resolveBranch() - .then((branch) => { - if (branch) { - const normalized = normalizeBranchName(branch); - const previous = _resolvedVersion; - if (previous && previous !== normalized) { - logger.info( - { previous, resolved: normalized, source: "github-api-async" }, - "Cache version upgraded from startup fallback to GitHub API resolved branch", - ); - _resolvedVersion = normalized; - } - } - }) - .catch(() => { - // Silently ignore async failure — already logged in resolveBranch - }); - -/** - * Get the current vision model version (cached at module load time). - * Version is derived from git branch name and remains constant for this process. - */ -export const VISION_MODEL_VERSION = getVisionModelVersion(); - -logger.info({ version: VISION_MODEL_VERSION }, "Vision model version initialized"); - export interface TextCacheEntry { text: string; flags: string[]; @@ -253,17 +15,17 @@ export interface TextCacheEntry { /** * Lookup cached analysis result for a normalized text string. - * Returns null if not found, expired, or model version mismatch. + * 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, model_version + `SELECT text, flags, source, analyzed_at, expires_at, hit_count FROM text_analysis_cache - WHERE text = $1 AND expires_at > $2 AND model_version = $3`, - [text, Date.now(), VISION_MODEL_VERSION], + WHERE text = $1 AND expires_at > $2`, + [text, Date.now()], ); if (!row) return null; @@ -286,7 +48,7 @@ export async function getCachedText( } /** - * Insert or update a text analysis cache entry with model version. + * Insert or update a text analysis cache entry. */ export async function upsertCachedText( text: string, @@ -298,15 +60,14 @@ export async function upsertCachedText( try { await executeAll( - `INSERT INTO text_analysis_cache (text, flags, source, analyzed_at, expires_at, hit_count, model_version) - VALUES ($1, $2, $3, $4, $5, 0, $6) + `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, - model_version = EXCLUDED.model_version`, - [text, JSON.stringify(flags), source, now, expiresAt, VISION_MODEL_VERSION], + expires_at = EXCLUDED.expires_at`, + [text, JSON.stringify(flags), source, now, expiresAt], ); } catch (error) { logger.error( @@ -391,7 +152,7 @@ export async function getTextCacheStats(): Promise<{ } // --------------------------------------------------------------------------- -// Media / Vision analysis cache helpers (reuses text_analysis_cache table) +// Media / vision analysis cache helpers (reuses text_analysis_cache table) // --------------------------------------------------------------------------- /** @@ -422,17 +183,17 @@ export function makeImageCacheKey(dataUrl: string): string { /** * Lookup a cached media analysis result. - * Returns the full cached text (the analysis summary string) or null if not found, expired, or version mismatch. + * Returns the full cached text (the analysis summary string) or null if not found or expired. */ export async function getCachedMediaAnalysis( cacheKey: string, ): Promise { try { const row = await executeGet( - `SELECT flags, hit_count, model_version + `SELECT flags, hit_count FROM text_analysis_cache - WHERE text = $1 AND expires_at > $2 AND model_version = $3`, - [cacheKey, Date.now(), VISION_MODEL_VERSION], + WHERE text = $1 AND expires_at > $2`, + [cacheKey, Date.now()], ); if (!row) return null; @@ -450,7 +211,7 @@ export async function getCachedMediaAnalysis( } /** - * Store a media analysis result in the cache with model version tracking. + * Store a media analysis result in the cache. */ export async function upsertCachedMediaAnalysis( cacheKey: string, @@ -462,15 +223,14 @@ export async function upsertCachedMediaAnalysis( try { await executeAll( - `INSERT INTO text_analysis_cache (text, flags, source, analyzed_at, expires_at, hit_count, model_version) - VALUES ($1, $2, $3, $4, $5, 0, $6) + `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, - model_version = EXCLUDED.model_version`, - [cacheKey, JSON.stringify(analysisResult), source, now, expiresAt, VISION_MODEL_VERSION], + expires_at = EXCLUDED.expires_at`, + [cacheKey, JSON.stringify(analysisResult), source, now, expiresAt], ); } catch (error) { logger.error( diff --git a/services/discord-gateway/src/shared/config/config.ts b/services/discord-gateway/src/shared/config/config.ts index ba0405e..8888543 100644 --- a/services/discord-gateway/src/shared/config/config.ts +++ b/services/discord-gateway/src/shared/config/config.ts @@ -171,8 +171,6 @@ const configSchema = z AUTO_DELETE_ALLOWED_CATEGORIES: z.string().default(""), AUTO_DELETE_EXCLUDED_CHANNEL_IDS: z.string().default(""), AUTO_DELETE_EXCLUDED_USER_IDS: z.string().default(""), - STICKER_CACHE_DIR: z.string().default("./sticker-cache"), - STICKER_CACHE_MAX_SIZE_MB: z.coerce.number().int().positive().default(100), RETENTION_MESSAGES_DAYS: z.coerce.number().int().min(0).default(0), RETENTION_ATTACHMENTS_DAYS: z.coerce.number().int().min(0).default(0), RETENTION_VOICE_DAYS: z.coerce.number().int().min(0).default(0), diff --git a/services/discord-gateway/src/shared/database/migrations/add_model_version_to_cache.sql b/services/discord-gateway/src/shared/database/migrations/add_model_version_to_cache.sql deleted file mode 100644 index 40d1387..0000000 --- a/services/discord-gateway/src/shared/database/migrations/add_model_version_to_cache.sql +++ /dev/null @@ -1,24 +0,0 @@ --- Migration: Add model_version column to text_analysis_cache table --- Purpose: Track which vision/LLM model version produced each cache entry --- Reason: Invalidate old cache entries when model prompts change (e.g., terminal screenshot false positive fix) --- Date: 2026-06-02 - -BEGIN; - --- Add model_version column with default value -ALTER TABLE text_analysis_cache -ADD COLUMN model_version VARCHAR(50) NOT NULL DEFAULT 'v1'; - --- Create index for efficient filtering by model version -CREATE INDEX idx_text_analysis_cache_model_version -ON text_analysis_cache(model_version); - --- Create composite index for source + model_version queries (common pattern) -CREATE INDEX idx_text_analysis_cache_source_model_version -ON text_analysis_cache(source, model_version); - --- Optional: Clean up old vision_llm entries that may have stale/incorrect analysis --- Uncomment to remove all old vision analysis cache on deployment: --- DELETE FROM text_analysis_cache WHERE source = 'vision_llm' AND model_version = 'v1'; - -COMMIT; diff --git a/services/discord-gateway/src/shared/database/schema.ts b/services/discord-gateway/src/shared/database/schema.ts index d935bba..5340b21 100644 --- a/services/discord-gateway/src/shared/database/schema.ts +++ b/services/discord-gateway/src/shared/database/schema.ts @@ -361,9 +361,6 @@ export const pgRetentionPoliciesTable = pgTable( * * Uses the FULL normalized text (not per-word) because context matters: * "kau" alone is clean, but "awas kau" can be a threat. - * - * model_version tracks the vision/LLM model version that produced this cache entry. - * On model updates, bump the version to invalidate all old cache entries automatically. */ export const pgTextAnalysisCacheTable = pgTable( "text_analysis_cache", @@ -384,20 +381,39 @@ export const pgTextAnalysisCacheTable = pgTable( expires_at: pgBigint("expires_at", { mode: "number" }).notNull(), /** How many times this cached text has been reused. */ hit_count: pgInteger("hit_count").notNull().default(0), - /** Model version that produced this cache entry (e.g. "v1", "v2-2026-06-02"). Mismatched versions are ignored. */ - model_version: pgText("model_version").notNull().default("v1"), }, (table) => ({ expiresAtIdx: pgIndex("idx_text_analysis_cache_expires_at").on( table.expires_at, ), sourceIdx: pgIndex("idx_text_analysis_cache_source").on(table.source), - modelVersionIdx: pgIndex("idx_text_analysis_cache_model_version").on( - table.model_version, - ), - sourceModelVersionIdx: pgIndex( - "idx_text_analysis_cache_source_model_version", - ).on(table.source, table.model_version), + }), +); + +/** + * Sticker Cache Table (PostgreSQL) + * Stores base64-encoded sticker images for fast retrieval in media moderation. + * Replaces the file-based .dat + index.json cache. + * + * TTL: 7 days (enforced at query time via fetched_at) + * Eviction: LRU by fetched_at, max 100MB total + */ +export const pgStickerCacheTable = pgTable( + "sticker_cache", + { + /** Sanitized sticker name (encodeURIComponent + %→_) — primary key. */ + name: pgText("name").primaryKey(), + /** Base64-encoded image data. */ + base64: pgText("base64").notNull(), + /** MIME type of the image (e.g. "image/png", "image/gif"). */ + mime_type: pgText("mime_type").notNull(), + /** Byte length of the base64 string (for efficient SUM() eviction queries). */ + size: pgInteger("size").notNull(), + /** Epoch millis when this entry was stored. Used for TTL and LRU eviction. */ + fetched_at: pgBigint("fetched_at", { mode: "number" }).notNull(), + }, + (table) => ({ + fetchedAtIdx: pgIndex("idx_sticker_cache_fetched_at").on(table.fetched_at), }), ); @@ -414,6 +430,7 @@ export const messageReviewsTable = pgMessageReviewsTable; export const moderationActionsTable = pgModerationActionsTable; export const retentionPoliciesTable = pgRetentionPoliciesTable; export const textAnalysisCacheTable = pgTextAnalysisCacheTable; +export const stickerCacheTable = pgStickerCacheTable; // Export table types for use in queries export type MuxerJob = typeof muxerJobsTable.$inferSelect; @@ -442,3 +459,6 @@ export type ModerationActionInsert = typeof moderationActionsTable.$inferInsert; export type RetentionPolicy = typeof retentionPoliciesTable.$inferSelect; export type RetentionPolicyInsert = typeof retentionPoliciesTable.$inferInsert; + +export type StickerCacheRecord = typeof stickerCacheTable.$inferSelect; +export type StickerCacheInsert = typeof stickerCacheTable.$inferInsert; diff --git a/src/config.ts b/src/config.ts index 48422be..f7d3d34 100644 --- a/src/config.ts +++ b/src/config.ts @@ -171,8 +171,6 @@ const configSchema = z AUTO_DELETE_ALLOWED_CATEGORIES: z.string().default(""), AUTO_DELETE_EXCLUDED_CHANNEL_IDS: z.string().default(""), AUTO_DELETE_EXCLUDED_USER_IDS: z.string().default(""), - STICKER_CACHE_DIR: z.string().default("./sticker-cache"), - STICKER_CACHE_MAX_SIZE_MB: z.coerce.number().int().positive().default(100), RETENTION_MESSAGES_DAYS: z.coerce.number().int().min(0).default(0), RETENTION_ATTACHMENTS_DAYS: z.coerce.number().int().min(0).default(0), RETENTION_VOICE_DAYS: z.coerce.number().int().min(0).default(0), diff --git a/src/database/schema.ts b/src/database/schema.ts index d122471..5340b21 100644 --- a/src/database/schema.ts +++ b/src/database/schema.ts @@ -390,6 +390,33 @@ export const pgTextAnalysisCacheTable = pgTable( }), ); +/** + * Sticker Cache Table (PostgreSQL) + * Stores base64-encoded sticker images for fast retrieval in media moderation. + * Replaces the file-based .dat + index.json cache. + * + * TTL: 7 days (enforced at query time via fetched_at) + * Eviction: LRU by fetched_at, max 100MB total + */ +export const pgStickerCacheTable = pgTable( + "sticker_cache", + { + /** Sanitized sticker name (encodeURIComponent + %→_) — primary key. */ + name: pgText("name").primaryKey(), + /** Base64-encoded image data. */ + base64: pgText("base64").notNull(), + /** MIME type of the image (e.g. "image/png", "image/gif"). */ + mime_type: pgText("mime_type").notNull(), + /** Byte length of the base64 string (for efficient SUM() eviction queries). */ + size: pgInteger("size").notNull(), + /** Epoch millis when this entry was stored. Used for TTL and LRU eviction. */ + fetched_at: pgBigint("fetched_at", { mode: "number" }).notNull(), + }, + (table) => ({ + fetchedAtIdx: pgIndex("idx_sticker_cache_fetched_at").on(table.fetched_at), + }), +); + // Runtime table exports // ===================== @@ -403,6 +430,7 @@ export const messageReviewsTable = pgMessageReviewsTable; export const moderationActionsTable = pgModerationActionsTable; export const retentionPoliciesTable = pgRetentionPoliciesTable; export const textAnalysisCacheTable = pgTextAnalysisCacheTable; +export const stickerCacheTable = pgStickerCacheTable; // Export table types for use in queries export type MuxerJob = typeof muxerJobsTable.$inferSelect; @@ -431,3 +459,6 @@ export type ModerationActionInsert = typeof moderationActionsTable.$inferInsert; export type RetentionPolicy = typeof retentionPoliciesTable.$inferSelect; export type RetentionPolicyInsert = typeof retentionPoliciesTable.$inferInsert; + +export type StickerCacheRecord = typeof stickerCacheTable.$inferSelect; +export type StickerCacheInsert = typeof stickerCacheTable.$inferInsert; diff --git a/src/moderation/llmModerationClient.ts b/src/moderation/llmModerationClient.ts index f3af53b..1c5a228 100644 --- a/src/moderation/llmModerationClient.ts +++ b/src/moderation/llmModerationClient.ts @@ -5,8 +5,8 @@ import { config } from "../config.js"; import { createChildLogger } from "../logger.js"; import { retryWithBackoff } from "../retry.js"; import { withLlmConcurrency } from "./concurrencyLimiter.js"; -import { formatModerationTextEvidenceForPrompt } from "./indonesianTextNormalizer.js"; import { resizeImageForVision } from "./imageResizer.js"; +import { formatModerationTextEvidenceForPrompt } from "./indonesianTextNormalizer.js"; import { extractMessageMediaEvidence } from "./messageMetadata.js"; import { buildSystemPrompt as buildSystemPromptModular } from "./moderationPrompt.js"; import { @@ -966,10 +966,7 @@ async function _runSingleMediaAnalysis( ): Promise<{ results: AnalysisResult[]; raw: unknown }> { // Lazy init sticker cache if (!isStickerCacheReady()) { - await initStickerCache({ - cacheDir: config.STICKER_CACHE_DIR, - maxSizeBytes: config.STICKER_CACHE_MAX_SIZE_MB * 1024 * 1024, - }).catch((err) => { + await initStickerCache().catch((err) => { log.warn( { error: err instanceof Error ? err.message : String(err) }, "Sticker cache init failed — continuing without cache", diff --git a/src/moderation/stickerCache.ts b/src/moderation/stickerCache.ts index 8ebf93b..b51e4b2 100644 --- a/src/moderation/stickerCache.ts +++ b/src/moderation/stickerCache.ts @@ -1,9 +1,14 @@ -import { mkdir, readFile, unlink, writeFile } from "node:fs/promises"; -import { join } from "node:path"; +import { executeAll, executeGet } from "../database/drizzle.js"; import { createChildLogger } from "../logger.js"; const logger = createChildLogger("sticker-cache"); +const TTL_MS = 7 * 24 * 60 * 60 * 1000; // 7 days +const MAX_SIZE_BYTES = 100 * 1024 * 1024; // 100MB hardcoded + +let ready = false; +let statsCache = { entryCount: 0, totalSizeBytes: 0 }; + export interface StickerCacheEntry { base64: string; mimeType: string; @@ -11,87 +16,38 @@ export interface StickerCacheEntry { size: number; } -interface CacheIndexEntry { - file: string; - mimeType: string; - size: number; - fetchedAt: number; -} - -interface CacheIndex { - entries: Record; - totalSizeBytes: number; -} - -export interface StickerCacheOptions { - cacheDir: string; - maxSizeBytes: number; - ttlMs?: number; -} - -let cacheDir = ""; -let maxSizeBytes = 0; -let ttlMs = 7 * 24 * 60 * 60 * 1000; // 7 days default -let index: CacheIndex = { entries: {}, totalSizeBytes: 0 }; -let ready = false; - function sanitizeKey(name: string): string { return encodeURIComponent(name).replace(/%/g, "_"); } -async function loadIndex(): Promise { - try { - const raw = await readFile(join(cacheDir, "index.json"), "utf-8"); - return JSON.parse(raw) as CacheIndex; - } catch { - return { entries: {}, totalSizeBytes: 0 }; - } -} - -async function saveIndex(idx: CacheIndex): Promise { - await writeFile( - join(cacheDir, "index.json"), - JSON.stringify(idx, null, 2), - "utf-8", - ); -} - /** - * Initialise the sticker cache: create directory, load index. + * Initialise the sticker cache from PostgreSQL. * Idempotent — safe to call multiple times. */ -export async function initStickerCache( - opts: StickerCacheOptions, -): Promise { +export async function initStickerCache(): Promise { if (ready) return; - cacheDir = opts.cacheDir; - maxSizeBytes = opts.maxSizeBytes; - ttlMs = opts.ttlMs ?? 7 * 24 * 60 * 60 * 1000; - - await mkdir(cacheDir, { recursive: true }); - index = await loadIndex(); - - // Prune expired entries on startup - const now = Date.now(); - let changed = false; - for (const [key, meta] of Object.entries(index.entries)) { - if (now - meta.fetchedAt > ttlMs) { - await unlink(join(cacheDir, meta.file)).catch(() => {}); - index.totalSizeBytes -= meta.size; - delete index.entries[key]; - changed = true; + try { + await executeAll( + "DELETE FROM sticker_cache WHERE fetched_at < ?", + [Date.now() - TTL_MS], + ); + const row = await executeGet( + "SELECT count(*) as cnt, COALESCE(SUM(size), 0) as total FROM sticker_cache", + ); + if (row) { + statsCache = { + entryCount: Number(row.cnt), + totalSizeBytes: Number(row.total), + }; } + } catch (err) { + logger.warn( + { error: String(err) }, + "Failed to prune expired stickers on init", + ); } - if (changed) await saveIndex(index); - ready = true; - logger.info( - { - entryCount: Object.keys(index.entries).length, - totalSizeBytes: index.totalSizeBytes, - }, - "Sticker cache initialized", - ); + logger.info(statsCache, "Sticker cache initialized (PostgreSQL)"); } /** @@ -101,32 +57,24 @@ export async function getStickerFromCache( stickerName: string, ): Promise { if (!ready) return null; - const key = sanitizeKey(stickerName); - const meta = index.entries[key]; - if (!meta) return null; - - // TTL check - if (Date.now() - meta.fetchedAt > ttlMs) { - await unlink(join(cacheDir, meta.file)).catch(() => {}); - index.totalSizeBytes -= meta.size; - delete index.entries[key]; - await saveIndex(index); - return null; - } - try { - const raw = await readFile(join(cacheDir, meta.file), "utf-8"); + const row = await executeGet( + "SELECT base64, mime_type, size, fetched_at FROM sticker_cache WHERE name = ? AND fetched_at > ?", + [key, Date.now() - TTL_MS], + ); + if (!row) return null; return { - base64: raw, - mimeType: meta.mimeType, - fetchedAt: meta.fetchedAt, - size: meta.size, + base64: row.base64, + mimeType: row.mime_type, + fetchedAt: Number(row.fetched_at), + size: Number(row.size), }; - } catch { - // File missing — clean up index entry - delete index.entries[key]; - await saveIndex(index); + } catch (err) { + logger.error( + { error: String(err), stickerName }, + "Failed to get sticker from cache", + ); return null; } } @@ -140,52 +88,44 @@ export async function setStickerInCache( mimeType: string, ): Promise { if (!ready) return; - const key = sanitizeKey(stickerName); - const fileName = `${key}.dat`; const size = Buffer.byteLength(base64, "utf-8"); - - // Evict if needed - await evictIfNeeded(size); - + const now = Date.now(); try { - await writeFile(join(cacheDir, fileName), base64, "utf-8"); - index.entries[key] = { - file: fileName, - mimeType, - size, - fetchedAt: Date.now(), - }; - index.totalSizeBytes += size; - await saveIndex(index); + await evictIfNeeded(size); + await executeAll( + `INSERT INTO sticker_cache (name, base64, mime_type, size, fetched_at) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT (name) DO UPDATE SET + base64 = EXCLUDED.base64, + mime_type = EXCLUDED.mime_type, + size = EXCLUDED.size, + fetched_at = EXCLUDED.fetched_at`, + [key, base64, mimeType, size, now], + ); + statsCache.entryCount++; + statsCache.totalSizeBytes += size; logger.debug({ stickerName, size }, "Sticker cached"); } catch (err) { logger.warn( - { stickerName, error: err instanceof Error ? err.message : String(err) }, + { stickerName, error: String(err) }, "Failed to write sticker to cache", ); } } async function evictIfNeeded(newSize: number): Promise { - while (index.totalSizeBytes + newSize > maxSizeBytes) { - // Find oldest entry - let oldestKey: string | null = null; - let oldestTime = Infinity; - for (const [key, meta] of Object.entries(index.entries)) { - if (meta.fetchedAt < oldestTime) { - oldestTime = meta.fetchedAt; - oldestKey = key; - } - } - if (!oldestKey) break; - - const meta = index.entries[oldestKey]; - await unlink(join(cacheDir, meta.file)).catch(() => {}); - index.totalSizeBytes -= meta.size; - delete index.entries[oldestKey]; + while (statsCache.totalSizeBytes + newSize > MAX_SIZE_BYTES) { + const oldest = await executeGet( + "SELECT name, size FROM sticker_cache ORDER BY fetched_at ASC LIMIT 1", + ); + if (!oldest) break; + await executeAll("DELETE FROM sticker_cache WHERE name = ?", [ + oldest.name, + ]); + statsCache.totalSizeBytes -= Number(oldest.size); + statsCache.entryCount--; } - await saveIndex(index); } /** @@ -195,10 +135,7 @@ export function getStickerCacheStats(): { entryCount: number; totalSizeBytes: number; } { - return { - entryCount: Object.keys(index.entries).length, - totalSizeBytes: index.totalSizeBytes, - }; + return { ...statsCache }; } /**