feat(chatbot): per-user history via X-User-Id + agentic tools calling

Backend: - New chatbot.tools.ts: 4 tools (get_server_stats, get_top_channels, get_recent_activity, get_top_flagged) with real DB executors - chatbot.service: agentic loop — stream:true, parse SSE, execute tool_calls, feed results back, up to 4 rounds - controller: resolve userId from X-User-Id header (no-login device uuid) with auth middleware precedence; history/clear scoped per user Frontend: - use-chatbot-user: mint UUID in localStorage, send as X-User-Id - chatbotApi.send/getHistory/clearHistory accept userId header - client.ts: apiRequest supports custom headers per call - provider: history load + send + clear keyed to device user id
2026-08-03 06:24:19 +07:00
parent 7513681b4b
commit d1c1f3e4a7
7 changed files with 506 additions and 56 deletions
@@ -9,6 +9,18 @@ interface AuthenticatedRequest extends Request {
  userId?: string;
 }

+/**
+ * Resolve the actor id for a request. Frontend (no-login) sends a per-device
+ * UUID via X-User-Id so chat history stays isolated per visitor; a registered
+ * auth middleware userId takes precedence when present.
+ */
+function resolveUserId(req: Request): string {
+  const authId = (req as AuthenticatedRequest).userId;
+  if (authId) return authId;
+  const header = (req.headers["x-user-id"] as string | undefined)?.trim();
+  return header || "anonymous";
+}
+
 export const handleChatbotChat = asyncHandler(
  async (req: Request, res: Response) => {
    const { message, context } = req.body as {
@@ -24,8 +36,8 @@ export const handleChatbotChat = asyncHandler(
      });
    }

-    // Get user ID from auth middleware (if available)
-    const userId = (req as AuthenticatedRequest).userId || "anonymous";
+    // Get user ID from X-User-Id header (no-login device uuid) or auth
+    const userId = resolveUserId(req);

    logger.debug(
      { userId, messageLength: message.length, context },
@@ -59,7 +71,7 @@ export const handleChatbotChat = asyncHandler(

 export const getChatbotHistory = asyncHandler(
  async (req: Request, res: Response) => {
-    const userId = (req as AuthenticatedRequest).userId || "anonymous";
+    const userId = resolveUserId(req);
    const limit = Math.min(parseInt(req.query.limit as string, 10) || 50, 100);

    const history = await chatbotService.getChatHistory(userId, limit);
@@ -73,7 +85,7 @@ export const getChatbotHistory = asyncHandler(

 export const clearChatbotHistory = asyncHandler(
  async (req: Request, res: Response) => {
-    const userId = (req as AuthenticatedRequest).userId || "anonymous";
+    const userId = resolveUserId(req);

    await chatbotService.clearChatHistory(userId);

@@ -6,6 +6,7 @@ import type {
  SaveConversationInput,
 } from "./chatbot.repository.js";
 import { chatbotRepository } from "./chatbot.repository.js";
+import { executeTool, tools } from "./chatbot.tools.js";

 const logger = createChildLogger("chatbot.service");

@@ -118,41 +119,104 @@ Gaya ngobrol:
    try {
      const { default: axios } = await import("axios");

-      // Gateway tidak handle role system — gabung konteks ke user message
+      // Gateway tidak handle role system — gabung konteks ke user message.
+      // The system section stays visible to the model as the first user turn.
      const contextPrefixed = `${systemPrompt}\n\nPertanyaan user: ${userMessage}`;

-      const messages: Array<{ role: "user" | "assistant"; content: string }> = [
-        ...history,
-        { role: "user", content: contextPrefixed },
-      ];
+      // Seed conversation: prior turns + current question.
+      const messages: Array<
+        | { role: "user" | "assistant"; content: string }
+        | {
+            role: "assistant";
+            content: string | null;
+            tool_calls: Array<{
+              id: string;
+              type: "function";
+              function: { name: string; arguments: string };
+            }>;
+          }
+        | { role: "tool"; tool_call_id: string; content: string }
+      > = [...history, { role: "user", content: contextPrefixed }];

-      const response = await axios.post(
-        `${baseUrl}/chat/completions`,
-        {
-          model,
-          messages,
-          max_tokens: 500,
-          temperature: 0.4,
-        },
-        {
-          headers: {
-            Authorization: `Bearer ${apiKey}`,
-            "Content-Type": "application/json",
+      // ── Agentic tool loop ─────────────────────────────────────────
+      const MAX_TOOL_ROUNDS = 4;
+      for (let round = 0; round <= MAX_TOOL_ROUNDS; round += 1) {
+        const response = await axios.post(
+          `${baseUrl}/chat/completions`,
+          {
+            model,
+            messages,
+            tools,
+            tool_choice: "auto",
+            max_tokens: 600,
+            temperature: 0.4,
+            stream: true,
          },
-          timeout: 30_000,
-        },
-      );
+          {
+            headers: {
+              Authorization: `Bearer ${apiKey}`,
+              "Content-Type": "application/json",
+            },
+            timeout: 45_000,
+            // 9router returns SSE even without stream:true; force stream:true
+            // in the body and read the raw SSE text.
+            responseType: "text",
+          },
+        );

-      const result = response.data as {
-        choices?: Array<{ message?: { content?: string } }>;
-      };
-      const content = result?.choices?.[0]?.message?.content?.trim();
+        // Parse SSE `data:` lines → content + tool_calls.
+        const { content, toolCalls } = this.parseSse(response.data as string);

-      if (content) {
-        return content;
+        logger.debug(
+          {
+            round,
+            hasToolCalls: toolCalls.length > 0,
+            toolNames: toolCalls.map((t) => t.name),
+          },
+          "LLM round parsed",
+        );
+
+        if (toolCalls.length > 0) {
+          // Execute each tool, append tool results, continue loop.
+          for (const tc of toolCalls) {
+            messages.push({
+              role: "assistant",
+              content: null,
+              tool_calls: [
+                {
+                  id: tc.id,
+                  type: "function",
+                  function: { name: tc.name, arguments: tc.arguments },
+                },
+              ],
+            });
+            let result = "";
+            try {
+              result = await executeTool(tc.name, tc.args);
+            } catch (e) {
+              result = `Tool error: ${(e as Error).message}`;
+            }
+            messages.push({
+              role: "tool",
+              tool_call_id: tc.id,
+              content: result,
+            });
+          }
+          if (round === MAX_TOOL_ROUNDS) {
+            logger.warn("Hit max tool rounds; returning what we have");
+          }
+          continue;
+        }
+
+        if (content?.trim()) {
+          return content.trim();
+        }
+
+        logger.warn("LLM returned empty response (no tools, no content)");
+        return this.fallbackResponse(userMessage);
      }

-      logger.warn({ response: result }, "LLM returned empty response");
+      logger.warn("Tool loop exhausted without final content");
      return this.fallbackResponse(userMessage);
    } catch (error) {
      logger.warn({ error }, "LLM call failed, using fallback response");
@@ -160,6 +224,93 @@ Gaya ngobrol:
    }
  }

+  /**
+   * Parse an SSE stream body into accumulated content + any tool_calls.
+   * 9router (and most OpenAI-compatible routers) emit `data: {json}` lines
+   * even when stream is only implied; we must collect deltas manually.
+   */
+  private parseSse(body: string): {
+    content: string;
+    toolCalls: Array<{
+      id: string;
+      name: string;
+      arguments: string;
+      args: Record<string, unknown>;
+    }>;
+  } {
+    const contentParts: string[] = [];
+    const toolById = new Map<
+      string,
+      { id: string; name: string; arguments: string }
+    >();
+
+    const lines = body.split("\n");
+    for (const rawLine of lines) {
+      const line = rawLine.trim();
+      if (!line.startsWith("data:")) continue;
+      const payload = line.slice(5).trim();
+      if (!payload || payload === "[DONE]") continue;
+      try {
+        const json = JSON.parse(payload) as {
+          choices?: Array<{
+            delta?: {
+              content?: string;
+              tool_calls?: Array<{
+                id?: string;
+                index?: number;
+                type?: string;
+                function?: { name?: string; arguments?: string };
+              }>;
+            };
+            finish_reason?: string | null;
+          }>;
+        };
+        const delta = json.choices?.[0]?.delta;
+        if (!delta) continue;
+        if (delta.content) contentParts.push(delta.content);
+        if (delta.tool_calls) {
+          for (const tc of delta.tool_calls) {
+            const idx = String(tc.index ?? 0);
+            const cur = toolById.get(idx) ?? {
+              id: tc.id ?? "",
+              name: "",
+              arguments: "",
+            };
+            // Keep the first non-empty id for this call index.
+            if (tc.id && !cur.id) cur.id = tc.id;
+            if (tc.function?.name) cur.name += tc.function.name;
+            if (tc.function?.arguments) cur.arguments += tc.function.arguments;
+            toolById.set(idx, cur);
+          }
+        }
+      } catch {
+        // Skip malformed lines (keepalives, etc.)
+      }
+    }
+
+    // Build a de-duplicated id for any call the stream never assigned one.
+    let fallbackId = 0;
+    const toolCalls = Array.from(toolById.values()).map((tc) => {
+      const id = tc.id || `tool_${fallbackId++}_${Date.now()}`;
+      return {
+        id,
+        name: tc.name,
+        arguments: tc.arguments,
+        args: this.safeJsonParse(tc.arguments),
+      };
+    });
+
+    return { content: contentParts.join(""), toolCalls };
+  }
+
+  private safeJsonParse(s: string): Record<string, unknown> {
+    try {
+      return JSON.parse(s) as Record<string, unknown>;
+    } catch {
+      return {};
+    }
+  }
+
  private fallbackResponse(input: string): string {
    const lower = input.toLowerCase();

@@ -0,0 +1,232 @@
+import { sql } from "drizzle-orm";
+import { getDatabase } from "../../shared/database/index.js";
+
+/**
+ * Tools the chatbot LLM can call. Definitions describe the schema to the
+ * model; the executor implements each one against the real database.
+ * This turns the chatbot from "blind stats guesser" into an agent that
+ * pulls real, current server data on demand.
+ */
+
+export type ToolResult = string;
+
+/** JSON schema for a tool definition (OpenAI function-calling format). */
+export interface ToolDef {
+  type: "function";
+  function: {
+    name: string;
+    description: string;
+    parameters: {
+      type: "object";
+      properties: Record<string, unknown>;
+      required?: string[];
+    };
+  };
+}
+
+export const tools: ToolDef[] = [
+  {
+    type: "function",
+    function: {
+      name: "get_server_stats",
+      description:
+        "Ambil statistik ringkas server/guild saat ini: total pesan, user aktif, jumlah pesan flagged, dan jumlah warning. Panggil ini untuk menjawab pertanyaan umum tentang kondisi server. Opsional fill guild_id untuk scope ke guild tertentu, channel_id untuk scope ke channel.",
+      parameters: {
+        type: "object",
+        properties: {
+          guildId: {
+            type: "string",
+            description: "ID guild/server (opsional). Kosongkan = semua data.",
+          },
+          channelId: {
+            type: "string",
+            description: "ID channel (opsional).",
+          },
+        },
+      },
+    },
+  },
+  {
+    type: "function",
+    function: {
+      name: "get_top_channels",
+      description:
+        "Ambil daftar channel paling aktif (jumlah pesan terbanyak) di server. Panggil buat jawab 'channel mana paling ramai' atau aktivitas per-channel.",
+      parameters: {
+        type: "object",
+        properties: {
+          guildId: {
+            type: "string",
+            description: "ID server (opsional).",
+          },
+          limit: {
+            type: "number",
+            description: "Jumlah channel teratas (default 5, max 10).",
+          },
+        },
+      },
+    },
+  },
+  {
+    type: "function",
+    function: {
+      name: "get_recent_activity",
+      description:
+        "Ambil aktivitas/pesan terbaru di server: siapa yang baru ngomong, di channel mana, jam berapa. Panggil buat jawaban soal 'lagi ngapain' / aktivitas terbaru di server.",
+      parameters: {
+        type: "object",
+        properties: {
+          guildId: {
+            type: "string",
+            description: "ID server (opsional).",
+          },
+          limit: {
+            type: "number",
+            description: "Jumlah pesan terakhir (default 5).",
+          },
+        },
+      },
+    },
+  },
+  {
+    type: "function",
+    function: {
+      name: "get_top_flagged",
+      description:
+        "Ambil pesan yang paling sering di-flag atau kena warning. Panggil buat jawab soal pesan bermasalah / moderator.",
+      parameters: {
+        type: "object",
+        properties: {
+          guildId: {
+            type: "string",
+            description: "ID server (opsional).",
+          },
+          limit: {
+            type: "number",
+            description: "Jumlah pesan (default 5).",
+          },
+        },
+      },
+    },
+  },
+];
+
+/** Executes a tool call against the real DB and returns a readable result. */
+export async function executeTool(
+  name: string,
+  args: Record<string, unknown>,
+): Promise<string> {
+  const guildId =
+    typeof args.guildId === "string" && args.guildId ? args.guildId : undefined;
+  const channelId =
+    typeof args.channelId === "string" && args.channelId
+      ? args.channelId
+      : undefined;
+  const limitRaw =
+    typeof args.limit === "number" ? args.limit : Number(args.limit) || 5;
+  const limit = Math.min(Math.max(1, Math.round(limitRaw)), 10);
+
+  try {
+    switch (name) {
+      case "get_server_stats":
+        return await serverStats(guildId, channelId);
+      case "get_top_channels":
+        return await topChannels(guildId, limit);
+      case "get_recent_activity":
+        return await recentActivity(guildId, limit);
+      case "get_top_flagged":
+        return await topFlagged(guildId, limit);
+      default:
+        return `Unknown tool: ${name}`;
+    }
+  } catch (error) {
+    // Best-effort: if a tool fails, return readable error instead of crashing
+    return `Terjadi kesalahan saat ambil data: ${(error as Error).message ?? "unknown"}`;
+  }
+}
+
+// ── Tool executors ──────────────────────────────────────────
+
+async function serverStats(
+  guildId?: string,
+  channelId?: string,
+): Promise<string> {
+  const db = getDatabase();
+  const conditions: string[] = [];
+  if (guildId) conditions.push(`guild_id = '${guildId}'`);
+  if (channelId) conditions.push(`channel_id = '${channelId}'`);
+  const cond = conditions.length ? `WHERE ${conditions.join(" AND ")}` : "";
+
+  const result = await db.execute(
+    sql.raw(
+      `SELECT COUNT(*)::int AS total_messages,
+              COUNT(DISTINCT user_id)::int AS active_users,
+              COUNT(*) FILTER (WHERE ai_status = 'flagged')::int AS flagged,
+              COUNT(*) FILTER (WHERE ai_status = 'warn')::int AS warned
+       FROM messages ${cond}`,
+    ),
+  );
+  const rows =
+    (result as unknown as { rows: Record<string, unknown>[] }).rows ?? [];
+  const r = rows[0] ?? {};
+  return JSON.stringify({
+    total_messages: r.total_messages ?? 0,
+    active_users: r.active_users ?? 0,
+    flagged: r.flagged ?? 0,
+    warned: r.warned ?? 0,
+  });
+}
+
+async function topChannels(guildId?: string, limit = 5): Promise<string> {
+  const db = getDatabase();
+  const conditions: string[] = [];
+  if (guildId) conditions.push(`guild_id = '${guildId}'`);
+  const cond = conditions.length ? `WHERE ${conditions.join(" AND ")}` : "";
+
+  const result = await db.execute(
+    sql.raw(
+      `SELECT channel_id,
+              COUNT(*)::int AS count
+       FROM messages ${cond}
+       GROUP BY channel_id
+       ORDER BY count DESC
+       LIMIT ${limit}`,
+    ),
+  );
+  const rows = (result as unknown as { rows: unknown[] }).rows ?? [];
+  return JSON.stringify(rows.slice(0, limit));
+}
+
+async function recentActivity(guildId?: string, limit = 5): Promise<string> {
+  const db = getDatabase();
+  const conditions: string[] = [];
+  if (guildId) conditions.push(`guild_id = '${guildId}'`);
+  const cond = conditions.length ? `WHERE ${conditions.join(" AND ")}` : "";
+
+  const result = await db.execute(
+    sql.raw(
+      `SELECT username, content, channel_id, created_at
+       FROM messages ${cond}
+       ORDER BY created_at DESC
+       LIMIT ${limit}`,
+    ),
+  );
+  return JSON.stringify((result as unknown as { rows: unknown[] }).rows ?? []);
+}
+
+async function topFlagged(guildId?: string, limit = 5): Promise<string> {
+  const db = getDatabase();
+  const conditions = ["ai_status IN ('flagged', 'warn')"];
+  if (guildId) conditions.push(`guild_id = '${guildId}'`);
+  const cond = `WHERE ${conditions.join(" AND ")}`;
+
+  const result = await db.execute(
+    sql.raw(
+      `SELECT username, content, channel_id, ai_status, created_at
+       FROM messages ${cond}
+       ORDER BY created_at DESC
+       LIMIT ${limit}`,
+    ),
+  );
+  return JSON.stringify((result as unknown as { rows: unknown[] }).rows ?? []);
+}