diff --git a/services/discord-gateway/src/modules/ai-moderation/aiAnalyzer.ts b/services/discord-gateway/src/modules/ai-moderation/aiAnalyzer.ts index 17d3981..f82b3b3 100644 --- a/services/discord-gateway/src/modules/ai-moderation/aiAnalyzer.ts +++ b/services/discord-gateway/src/modules/ai-moderation/aiAnalyzer.ts @@ -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( diff --git a/services/discord-gateway/src/modules/ai-moderation/channelCultureStore.ts b/services/discord-gateway/src/modules/ai-moderation/channelCultureStore.ts new file mode 100644 index 0000000..0156fb6 --- /dev/null +++ b/services/discord-gateway/src/modules/ai-moderation/channelCultureStore.ts @@ -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 { + 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 { + 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(), + }, + }); +} diff --git a/services/discord-gateway/src/modules/ai-moderation/cultureLearner.ts b/services/discord-gateway/src/modules/ai-moderation/cultureLearner.ts new file mode 100644 index 0000000..a1e30e8 --- /dev/null +++ b/services/discord-gateway/src/modules/ai-moderation/cultureLearner.ts @@ -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 { + 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: + +${messagesText} + + +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 { + 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[] + 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"); +} diff --git a/services/discord-gateway/src/modules/ai-moderation/llmModerationClient.ts b/services/discord-gateway/src/modules/ai-moderation/llmModerationClient.ts index e22378d..17f16a0 100644 --- a/services/discord-gateway/src/modules/ai-moderation/llmModerationClient.ts +++ b/services/discord-gateway/src/modules/ai-moderation/llmModerationClient.ts @@ -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(); + 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 \n${history.map(h => ` - Flagged for ${h.flags} (Severity: ${h.severity}) pada pesan: "${h.content}"`).join("\n")}\n `; + } + + const contextStr = `${historyStr}`; + userContexts.set(msg.user_id, contextStr); + } + } + const buildContent = async (state: RetryState): Promise => { 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 `${content}${textContext}${webContext}`; + return `\n ${userCtx}\n ${content}${textContext}${webContext}\n`; }) .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 \n${history.map(h => ` - Flagged for ${h.flags} (Severity: ${h.severity}) pada pesan: "${h.content}"`).join("\n")}\n `; + } + const userCtx = `${historyStr}`; + // XML delimiters wrap the message content (R1) - const messageBlock = `${content}${mediaContext ? ` ${mediaContext}` : ""}${textContext}${webContext}${mediaAnalysisContext}`; + const messageBlock = `\n ${userCtx}\n ${content}${mediaContext ? ` ${mediaContext}` : ""}${textContext}${webContext}${mediaAnalysisContext}\n`; // 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\n${messageBlock}\n`; diff --git a/services/discord-gateway/src/modules/ai-moderation/moderationPrompt.ts b/services/discord-gateway/src/modules/ai-moderation/moderationPrompt.ts index ada854e..7ff503a 100644 --- a/services/discord-gateway/src/modules/ai-moderation/moderationPrompt.ts +++ b/services/discord-gateway/src/modules/ai-moderation/moderationPrompt.ts @@ -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 dan . *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 diff --git a/services/discord-gateway/src/modules/ai-moderation/userReputationStore.ts b/services/discord-gateway/src/modules/ai-moderation/userReputationStore.ts new file mode 100644 index 0000000..d89c88f --- /dev/null +++ b/services/discord-gateway/src/modules/ai-moderation/userReputationStore.ts @@ -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 { + 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 { + 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 { + 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 { + 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); +} diff --git a/services/discord-gateway/src/shared/database/schema.ts b/services/discord-gateway/src/shared/database/schema.ts index 9987be2..0c03305 100644 --- a/services/discord-gateway/src/shared/database/schema.ts +++ b/services/discord-gateway/src/shared/database/schema.ts @@ -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;