feat: refactor database access in analyticsStore to use executeAll and executeGet for improved query handling

src/database/drizzle.ts

@@ -12,6 +12,8 @@ let db:
   | ReturnType<typeof drizzlePostgres>
   | ReturnType<typeof drizzleSqlite>
   | null = null;
+let rawSqlite: ReturnType<typeof Database> | null = null;
+let rawPool: Pool | null = null;
 
 /**
  * Initialize the database connection based on DATABASE_TYPE config
@@ -49,6 +51,7 @@ export async function initializeDatabase() {
       });
     }
 
+    rawPool = pool;
     db = drizzlePostgres(pool, { schema });
     // Provide a simple `run` helper for tests that expect it.
     try {
@@ -61,6 +64,7 @@ export async function initializeDatabase() {
     const sqlite = new Database(".muxer-queue.db");
     sqlite.pragma("journal_mode = WAL");
 
+    rawSqlite = sqlite;
     db = drizzleSqlite(sqlite, { schema });
     // Expose a convenience `run` method used by tests that expect a simple API.
     // `sqlite` is the underlying better-sqlite3 Database instance.
@@ -88,6 +92,41 @@ export function getDatabase() {
   return db;
 }
 
+function convertPlaceholdersForPostgres(sql: string) {
+  let i = 0;
+  return sql.replace(/\?/g, () => `$${++i}`);
+}
+
+export async function executeAll(sql: string, params?: any[]) {
+  if (rawPool) {
+    const q = convertPlaceholdersForPostgres(sql);
+    const res = await rawPool.query(q, params || []);
+    return res.rows;
+  }
+
+  if (rawSqlite) {
+    const stmt = rawSqlite.prepare(sql);
+    return stmt.all(...(params || []));
+  }
+
+  throw new Error("Database not initialized. Call initializeDatabase() first.");
+}
+
+export async function executeGet(sql: string, params?: any[]) {
+  if (rawPool) {
+    const q = convertPlaceholdersForPostgres(sql);
+    const res = await rawPool.query(q, params || []);
+    return res.rows[0] ?? null;
+  }
+
+  if (rawSqlite) {
+    const stmt = rawSqlite.prepare(sql);
+    return stmt.get(...(params || []));
+  }
+
+  throw new Error("Database not initialized. Call initializeDatabase() first.");
+}
+
 /**
  * Close the database connection
  * For PostgreSQL, the pool will close on process exit

src/moderation/analyticsStore.ts

@@ -1,4 +1,4 @@
-import { getDatabase } from "../database/drizzle.js";
+import { executeAll, executeGet } from "../database/drizzle.js";
 import { createChildLogger } from "../logger.js";
 import type { MessageRecord } from "./types.js";
 
@@ -52,23 +52,44 @@ export interface AnalyticsOverview {
   total_channels: number;
 }
 
-// ── Cache for topic trends ─────────────────────────────────────────────
+// ══════════════════════════════════════════════════════════════════════════
+// GENERIC QUERY CACHE (reduces duplicate DB calls from 5s auto-refresh)
+// ══════════════════════════════════════════════════════════════════════════
 
-interface TopicCacheEntry {
-  data: TopicTrend[];
+interface CacheEntry<T> {
+  data: T;
   expiresAt: number;
-  key: string;
 }
 
-const topicCache = new Map<string, TopicCacheEntry>();
-const TOPIC_CACHE_TTL_MS = 60_000; // 1 minute TTL
+const queryCache = new Map<string, CacheEntry<any>>();
 
-function makeTopicCacheKey(input: {
-  guildId: string;
-  channelId?: string;
-  hours: number;
-}): string {
-  return `${input.guildId}:${input.channelId ?? "*"}:${input.hours}`;
+/** Default TTL for aggregate queries — 10s is long enough to prevent redundant
+ *  calls from the 5s auto-refresh but short enough to feel real-time. */
+const AGGREGATE_CACHE_TTL_MS = 10_000;
+
+/** Topic extraction is expensive (JSON parsing). Cache longer. */
+const TOPIC_CACHE_TTL_MS = 120_000;
+
+function makeCacheKey(prefix: string, params: Record<string, any>): string {
+  return `${prefix}:${JSON.stringify(params)}`;
+}
+
+function getCached<T>(key: string): T | undefined {
+  const entry = queryCache.get(key);
+  if (entry && entry.expiresAt > Date.now()) return entry.data;
+  if (entry) queryCache.delete(key); // expired
+  return undefined;
+}
+
+function setCache<T>(key: string, data: T, ttl: number): void {
+  queryCache.set(key, { data, expiresAt: Date.now() + ttl });
+  // Prune old entries if cache grows too large (>200 entries)
+  if (queryCache.size > 200) {
+    const now = Date.now();
+    for (const [k, v] of queryCache) {
+      if (v.expiresAt <= now) queryCache.delete(k);
+    }
+  }
 }
 
 // ── Hourly Message Stats ───────────────────────────────────────────────
@@ -78,11 +99,14 @@ export async function getHourlyStats(input: {
   channelId?: string;
   hours?: number;
 }): Promise<HourlyBucket[]> {
+  const { guildId, channelId, hours = 24 } = input;
+  const cacheKey = makeCacheKey("hourly", { guildId, channelId, hours });
+  const cached = getCached<HourlyBucket[]>(cacheKey);
+  if (cached) return cached;
+
   try {
-    const { guildId, channelId, hours = 24 } = input;
     const since = Date.now() - hours * 3600_000;
-    const rawDb = getDatabase() as any;
-    const sqliteRows = rawDb.all(
+    const sqliteRows = await executeAll(
       `
       SELECT
         datetime((created_at / 3600000) * 3600, 'unixepoch') as hour,
@@ -107,13 +131,7 @@ export async function getHourlyStats(input: {
     // Initialize all hour buckets (fill gaps with zeros)
     const buckets = new Map<
       string,
-      {
-        count: number;
-        clean: number;
-        warned: number;
-        flagged: number;
-        error: number;
-      }
+      { count: number; clean: number; warned: number; flagged: number; error: number }
     >();
 
     for (let h = 0; h < hours; h++) {
@@ -124,13 +142,10 @@ export async function getHourlyStats(input: {
     }
 
     for (const row of sqliteRows) {
-      // Normalize the SQL hour key to match our bucket format
       const d = new Date(row.hour.replace(" ", "T") + "Z");
       const key = d.toISOString().slice(0, 13) + ":00:00Z";
-
       const bucket = buckets.get(key);
       if (!bucket) continue;
-
       bucket.count = row.count;
       bucket.clean = row.clean;
       bucket.warned = row.warned;
@@ -138,9 +153,12 @@ export async function getHourlyStats(input: {
       bucket.error = row.error;
     }
 
-    return Array.from(buckets.entries())
+    const result = Array.from(buckets.entries())
       .sort(([a], [b]) => a.localeCompare(b))
       .map(([hour, data]) => ({ hour, ...data }));
+
+    setCache(cacheKey, result, AGGREGATE_CACHE_TTL_MS);
+    return result;
   } catch (error) {
     logger.error(
       { error: error instanceof Error ? error.message : String(error) },
@@ -153,156 +171,25 @@ export async function getHourlyStats(input: {
 // ── Topic Trends ───────────────────────────────────────────────────────
 
 const STOP_WORDS = new Set([
-  "yang",
-  "dan",
-  "itu",
-  "ini",
-  "dengan",
-  "akan",
-  "pada",
-  "dari",
-  "di",
-  "ke",
-  "untuk",
-  "tidak",
-  "ada",
-  "juga",
-  "sudah",
-  "saya",
-  "kamu",
-  "dia",
-  "mereka",
-  "kami",
-  "aku",
-  "lo",
-  "lu",
-  "gua",
-  "gue",
-  "org",
-  "orang",
-  "aja",
-  "sama",
-  "kalo",
-  "kalau",
-  "bisa",
-  "karena",
-  "gak",
-  "nggak",
-  "ga",
-  "tak",
-  "belum",
-  "udah",
-  "dah",
-  "lah",
-  "kah",
-  "pun",
-  "nih",
-  "tuh",
-  "deh",
-  "dong",
-  "si",
-  "nya",
-  "kan",
-  "ya",
-  "yah",
-  "yuk",
-  "kok",
-  "loh",
-  "nah",
-  "wow",
-  "eh",
-  "the",
-  "a",
-  "an",
-  "is",
-  "are",
-  "was",
-  "were",
-  "be",
-  "been",
-  "being",
-  "have",
-  "has",
-  "had",
-  "having",
-  "do",
-  "does",
-  "did",
-  "doing",
-  "will",
-  "would",
-  "could",
-  "should",
-  "may",
-  "might",
-  "must",
-  "shall",
-  "i",
-  "you",
-  "he",
-  "she",
-  "it",
-  "we",
-  "they",
-  "me",
-  "him",
-  "her",
-  "us",
-  "them",
-  "my",
-  "your",
-  "his",
-  "its",
-  "our",
-  "their",
-  "and",
-  "but",
-  "or",
-  "nor",
-  "not",
-  "so",
-  "yet",
-  "for",
-  "if",
-  "to",
-  "of",
-  "in",
-  "on",
-  "at",
-  "by",
-  "as",
-  "with",
-  "about",
-  "just",
-  "then",
-  "now",
-  "here",
-  "there",
-  "when",
-  "where",
-  "why",
-  "how",
-  "all",
-  "both",
-  "each",
-  "few",
-  "more",
-  "most",
-  "other",
-  "some",
-  "such",
-  "only",
-  "own",
-  "same",
-  "too",
-  "very",
-  "can",
-  "go",
-  "ok",
-  "okay",
-  "yeah",
-  "yes",
-  "no",
+  "yang", "dan", "itu", "ini", "dengan", "akan", "pada", "dari", "di", "ke",
+  "untuk", "tidak", "ada", "juga", "sudah", "saya", "kamu", "dia", "mereka",
+  "kami", "aku", "lo", "lu", "gua", "gue", "org", "orang", "aja", "sama",
+  "kalo", "kalau", "bisa", "karena", "gak", "nggak", "ga", "tak", "belum",
+  "udah", "dah", "lah", "kah", "pun", "nih", "tuh", "deh", "dong", "si",
+  "nya", "kan", "ya", "yah", "yuk", "kok", "loh", "nah", "wow", "eh",
+  "the", "a", "an", "is", "are", "was", "were", "be", "been", "being",
+  "have", "has", "had", "having", "do", "does", "did", "doing",
+  "will", "would", "could", "should", "may", "might", "must", "shall",
+  "i", "you", "he", "she", "it", "we", "they",
+  "me", "him", "her", "us", "them",
+  "my", "your", "his", "its", "our", "their",
+  "and", "but", "or", "nor", "not", "so", "yet", "for", "if",
+  "to", "of", "in", "on", "at", "by", "as", "with",
+  "about", "just", "then", "now", "here", "there",
+  "when", "where", "why", "how",
+  "all", "both", "each", "few", "more", "most", "other",
+  "some", "such", "only", "own", "same", "too", "very",
+  "can", "go", "ok", "okay", "yeah", "yes", "no",
 ]);
 
 function extractTopics(messages: MessageRecord[], topN = 15): TopicTrend[] {
@@ -338,10 +225,7 @@ function extractTopics(messages: MessageRecord[], topN = 15): TopicTrend[] {
             existing.count++;
             existing.score += msg.ai_moderation_score || 0;
           } else {
-            topicScores.set(cat, {
-              count: 1,
-              score: msg.ai_moderation_score || 0,
-            });
+            topicScores.set(cat, { count: 1, score: msg.ai_moderation_score || 0 });
           }
         }
       } catch {
@@ -376,11 +260,7 @@ function extractTopics(messages: MessageRecord[], topN = 15): TopicTrend[] {
 
   for (const [word, count] of sortedWords) {
     if (!topicScores.has(word)) {
-      results.push({
-        topic: word,
-        count,
-        score: flaggedWordFreq.get(word) || 0,
-      });
+      results.push({ topic: word, count, score: flaggedWordFreq.get(word) || 0 });
     }
   }
 
@@ -393,19 +273,16 @@ export async function getTopicTrends(input: {
   hours?: number;
 }): Promise<TopicTrend[]> {
   const { guildId, channelId, hours = 24 } = input;
-  const cacheKey = makeTopicCacheKey({ guildId, channelId, hours });
-
-  // Check cache first (P2: cache topic extraction)
-  const cached = topicCache.get(cacheKey);
-  if (cached && cached.expiresAt > Date.now()) {
-    return cached.data;
-  }
+  const cacheKey = makeCacheKey("topics", { guildId, channelId, hours });
+  const cached = getCached<TopicTrend[]>(cacheKey);
+  if (cached) return cached;
 
   try {
     const since = Date.now() - hours * 3600_000;
-    const rawDb = getDatabase() as any;
 
-    const rows = rawDb.all(
+    // Only fetch messages that have ai_analysis (the ones that actually have topics)
+    // This dramatically reduces rows for large guilds
+    const rows = await executeAll(
       `
       SELECT
         id, content, ai_status, ai_analysis, ai_moderation_score,
@@ -414,9 +291,10 @@ export async function getTopicTrends(input: {
       WHERE guild_id = ?
         AND created_at >= ?
         AND deleted_at IS NULL
+        AND ai_analysis IS NOT NULL
         ${channelId ? `AND (channel_id = ? OR thread_id = ?)` : ""}
       ORDER BY created_at DESC
-      LIMIT 1000
+      LIMIT 2000
       `,
       channelId
         ? [guildId, since, channelId, channelId]
@@ -424,14 +302,7 @@ export async function getTopicTrends(input: {
     ) as MessageRecord[];
 
     const result = extractTopics(rows);
-
-    // Store in cache
-    topicCache.set(cacheKey, {
-      data: result,
-      expiresAt: Date.now() + TOPIC_CACHE_TTL_MS,
-      key: cacheKey,
-    });
-
+    setCache(cacheKey, result, TOPIC_CACHE_TTL_MS);
     return result;
   } catch (error) {
     logger.error(
@@ -450,13 +321,14 @@ export async function getUserLeaderboard(input: {
   hours?: number;
   limit?: number;
 }): Promise<UserStat[]> {
-  try {
-    const { guildId, channelId, hours = 24, limit = 20 } = input;
-    const since = Date.now() - hours * 3600_000;
-    const rawDb = getDatabase() as any;
+  const { guildId, channelId, hours = 24, limit = 20 } = input;
+  const cacheKey = makeCacheKey("leaderboard", { guildId, channelId, hours, limit });
+  const cached = getCached<UserStat[]>(cacheKey);
+  if (cached) return cached;
 
-    // SQL-level GROUP BY aggregate instead of SELECT * + in-memory map
-    const rows = rawDb.all(
+  try {
+    const since = Date.now() - hours * 3600_000;
+    const rows = await executeAll(
       `
       SELECT
         user_id,
@@ -481,7 +353,9 @@ export async function getUserLeaderboard(input: {
         : [guildId, since, limit],
     );
 
-    return rows as UserStat[];
+    const result = rows as UserStat[];
+    setCache(cacheKey, result, AGGREGATE_CACHE_TTL_MS);
+    return result;
   } catch (error) {
     logger.error(
       { error: error instanceof Error ? error.message : String(error) },
@@ -498,13 +372,14 @@ export async function getModerationStats(input: {
   channelId?: string;
   hours?: number;
 }): Promise<ModerationBreakdown> {
-  try {
-    const { guildId, channelId, hours = 24 } = input;
-    const since = Date.now() - hours * 3600_000;
-    const rawDb = getDatabase() as any;
+  const { guildId, channelId, hours = 24 } = input;
+  const cacheKey = makeCacheKey("modstats", { guildId, channelId, hours });
+  const cached = getCached<ModerationBreakdown>(cacheKey);
+  if (cached) return cached;
 
-    // SQL-level aggregate instead of SELECT * + in-memory counting
-    const row = rawDb.get(
+  try {
+    const since = Date.now() - hours * 3600_000;
+    const row = await executeGet(
       `
       SELECT
         count(*) as total,
@@ -525,41 +400,26 @@ export async function getModerationStats(input: {
         : [guildId, since],
     );
 
-    if (!row) {
-      return {
-        total: 0,
-        clean: 0,
-        warned: 0,
-        flagged: 0,
-        error: 0,
-        pending: 0,
-        average_score: 0,
-      };
-    }
+    const result: ModerationBreakdown = row
+      ? {
+          total: row.total ?? 0,
+          clean: row.clean ?? 0,
+          warned: row.warned ?? 0,
+          flagged: row.flagged ?? 0,
+          error: row.error ?? 0,
+          pending: row.pending ?? 0,
+          average_score: row.average_score ?? 0,
+        }
+      : { total: 0, clean: 0, warned: 0, flagged: 0, error: 0, pending: 0, average_score: 0 };
 
-    return {
-      total: row.total ?? 0,
-      clean: row.clean ?? 0,
-      warned: row.warned ?? 0,
-      flagged: row.flagged ?? 0,
-      error: row.error ?? 0,
-      pending: row.pending ?? 0,
-      average_score: row.average_score ?? 0,
-    };
+    setCache(cacheKey, result, AGGREGATE_CACHE_TTL_MS);
+    return result;
   } catch (error) {
     logger.error(
       { error: error instanceof Error ? error.message : String(error) },
       "Failed to get moderation stats",
     );
-    return {
-      total: 0,
-      clean: 0,
-      warned: 0,
-      flagged: 0,
-      error: 0,
-      pending: 0,
-      average_score: 0,
-    };
+    return { total: 0, clean: 0, warned: 0, flagged: 0, error: 0, pending: 0, average_score: 0 };
   }
 }
 
@@ -569,12 +429,14 @@ export async function getActiveChannelCount(input: {
   guildId: string;
   hours?: number;
 }): Promise<number> {
-  try {
-    const { guildId, hours = 24 } = input;
-    const since = Date.now() - hours * 3600_000;
-    const rawDb = getDatabase() as any;
+  const { guildId, hours = 24 } = input;
+  const cacheKey = makeCacheKey("channels", { guildId, hours });
+  const cached = getCached<number>(cacheKey);
+  if (cached !== undefined) return cached;
 
-    const row = rawDb.get(
+  try {
+    const since = Date.now() - hours * 3600_000;
+    const row = await executeGet(
       `
       SELECT count(DISTINCT channel_id) as cnt
       FROM messages
@@ -585,7 +447,9 @@ export async function getActiveChannelCount(input: {
       [guildId, since],
     );
 
-    return row?.cnt ?? 0;
+    const result = row?.cnt ?? 0;
+    setCache(cacheKey, result, AGGREGATE_CACHE_TTL_MS);
+    return result;
   } catch (error) {
     logger.error(
       { error: error instanceof Error ? error.message : String(error) },
@@ -604,8 +468,8 @@ export interface ViolatorStat {
   total_messages: number;
   flagged_count: number;
   warned_count: number;
-  violation_score: number; // weighted: flagged*3 + warned*1
-  worst_flags: string[]; // unique flag types
+  violation_score: number;
+  worst_flags: string[];
   last_violation: number;
 }
 
@@ -615,13 +479,14 @@ export async function getTopViolators(input: {
   hours?: number;
   limit?: number;
 }): Promise<ViolatorStat[]> {
-  try {
-    const { guildId, channelId, hours = 24, limit = 20 } = input;
-    const since = Date.now() - hours * 3600_000;
-    const rawDb = getDatabase() as any;
+  const { guildId, channelId, hours = 24, limit = 20 } = input;
+  const cacheKey = makeCacheKey("violators", { guildId, channelId, hours, limit });
+  const cached = getCached<ViolatorStat[]>(cacheKey);
+  if (cached) return cached;
 
-    // SQL-level GROUP BY aggregate for base stats
-    const rows = rawDb.all(
+  try {
+    const since = Date.now() - hours * 3600_000;
+    const rows = await executeAll(
       `
       SELECT
         user_id,
@@ -654,10 +519,11 @@ export async function getTopViolators(input: {
       flagged_count: row.flagged_count,
       warned_count: row.warned_count,
       violation_score: row.flagged_count * 3 + row.warned_count,
-      worst_flags: [], // flags require parsing JSON per-row; skip for perf
+      worst_flags: [],
       last_violation: row.last_violation,
     }));
 
+    setCache(cacheKey, violators, AGGREGATE_CACHE_TTL_MS);
     return violators;
   } catch (error) {
     logger.error(
@@ -668,6 +534,18 @@ export async function getTopViolators(input: {
   }
 }
 
+// ── Cache Invalidation (called when new messages arrive) ───────────────
+
+export function invalidateAnalyticsCache(guildId: string): void {
+  const now = Date.now();
+  const needle = `"${guildId}"`;
+  for (const [key, entry] of queryCache) {
+    if (key.includes(needle) && entry.expiresAt > now) {
+      entry.expiresAt = 0; // expire immediately
+    }
+  }
+}
+
 // ── Combined Overview ──────────────────────────────────────────────────
 
 export async function getAnalyticsOverview(input: {
@@ -679,15 +557,13 @@ export async function getAnalyticsOverview(input: {
   const now = Date.now();
   const since = now - hours * 3600_000;
 
-  const [messages, hourly, topics, topUsers, totalChannels] = await Promise.all(
-    [
-      getModerationStats(input),
-      getHourlyStats(input),
-      getTopicTrends(input),
-      getUserLeaderboard(input),
-      getActiveChannelCount({ guildId, hours }),
-    ],
-  );
+  const [messages, hourly, topics, topUsers, totalChannels] = await Promise.all([
+    getModerationStats(input),
+    getHourlyStats(input),
+    getTopicTrends(input),
+    getUserLeaderboard(input),
+    getActiveChannelCount({ guildId, hours }),
+  ]);
 
   return {
     period: { start: since, end: now },