feat(ai-moderation): introduce user reputation and channel culture context

Implements a context-aware moderation system by tracking user behavior and channel-specific norms to improve AI decision-making accuracy. - Adds `user_reputations` table to track trust scores, clean streaks, and infraction history. - Adds `channel_cultures` table to store AI-generated summaries of channel-specific norms and slang. - Implements `userReputationStore` to autonomously update user scores based on moderation outcomes (clean vs. flagged). - Implements `cultureLearner` and `channelCultureStore` to manage evolving channel contexts. - Enhances LLM prompts to inject user reputation (trust scores, history) and channel culture summaries, enabling "wisdom-based" moderation (e.g., giving benefit of the doubt to high-trust users). - Integrates reputation and culture updates into the existing `aiAnalyzer` pipeline.
2026-06-05 18:04:57 +07:00
parent f057bf1f0b
commit 2f3d7e1d61
7 changed files with 445 additions and 2 deletions
@@ -488,6 +488,17 @@ async function processIndividualFallback(
      broadcastAnalysisCompleted(row);
      invalidateAnalyticsCache(row.guild_id);
      scheduleAutoDelete(row);
+
+      // Update reputation autonomously (Belajar & Kebijaksanaan)
+      if (row.ai_status === "clean") {
+        import("./userReputationStore.js")
+          .then((store) => store.recordCleanMessage(row.user_id, row.guild_id))
+          .catch((e) => logger.error({ error: e }, "Failed to record clean message streak in fallback"));
+      } else if (row.ai_status === "flagged" && row.ai_severity !== "none") {
+        import("./userReputationStore.js")
+          .then((store) => store.recordInfraction(row.user_id, row.guild_id, row.ai_severity as "low"|"medium"|"high"|"critical"))
+          .catch((e) => logger.error({ error: e }, "Failed to record infraction penalty in fallback"));
+      }
    }

    const resultSummary = analysisResult.results[0];
@@ -701,6 +712,17 @@ async function processBatch(
      if (!isApiFailure) {
        broadcastAnalysisCompleted(row);
        scheduleAutoDelete(row);
+
+        // Update reputation autonomously (Belajar & Kebijaksanaan)
+        if (row.ai_status === "clean") {
+          import("./userReputationStore.js")
+            .then((store) => store.recordCleanMessage(row.user_id, row.guild_id))
+            .catch((e) => logger.error({ error: e }, "Failed to record clean message streak"));
+        } else if (row.ai_status === "flagged" && row.ai_severity !== "none") {
+          import("./userReputationStore.js")
+            .then((store) => store.recordInfraction(row.user_id, row.guild_id, row.ai_severity as "low"|"medium"|"high"|"critical"))
+            .catch((e) => logger.error({ error: e }, "Failed to record infraction penalty"));
+        }
      }
    }

@@ -1097,6 +1119,8 @@ export function startPendingAIAnalysisWorker(
  _redisEventBroadcaster = eventBroadcaster;
  if (!config.AI_ANALYSIS_ENABLED) return;

+  import("./cultureLearner.js").then(m => m.startCultureLearnerWorker()).catch(console.error);
+
  setInterval(() => {
    revertStuckProcessingMessages(300000).catch((err: unknown) => {
      logger.error(
@@ -0,0 +1,47 @@
+import { eq } from "drizzle-orm";
+import { getDatabase } from "../../shared/database/drizzle.js";
+import {
+  channelCulturesTable,
+  ChannelCulture,
+} from "../../shared/database/schema.js";
+
+/**
+ * Fetch the AI-generated culture summary for a channel.
+ */
+export async function getChannelCulture(channelId: string): Promise<ChannelCulture | null> {
+  const db = getDatabase();
+  const existing = await db
+    .select()
+    .from(channelCulturesTable)
+    .where(eq(channelCulturesTable.channel_id, channelId))
+    .limit(1);
+
+  return existing[0] || null;
+}
+
+/**
+ * Update the AI-generated culture summary for a channel.
+ */
+export async function updateChannelCulture(
+  channelId: string,
+  guildId: string,
+  cultureSummary: string,
+): Promise<void> {
+  const db = getDatabase();
+  
+  await db
+    .insert(channelCulturesTable)
+    .values({
+      channel_id: channelId,
+      guild_id: guildId,
+      culture_summary: cultureSummary,
+      last_analyzed_at: Date.now(),
+    })
+    .onConflictDoUpdate({
+      target: channelCulturesTable.channel_id,
+      set: {
+        culture_summary: cultureSummary,
+        last_analyzed_at: Date.now(),
+      },
+    });
+}
@@ -0,0 +1,117 @@
+import { eq, desc, sql, and } from "drizzle-orm";
+import { getDatabase } from "../../shared/database/drizzle.js";
+import {
+  messagesTable,
+  channelCulturesTable,
+} from "../../shared/database/schema.js";
+import { config } from "../../shared/config/config.js";
+import { createChildLogger } from "@bete/shared/logger";
+import { llmChat } from "./llmClient.js";
+import { updateChannelCulture } from "./channelCultureStore.js";
+
+const CULTURE_LEARNING_INTERVAL = 1000 * 60 * 60 * 12; // 12 hours
+const log = createChildLogger("cultureLearner");
+
+async function learnChannelCulture(channelId: string, guildId: string): Promise<void> {
+  const db = getDatabase();
+  
+  // Get recent clean messages for this channel
+  const recentMessages = await db
+    .select({
+      content: messagesTable.content,
+      username: messagesTable.username,
+    })
+    .from(messagesTable)
+    .where(
+      and(
+        eq(messagesTable.channel_id, channelId),
+        eq(messagesTable.ai_status, "clean")
+      )
+    )
+    .orderBy(desc(messagesTable.created_at))
+    .limit(100);
+
+  if (recentMessages.length < 10) {
+    log.debug({ channelId }, "Not enough messages to learn culture");
+    return;
+  }
+
+  const messagesText = recentMessages
+    .reverse()
+    .map(m => `${m.username}: ${m.content}`)
+    .join("\n");
+
+  const prompt = `Anda adalah AI ahli perilaku sosiologis dan budaya online.
+Tugas Anda adalah merangkum budaya (culture) dari sebuah channel chat berdasarkan riwayat pesan-pesan yang dianggap bersih (clean/tidak melanggar).
+
+Pesan-pesan terakhir:
+<messages>
+${messagesText}
+</messages>
+
+Berdasarkan pesan-pesan di atas, buatlah ringkasan singkat (maksimal 3 paragraf) mengenai gaya bahasa, topik obrolan, dan norma sosial di channel ini. Ringkasan ini akan digunakan oleh sistem AI moderasi untuk memahami konteks dan "inside jokes" yang wajar di channel ini.
+Jangan menambahkan teks basa-basi, langsung berikan ringkasannya.`;
+
+  try {
+    const completion = await llmChat({
+      messages: [{ role: "user", content: prompt }],
+      max_tokens: 500,
+      temperature: 0.7, // Higher temp for summarization
+      retries: 2,
+    });
+    
+    if (!completion) throw new Error("Empty response from LLM");
+    const text = completion.choices[0]?.message?.content?.trim();
+    if (!text) throw new Error("Empty response from LLM");
+    
+    await updateChannelCulture(channelId, guildId, text);
+    log.info({ channelId, guildId }, "Successfully learned and updated channel culture");
+  } catch (error) {
+    log.error({ channelId, error }, "Failed to learn channel culture");
+  }
+}
+
+export async function runCultureLearningCycle(): Promise<void> {
+  const db = getDatabase();
+  log.info("Starting culture learning cycle");
+
+  try {
+    // Find channels that haven't been analyzed recently
+    // We do a simple distinct channel_id query with a left join to see if it's stale
+    const staleChannels = await db.execute(sql`
+      SELECT m.channel_id, m.guild_id
+      FROM (SELECT DISTINCT channel_id, guild_id FROM messages) m
+      LEFT JOIN channel_cultures c ON m.channel_id = c.channel_id
+      WHERE c.last_analyzed_at IS NULL 
+         OR c.last_analyzed_at < ${Date.now() - CULTURE_LEARNING_INTERVAL}
+      LIMIT 50
+    `);
+
+    for (const row of staleChannels.rows || staleChannels) {
+      // Cast the row because execute() returns untyped Record<string, unknown>[]
+      const channelId = String(row.channel_id);
+      const guildId = String(row.guild_id);
+      await learnChannelCulture(channelId, guildId);
+    }
+  } catch (error) {
+    log.error({ error }, "Error in culture learning cycle");
+  }
+}
+
+let cultureInterval: NodeJS.Timeout | null = null;
+
+export function startCultureLearnerWorker(): void {
+  if (!config.AI_ANALYSIS_ENABLED) return;
+  if (cultureInterval) return;
+
+  // Run once on startup after 1 minute, then every 1 hour
+  setTimeout(() => {
+    runCultureLearningCycle().catch(e => log.error(e));
+  }, 60000);
+
+  cultureInterval = setInterval(() => {
+    runCultureLearningCycle().catch(e => log.error(e));
+  }, 1000 * 60 * 60); // Check every hour for channels that reached 12h expiry
+
+  log.info("Started background culture learner worker");
+}
@@ -15,6 +15,11 @@ import type {
 import { formatModerationTextEvidenceForPrompt } from "./indonesianTextNormalizer.js";
 import { llmChat, llmVision } from "./llmClient.js";
 import { buildSystemPrompt as buildSystemPromptModular } from "./moderationPrompt.js";
+import {
+  initializeUserReputation,
+  getUserRecentInfractions,
+} from "./userReputationStore.js";
+import { getChannelCulture } from "./channelCultureStore.js";
 import { logModerationAnalysis, logModerationError } from "./responseLogger.js";
 import {
  getStickerFromCache,
@@ -1077,11 +1082,33 @@ async function runTextOnlyBatch(
  const allResults: AnalysisResult[] = [];
  let lastRaw: unknown = null;

+  const channelId = targets.length > 0 ? targets[0].channel_id : "";
+  const guildId = targets.length > 0 ? targets[0].guild_id : "";
+  const channelCultureObj = channelId ? await getChannelCulture(channelId) : null;
+  const channelCulture = channelCultureObj ? channelCultureObj.culture_summary : undefined;
+
  // Run sub-batches sequentially to avoid rate limits
  for (let i = 0; i < subBatches.length; i++) {
    const batch = subBatches[i];
    const targetIds = batch.map((t) => t.id);

+    // Fetch user context for this batch
+    const userContexts = new Map<string, string>();
+    for (const msg of batch) {
+      if (!userContexts.has(msg.user_id)) {
+        const rep = await initializeUserReputation(msg.user_id, msg.guild_id);
+        const history = await getUserRecentInfractions(msg.user_id);
+        
+        let historyStr = "";
+        if (history.length > 0) {
+          historyStr = `\n  <user_history>\n${history.map(h => `    - Flagged for ${h.flags} (Severity: ${h.severity}) pada pesan: "${h.content}"`).join("\n")}\n  </user_history>`;
+        }
+        
+        const contextStr = `<user_reputation trust_score="${rep.trust_score}" clean_streak="${rep.clean_message_streak}" total_infractions="${rep.total_infractions}" />${historyStr}`;
+        userContexts.set(msg.user_id, contextStr);
+      }
+    }
+
    const buildContent = async (state: RetryState): Promise<string> => {
      const correction = state.lastParseError
        ? {
@@ -1097,6 +1124,7 @@ async function runTextOnlyBatch(
        mode: "text",
        correction,
        correctedExamples,
+        channelCulture,
      });

      const messagesBlock = batch
@@ -1116,9 +1144,10 @@ async function runTextOnlyBatch(
            .filter(Boolean)
            .join("\n");
          const webContext = urlContexts ? `\n${urlContexts}` : "";
+          const userCtx = userContexts.get(msg.user_id) ?? "";

          // XML delimiters wrap each message for prompt safety (R1)
-          return `<message id="${msg.id}" user="${msg.username}">${content}${textContext}${webContext}</message>`;
+          return `<message id="${msg.id}" user="${msg.username}">\n  ${userCtx}\n  <content>${content}</content>${textContext}${webContext}\n</message>`;
        })
        .join("\n");

@@ -1616,8 +1645,20 @@ async function _runSingleMediaAnalysis(
    .filter(Boolean)
    .join(" ");

+  const channelId = target.channel_id;
+  const channelCultureObj = channelId ? await getChannelCulture(channelId) : null;
+  const channelCulture = channelCultureObj ? channelCultureObj.culture_summary : undefined;
+
+  const rep = await initializeUserReputation(target.user_id, target.guild_id);
+  const history = await getUserRecentInfractions(target.user_id);
+  let historyStr = "";
+  if (history.length > 0) {
+    historyStr = `\n  <user_history>\n${history.map(h => `    - Flagged for ${h.flags} (Severity: ${h.severity}) pada pesan: "${h.content}"`).join("\n")}\n  </user_history>`;
+  }
+  const userCtx = `<user_reputation trust_score="${rep.trust_score}" clean_streak="${rep.clean_message_streak}" total_infractions="${rep.total_infractions}" />${historyStr}`;
+
  // XML delimiters wrap the message content (R1)
-  const messageBlock = `<message id="${target.id}" user="${target.username}">${content}${mediaContext ? ` ${mediaContext}` : ""}${textContext}${webContext}${mediaAnalysisContext}</message>`;
+  const messageBlock = `<message id="${target.id}" user="${target.username}">\n  ${userCtx}\n  <content>${content}</content>${mediaContext ? ` ${mediaContext}` : ""}${textContext}${webContext}${mediaAnalysisContext}\n</message>`;

  // Modular system prompt with XML delimiters (R1, R7, R8)
  const correctedExamples = await buildCorrectedFewShotExamples();
@@ -1625,6 +1666,7 @@ async function _runSingleMediaAnalysis(
    contextText,
    mode: "mixed",
    correctedExamples,
+    channelCulture,
  });

  const userContent = `${systemText}\n\n<messages_to_analyze>\n${messageBlock}\n</messages_to_analyze>`;
@@ -395,6 +395,10 @@ export interface BuildSystemPromptOptions {
   * examples. Injected between static examples and output instructions.
   */
  correctedExamples?: string;
+  /**
+   * Formatted XML block containing the AI-generated channel culture summary.
+   */
+  channelCulture?: string;
 }

 export function buildSystemPrompt(options: BuildSystemPromptOptions): string {
@@ -404,6 +408,7 @@ export function buildSystemPrompt(options: BuildSystemPromptOptions): string {
    includeMediaInstructions,
    correction,
    correctedExamples,
+    channelCulture,
  } = options;

  // Backward compatibility: if mode is not set but includeMediaInstructions is,
@@ -433,6 +438,13 @@ export function buildSystemPrompt(options: BuildSystemPromptOptions): string {
    parts.push(correctedExamples);
  }

+  // Channel Culture Injection (Learning)
+  if (channelCulture) {
+    parts.push(`## Kultur Channel (Pembelajaran AI)\n${channelCulture}`);
+  }
+
+  parts.push(`## Konteks Pengguna (Ingatan & Kebijaksanaan)\nSetiap pesan mungkin memiliki tag <user_reputation> dan <user_history>. *Gunakan Kebijaksanaan: Jika trust_score tinggi, beri benefit of the doubt pada ambiguitas. Jika trust_score rendah dan memiliki riwayat pelanggaran serupa, jadilah lebih tegas.*`);
+
  parts.push(OUTPUT_INSTRUCTIONS);

  // XML-delimited context — prevents prompt injection
@@ -0,0 +1,154 @@
+import { eq, and, desc } from "drizzle-orm";
+import { getDatabase } from "../../shared/database/drizzle.js";
+import {
+  userReputationsTable,
+  messagesTable,
+  UserReputation,
+} from "../../shared/database/schema.js";
+
+/**
+ * Ensures a user reputation record exists.
+ */
+export async function initializeUserReputation(
+  userId: string,
+  guildId: string,
+): Promise<UserReputation> {
+  const db = getDatabase();
+  const existing = await db
+    .select()
+    .from(userReputationsTable)
+    .where(eq(userReputationsTable.user_id, userId))
+    .limit(1);
+
+  if (existing.length > 0) {
+    return existing[0];
+  }
+
+  const [inserted] = await db
+    .insert(userReputationsTable)
+    .values({
+      user_id: userId,
+      guild_id: guildId,
+      trust_score: 50,
+      clean_message_streak: 0,
+      total_infractions: 0,
+      created_at: Date.now(),
+      updated_at: Date.now(),
+    })
+    .onConflictDoNothing()
+    .returning();
+
+  if (!inserted) {
+    // If concurrent insert happened
+    const retry = await db
+      .select()
+      .from(userReputationsTable)
+      .where(eq(userReputationsTable.user_id, userId))
+      .limit(1);
+    return retry[0];
+  }
+
+  return inserted;
+}
+
+/**
+ * Fetch a user's reputation score. Returns default 50 if none exists.
+ */
+export async function getUserReputation(userId: string): Promise<UserReputation | null> {
+  const db = getDatabase();
+  const existing = await db
+    .select()
+    .from(userReputationsTable)
+    .where(eq(userReputationsTable.user_id, userId))
+    .limit(1);
+
+  return existing[0] || null;
+}
+
+/**
+ * Increment the clean message streak and update trust score if threshold is met.
+ */
+export async function recordCleanMessage(userId: string, guildId: string): Promise<void> {
+  const rep = await initializeUserReputation(userId, guildId);
+  const db = getDatabase();
+  let newStreak = rep.clean_message_streak + 1;
+  let newScore = rep.trust_score;
+
+  // Every 100 clean messages, give +2 trust score up to 100
+  if (newStreak >= 100) {
+    newScore = Math.min(100, newScore + 2);
+    newStreak = 0;
+  }
+
+  await db
+    .update(userReputationsTable)
+    .set({
+      clean_message_streak: newStreak,
+      trust_score: newScore,
+      updated_at: Date.now(),
+    })
+    .where(eq(userReputationsTable.user_id, userId));
+}
+
+/**
+ * Apply an infraction penalty to a user.
+ */
+export async function recordInfraction(
+  userId: string,
+  guildId: string,
+  severity: "low" | "medium" | "high" | "critical",
+): Promise<void> {
+  const rep = await initializeUserReputation(userId, guildId);
+  const db = getDatabase();
+  let penalty = 0;
+  switch (severity) {
+    case "low":
+      penalty = 2;
+      break;
+    case "medium":
+      penalty = 5;
+      break;
+    case "high":
+      penalty = 15;
+      break;
+    case "critical":
+      penalty = 30;
+      break;
+  }
+
+  const newScore = Math.max(0, rep.trust_score - penalty);
+
+  await db
+    .update(userReputationsTable)
+    .set({
+      trust_score: newScore,
+      clean_message_streak: 0, // Reset streak on infraction
+      total_infractions: rep.total_infractions + 1,
+      last_infraction_at: Date.now(),
+      updated_at: Date.now(),
+    })
+    .where(eq(userReputationsTable.user_id, userId));
+}
+
+/**
+ * Fetch a user's past N flagged messages for context injection.
+ */
+export async function getUserRecentInfractions(userId: string, limit: number = 3) {
+  const db = getDatabase();
+  return await db
+    .select({
+      content: messagesTable.content,
+      flags: messagesTable.ai_moderation_flags,
+      severity: messagesTable.ai_severity,
+      created_at: messagesTable.created_at,
+    })
+    .from(messagesTable)
+    .where(
+      and(
+        eq(messagesTable.user_id, userId),
+        eq(messagesTable.ai_status, "flagged")
+      )
+    )
+    .orderBy(desc(messagesTable.created_at))
+    .limit(limit);
+}
@@ -248,6 +248,45 @@ export const pgVoiceRecordingsTable = pgTable(
  }),
 );

+/**
+ * User Reputations Table (PostgreSQL)
+ * Tracks user trust score and infractions to provide context to AI.
+ */
+export const pgUserReputationsTable = pgTable(
+  "user_reputations",
+  {
+    user_id: pgText("user_id").primaryKey(),
+    guild_id: pgText("guild_id").notNull(),
+    trust_score: pgInteger("trust_score").notNull().default(50),
+    clean_message_streak: pgInteger("clean_message_streak").notNull().default(0),
+    total_infractions: pgInteger("total_infractions").notNull().default(0),
+    last_infraction_at: pgBigint("last_infraction_at", { mode: "number" }),
+    created_at: pgBigint("created_at", { mode: "number" }).notNull(),
+    updated_at: pgBigint("updated_at", { mode: "number" }).notNull(),
+  },
+  (table) => ({
+    guildIdx: pgIndex("idx_user_reputations_guild_id").on(table.guild_id),
+    scoreIdx: pgIndex("idx_user_reputations_trust_score").on(table.trust_score),
+  }),
+);
+
+/**
+ * Channel Cultures Table (PostgreSQL)
+ * Stores AI-generated summaries of channel norms and slang to inject as context.
+ */
+export const pgChannelCulturesTable = pgTable(
+  "channel_cultures",
+  {
+    channel_id: pgText("channel_id").primaryKey(),
+    guild_id: pgText("guild_id").notNull(),
+    culture_summary: pgText("culture_summary").notNull(),
+    last_analyzed_at: pgBigint("last_analyzed_at", { mode: "number" }).notNull(),
+  },
+  (table) => ({
+    guildIdx: pgIndex("idx_channel_cultures_guild_id").on(table.guild_id),
+  }),
+);
+
 /**
 * Message Reviews Table (PostgreSQL)
 * Tracks manual reviews of messages flagged by AI moderation
@@ -464,6 +503,8 @@ export const retentionPoliciesTable = pgRetentionPoliciesTable;
 export const textAnalysisCacheTable = pgTextAnalysisCacheTable;
 export const stickerCacheTable = pgStickerCacheTable;
 export const correctedModerationsTable = pgCorrectedModerationsTable;
+export const userReputationsTable = pgUserReputationsTable;
+export const channelCulturesTable = pgChannelCulturesTable;

 // Export table types for use in queries
 export type MuxerJob = typeof muxerJobsTable.$inferSelect;
@@ -499,3 +540,9 @@ export type StickerCacheInsert = typeof stickerCacheTable.$inferInsert;
 export type CorrectedModeration = typeof correctedModerationsTable.$inferSelect;
 export type CorrectedModerationInsert =
  typeof correctedModerationsTable.$inferInsert;
+
+export type UserReputation = typeof userReputationsTable.$inferSelect;
+export type UserReputationInsert = typeof userReputationsTable.$inferInsert;
+
+export type ChannelCulture = typeof channelCulturesTable.$inferSelect;
+export type ChannelCultureInsert = typeof channelCulturesTable.$inferInsert;