|
| 1 | +// Streaming SSE bridge for the AI chatbot. |
| 2 | +// 1. Extracts userId from the JWT access-token cookie (claims decoded |
| 3 | +// without signature verification — good enough for a UX-level RBAC |
| 4 | +// hint; every downstream BFF route still enforces real auth/permissions). |
| 5 | +// 2. Runs guardrails (Layer 3 pattern filter + Layer 1/2 rate-limit stubs). |
| 6 | +// 3. Builds system prompt + trimmed history + tool defs. |
| 7 | +// 4. Streams the DeepSeek response back to the browser as SSE. |
| 8 | +// 5. Executes tool calls, then makes a second DeepSeek call with the result. |
| 9 | +// 6. Fires an async, non-blocking audit log request. |
| 10 | + |
| 11 | +import { NextRequest } from "next/server" |
| 12 | +import { |
| 13 | + streamDeepSeek, |
| 14 | + SYSTEM_PROMPT, |
| 15 | + type ChatMessage as DeepSeekMsg, |
| 16 | + type ToolCall, |
| 17 | +} from "@/lib/chatbot/deepseek-client" |
| 18 | +import { CHATBOT_TOOLS } from "@/lib/chatbot/tools" |
| 19 | +import { executeTool } from "@/lib/chatbot/tool-executor" |
| 20 | +import { checkGuardrails } from "@/lib/chatbot/guardrails" |
| 21 | +import { AUTH_COOKIES } from "@/lib/auth/config" |
| 22 | + |
| 23 | +export const runtime = "nodejs" |
| 24 | +// Streaming responses must never be cached or statically optimized. |
| 25 | +export const dynamic = "force-dynamic" |
| 26 | + |
| 27 | +interface ChatbotRequestBody { |
| 28 | + message: string |
| 29 | + history?: Array<{ role: string; content: string }> |
| 30 | +} |
| 31 | + |
| 32 | +// Decodes the userId claim from a JWT payload without verifying the |
| 33 | +// signature — this route only needs the user identity for scoping tool |
| 34 | +// calls and audit logs; every downstream BFF request still carries the |
| 35 | +// real cookie and is authorized independently. |
| 36 | +function extractUserId(token: string): string { |
| 37 | + try { |
| 38 | + const payloadSegment = token.split(".")[1] |
| 39 | + if (!payloadSegment) return "" |
| 40 | + const payload = JSON.parse(Buffer.from(payloadSegment, "base64url").toString("utf-8")) as { |
| 41 | + user_id?: string |
| 42 | + sub?: string |
| 43 | + } |
| 44 | + return payload.user_id ?? payload.sub ?? "" |
| 45 | + } catch { |
| 46 | + return "" |
| 47 | + } |
| 48 | +} |
| 49 | + |
| 50 | +export async function POST(request: NextRequest) { |
| 51 | + const token = request.cookies.get(AUTH_COOKIES.ACCESS_TOKEN)?.value |
| 52 | + if (!token) { |
| 53 | + return new Response("Unauthorized", { status: 401 }) |
| 54 | + } |
| 55 | + |
| 56 | + const userId = extractUserId(token) |
| 57 | + if (!userId) { |
| 58 | + return new Response("Unauthorized", { status: 401 }) |
| 59 | + } |
| 60 | + |
| 61 | + let body: ChatbotRequestBody |
| 62 | + try { |
| 63 | + body = (await request.json()) as ChatbotRequestBody |
| 64 | + } catch { |
| 65 | + return new Response("Invalid JSON body", { status: 400 }) |
| 66 | + } |
| 67 | + |
| 68 | + const { message, history = [] } = body |
| 69 | + if (!message || typeof message !== "string") { |
| 70 | + return new Response("message is required", { status: 400 }) |
| 71 | + } |
| 72 | + |
| 73 | + const encoder = new TextEncoder() |
| 74 | + |
| 75 | + // Layer 3: prompt injection / jailbreak pattern filter. |
| 76 | + const guardrail = await checkGuardrails({ message, userId }) |
| 77 | + if (guardrail.blocked) { |
| 78 | + const blockedStream = new ReadableStream<Uint8Array>({ |
| 79 | + start(controller) { |
| 80 | + controller.enqueue( |
| 81 | + encoder.encode(`data: ${JSON.stringify({ type: "blocked", reason: guardrail.reason })}\n\n`) |
| 82 | + ) |
| 83 | + controller.enqueue(encoder.encode("data: [DONE]\n\n")) |
| 84 | + controller.close() |
| 85 | + }, |
| 86 | + }) |
| 87 | + return new Response(blockedStream, { |
| 88 | + headers: { |
| 89 | + "Content-Type": "text/event-stream", |
| 90 | + "Cache-Control": "no-cache, no-transform", |
| 91 | + Connection: "keep-alive", |
| 92 | + }, |
| 93 | + }) |
| 94 | + } |
| 95 | + |
| 96 | + const maxHistory = parseInt(process.env.CHATBOT_MAX_HISTORY_MESSAGES ?? "20", 10) |
| 97 | + const maxResponseTokens = parseInt(process.env.CHATBOT_MAX_RESPONSE_TOKENS ?? "2000", 10) |
| 98 | + |
| 99 | + // Layer 4: system prompt is always first and immutable. |
| 100 | + const messages: DeepSeekMsg[] = [ |
| 101 | + { role: "system", content: SYSTEM_PROMPT }, |
| 102 | + ...history.slice(-maxHistory).map((m) => ({ role: m.role as DeepSeekMsg["role"], content: m.content })), |
| 103 | + { role: "user", content: message }, |
| 104 | + ] |
| 105 | + |
| 106 | + const toolsCalled: string[] = [] |
| 107 | + const requestTokenEstimate = messages.reduce((sum, m) => sum + Math.ceil(m.content.length / 4), 0) |
| 108 | + const sessionId = request.headers.get("x-chatbot-session-id") ?? "unknown" |
| 109 | + const cookieHeader = request.headers.get("cookie") ?? "" |
| 110 | + |
| 111 | + const stream = new ReadableStream<Uint8Array>({ |
| 112 | + async start(controller) { |
| 113 | + let closed = false |
| 114 | + const enqueue = (data: unknown) => { |
| 115 | + if (closed) return |
| 116 | + try { |
| 117 | + controller.enqueue(encoder.encode(`data: ${JSON.stringify(data)}\n\n`)) |
| 118 | + } catch { |
| 119 | + closed = true |
| 120 | + } |
| 121 | + } |
| 122 | + |
| 123 | + let responseTokens = 0 |
| 124 | + |
| 125 | + try { |
| 126 | + let pendingToolCall: Partial<ToolCall> | null = null |
| 127 | + |
| 128 | + for await (const chunk of streamDeepSeek(messages, CHATBOT_TOOLS, maxResponseTokens)) { |
| 129 | + if (chunk.type === "delta" && chunk.content) { |
| 130 | + responseTokens += 1 |
| 131 | + enqueue({ type: "delta", content: chunk.content }) |
| 132 | + } else if (chunk.type === "tool_call" && chunk.toolCall) { |
| 133 | + // Tool call arguments can arrive fragmented across chunks. |
| 134 | + if (!pendingToolCall) { |
| 135 | + pendingToolCall = { ...chunk.toolCall, function: { ...chunk.toolCall.function } } |
| 136 | + } else { |
| 137 | + if (chunk.toolCall.function?.name) { |
| 138 | + pendingToolCall.function!.name = chunk.toolCall.function.name |
| 139 | + } |
| 140 | + if (chunk.toolCall.function?.arguments) { |
| 141 | + pendingToolCall.function!.arguments = |
| 142 | + (pendingToolCall.function!.arguments ?? "") + chunk.toolCall.function.arguments |
| 143 | + } |
| 144 | + } |
| 145 | + } else if (chunk.type === "done") { |
| 146 | + if (pendingToolCall?.function?.name) { |
| 147 | + const toolName = pendingToolCall.function.name |
| 148 | + toolsCalled.push(toolName) |
| 149 | + enqueue({ type: "tool_call", name: toolName }) |
| 150 | + |
| 151 | + let toolArgs: unknown = {} |
| 152 | + try { |
| 153 | + toolArgs = JSON.parse(pendingToolCall.function.arguments ?? "{}") |
| 154 | + } catch { |
| 155 | + // invalid JSON args — executeTool's Zod validation will reject it |
| 156 | + } |
| 157 | + |
| 158 | + const toolResult = await executeTool(toolName, toolArgs, userId) |
| 159 | + enqueue({ type: "tool_result", name: toolName, data: toolResult.data, error: toolResult.error }) |
| 160 | + |
| 161 | + const messagesWithTool: DeepSeekMsg[] = [ |
| 162 | + ...messages, |
| 163 | + { role: "assistant", content: "", tool_call_id: pendingToolCall.id }, |
| 164 | + { |
| 165 | + role: "tool", |
| 166 | + content: JSON.stringify(toolResult.data ?? { error: toolResult.error }), |
| 167 | + tool_call_id: pendingToolCall.id ?? "", |
| 168 | + name: toolName, |
| 169 | + }, |
| 170 | + ] |
| 171 | + |
| 172 | + for await (const chunk2 of streamDeepSeek(messagesWithTool, [], maxResponseTokens)) { |
| 173 | + if (chunk2.type === "delta" && chunk2.content) { |
| 174 | + responseTokens += 1 |
| 175 | + enqueue({ type: "delta", content: chunk2.content }) |
| 176 | + } else if (chunk2.type === "done" || chunk2.type === "error") { |
| 177 | + break |
| 178 | + } |
| 179 | + } |
| 180 | + } |
| 181 | + |
| 182 | + // Fire-and-forget audit log — never blocks or fails the response. |
| 183 | + void fetch(new URL("/api/v1/chatbot/audit", request.url), { |
| 184 | + method: "POST", |
| 185 | + headers: { "Content-Type": "application/json", Cookie: cookieHeader }, |
| 186 | + body: JSON.stringify({ |
| 187 | + sessionId, |
| 188 | + requestTokens: requestTokenEstimate, |
| 189 | + responseTokens, |
| 190 | + toolsCalled, |
| 191 | + wasBlocked: false, |
| 192 | + }), |
| 193 | + }).catch(() => { |
| 194 | + // Audit logging is best-effort — swallow network errors. |
| 195 | + }) |
| 196 | + |
| 197 | + enqueue({ type: "done" }) |
| 198 | + break |
| 199 | + } else if (chunk.type === "error") { |
| 200 | + enqueue({ type: "error", error: chunk.error }) |
| 201 | + break |
| 202 | + } |
| 203 | + } |
| 204 | + } catch (err) { |
| 205 | + enqueue({ type: "error", error: err instanceof Error ? err.message : "Unknown error" }) |
| 206 | + } finally { |
| 207 | + closed = true |
| 208 | + try { |
| 209 | + controller.close() |
| 210 | + } catch { |
| 211 | + // already closed |
| 212 | + } |
| 213 | + } |
| 214 | + }, |
| 215 | + }) |
| 216 | + |
| 217 | + return new Response(stream, { |
| 218 | + headers: { |
| 219 | + "Content-Type": "text/event-stream", |
| 220 | + "Cache-Control": "no-cache, no-transform", |
| 221 | + Connection: "keep-alive", |
| 222 | + "X-Accel-Buffering": "no", |
| 223 | + }, |
| 224 | + }) |
| 225 | +} |
0 commit comments