MythEclipse/GMW
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor: migrate sticker cache to PostgreSQL, remove versioning

Browse Source 
- 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
This commit is contained in:
MythEclipse
2026-06-02 14:05:51 +07:00
parent e584a7941e
commit 406a6cbf79
11 changed files with 233 additions and 591 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/discord-gateway/src/modules/ai-moderation/llmModerationClient.ts
+12 -18
View File
@@ -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);
services/discord-gateway/src/modules/ai-moderation/responseLogger.ts
+2 -8
View File
@@ -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}`,
);
}
services/discord-gateway/src/modules/ai-moderation/stickerCache.ts
+68 -131
View File
@@ -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<string, CacheIndexEntry>;
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<CacheIndex> {
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<void> {
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<void> {
export async function initStickerCache(): Promise<void> {
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<StickerCacheEntry | null> {
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<void> {
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<void> {
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 };
}
/**
services/discord-gateway/src/modules/ai-moderation/textCacheStore.ts
+19 -259
View File
@@ -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<string | null> {
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<string | null> {
// 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<TextCacheEntry | null> {
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<string | null> {
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(
services/discord-gateway/src/shared/config/config.ts
-2
View File
@@ -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),
services/discord-gateway/src/shared/database/migrations/add_model_version_to_cache.sql
-24
View File
@@ -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;
services/discord-gateway/src/shared/database/schema.ts
+31 -11
View File
@@ -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;
src/config.ts
-2
View File
@@ -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),
src/database/schema.ts
+31
View File
@@ -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;
src/moderation/llmModerationClient.ts
+2 -5
View File
@@ -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",
src/moderation/stickerCache.ts
+68 -131
View File
@@ -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<string, CacheIndexEntry>;
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<CacheIndex> {
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<void> {
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<void> {
export async function initStickerCache(): Promise<void> {
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<StickerCacheEntry | null> {
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<void> {
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<void> {
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 };
}
/**