WIP: RAG based chats

Author: sunjay-dev

backend/package.json

@@ -26,10 +26,13 @@
     "@clerk/backend": "^2.33.0",
     "@hono/clerk-auth": "^3.1.0",
     "@langchain/core": "^1.1.30",
+    "@langchain/google-genai": "^2.1.24",
     "@langchain/groq": "^1.1.4",
     "@langchain/langgraph-checkpoint-redis": "^1.0.2",
     "@langchain/openrouter": "^0.1.5",
+    "@langchain/pinecone": "^1.0.1",
     "@langchain/tavily": "^1.2.0",
+    "@pinecone-database/pinecone": "^7.1.0",
     "@prisma/adapter-pg": "^7.4.2",
     "@prisma/client": "^7.4.2",
     "bullmq": "^5.70.2",

backend/prisma/migrations/20260307053831_init/migration.sql

@@ -0,0 +1,6 @@
+-- CreateEnum
+CREATE TYPE "RagStatus" AS ENUM ('IDLE', 'PROCESSING', 'COMPLETED', 'FAILED');
+
+-- AlterTable
+ALTER TABLE "Chat" ADD COLUMN     "isRag" BOOLEAN NOT NULL DEFAULT false,
+ADD COLUMN     "ragStatus" "RagStatus" NOT NULL DEFAULT 'IDLE';

backend/prisma/schema.prisma

@@ -48,10 +48,19 @@ model UserPreference {
   user               User     @relation(fields: [userId], references: [clerkId], onDelete: Cascade)
 }
 
+enum RagStatus {
+  IDLE
+  PROCESSING
+  COMPLETED
+  FAILED
+}
+
 model Chat {
   id        String    @id @default(uuid())
   userId    String
   title     String?
+  isRag     Boolean   @default(false)
+  ragStatus RagStatus @default(IDLE)
   createdAt DateTime  @default(now())
   updatedAt DateTime  @updatedAt
   user      User      @relation(fields: [userId], references: [clerkId])

backend/src/app.ts

@@ -15,7 +15,7 @@ app.use(
 import serverRoutes from "./routes/server.routes.js";
 import chatRouter from "./routes/chat.routes.js";
 import messageRouter from "./routes/message.routes.js";
-import clerkRouter from "./routes/clerk.routes.js";
+import webHookRouter from "./routes/webHook.routes.js";
 import userRouter from "./routes/user.routes.js";
 import searchRouter from "./routes/search.routes.js";
 import uploadRouter from "./routes/upload.routes.js";
@@ -27,34 +27,10 @@ app.route("/api/chat/:chatId/message", messageRouter);
 app.route("/api/user", userRouter);
 app.route("/api/upload", uploadRouter);
 app.route("/api/search", searchRouter);
-app.route("/api/clerk", clerkRouter);
+app.route("/api/webhooks", webHookRouter);
 
 app.onError(errorHandler);
 
-// import { Queue } from "bullmq";
-
-// const fileIngestQueue = new Queue("file-ingest", {
-//   connection: {
-//     url: process.env.REDIS_URL as string,
-//   },
-//   defaultJobOptions: {
-//     attempts: 3,
-//     backoff: {
-//       type: "exponential",
-//       delay: 3000,
-//     },
-//     removeOnComplete: true,
-//     removeOnFail: false,
-//   },
-// });
-
-// fileIngestQueue.add("ingest", {
-//   fileUrl: "https://res.cloudinary.com/dzxcifimr/raw/upload/v1768761888/chatty-ai/qytcyksxdqadx2xa4nen.txt",
-//   userId: "user_38KrOBnsnDtO35wpRPwMvdfPhQU",
-//   chatId: "1e4efe47-424f-4fa0-9824-57a518ad089d",
-//   fileId: "57ae0bf3-8ad8-4c14-abd0-f1eec0c60eed",
-// });
-
 export default {
   port: process.env.PORT as string,
   idleTimeout: 30,

backend/src/config/queue.config.ts

@@ -0,0 +1,10 @@
+import { Queue } from "bullmq";
+
+export const fileIngestQueue = new Queue("file-ingest", {
+  connection: {
+    url: process.env.REDIS_URL as string,
+  },
+  defaultJobOptions: {
+    attempts: 2,
+  },
+});

backend/src/config/vectorStore.config.ts

@@ -0,0 +1,18 @@
+import { GoogleGenerativeAIEmbeddings } from "@langchain/google-genai";
+import { PineconeStore } from "@langchain/pinecone";
+import { Pinecone } from "@pinecone-database/pinecone";
+
+export const embeddings = new GoogleGenerativeAIEmbeddings({
+  model: process.env.GOOGLE_EMBEDDINGS_MODEL as string,
+  apiKey: process.env.GOOGLE_API_KEY as string,
+});
+
+export const pinecone = new Pinecone({
+  apiKey: process.env.PINECONE_API_KEY as string,
+});
+
+export const index = pinecone.Index({ name: process.env.PINECONE_INDEX_NAME as string });
+
+export const vectorStore = await PineconeStore.fromExistingIndex(embeddings, {
+  pineconeIndex: index,
+});

backend/src/controllers/message.controllers.ts

@@ -5,35 +5,8 @@ import { NotFoundError } from "../utils/appError.utils.js";
 import { streamText as honoStreamText } from "hono/streaming";
 import logger from "../utils/logger.utils.js";
 import { streamLoading, streamText, streamError } from "../utils/stream.utils.js";
-import { type UploadedFile } from "@app/shared/src/schemas/message.schema.js"
-
-async function handleImageExtraction(messageFiles: UploadedFile[], stream: any): Promise<string> {
-  if (messageFiles?.length === 0) return "";
-
-  try {
-    await streamLoading(stream, "Analyzing image...");
-    const extractionPromises = messageFiles.map((file: UploadedFile) => {
-      if (file.fileUrl) {
-        return extractImageData(file.fileUrl);
-      }
-      return Promise.resolve(null);
-    });
-
-    const extractionResults = await Promise.allSettled(extractionPromises);
-
-    const successfulData = extractionResults
-      .filter((result) => result.status === "fulfilled" && result.value)
-      .map((result: any) => result.value as string);
-
-    if (successfulData.length > 0) {
-      return `\n\nExtracted information from attached images:\n${successfulData.join("\n---\n")}\n\nUse this information if relevant when answering.`;
-    }
-  } catch (error) {
-    logger.error({ message: "Failed to process attached images", error });
-  }
-
-  return "";
-}
+import { type UploadedFile } from "@app/shared/src/schemas/message.schema.js";
+import { vectorStore } from "../config/vectorStore.config.js";
 
 export async function handleUserMessageResponse(c: Context) {
   const requestStartTime = new Date();
@@ -50,7 +23,7 @@ export async function handleUserMessageResponse(c: Context) {
 
     try {
       const [chat, preferences, memories] = await Promise.all([
-        prisma.chat.findUnique({ where: { id: chatId, userId }, select: { id: true } }),
+        prisma.chat.findUnique({ where: { id: chatId, userId }, select: { id: true, isRag: true } }),
         prisma.userPreference.upsert({ where: { userId }, create: { userId }, update: {} }),
         prisma.userMemory.findMany({ where: { userId }, select: { content: true } }),
       ]);
@@ -60,6 +33,23 @@ export async function handleUserMessageResponse(c: Context) {
         return;
       }
 
+      if (chat.isRag && query.trim() !== "") {
+        try {
+          await streamLoading(stream, "Searching documents...");
+
+          const results = await vectorStore.similaritySearch(query, 4, {
+            chatId: chat.id,
+          });
+
+          if (results.length > 0) {
+            const context = results.map((r) => r.pageContent).join("\n\n---\n\n");
+            finalQuery = `Context Information:\n${context}\n\nUser Query: ${finalQuery}\n\nPlease answer the user query using the context information provided above if it is relevant.`;
+          }
+        } catch (err) {
+          logger.error({ message: "Vector search failed", error: err, chatId });
+        }
+      }
+
       const aiStream = generateAIResponse({
         query: finalQuery,
         threadId: chatId,
@@ -85,7 +75,7 @@ export async function handleUserMessageResponse(c: Context) {
 
       const filesToCreate = messageFiles?.length ? messageFiles : undefined;
 
-      await prisma.chat.update({
+      const createdChat = await prisma.chat.update({
         where: { id: chatId },
         data: {
           updatedAt: requestStartTime,
@@ -101,6 +91,7 @@ export async function handleUserMessageResponse(c: Context) {
             ],
           },
         },
+        include: { messages: { include: { messageFiles: true } } },
       });
 
       scheduleMemoryExtraction(userId, query, memories);
@@ -115,6 +106,34 @@ export async function handleUserMessageResponse(c: Context) {
   });
 }
 
+async function handleImageExtraction(messageFiles: UploadedFile[], stream: any) {
+  if (!messageFiles?.length) return "";
+
+  const imageFiles = messageFiles.filter((file) => file.fileType?.includes("image") && file.fileUrl);
+
+  if (imageFiles.length === 0) return "";
+
+  try {
+    await streamLoading(stream, "Analyzing image...");
+
+    const extractionPromises = imageFiles.map((file) => extractImageData(file.fileUrl));
+
+    const extractionResults = await Promise.allSettled(extractionPromises);
+
+    const successfulData = extractionResults
+      .filter((result) => result.status === "fulfilled" && result.value)
+      .map((result: any) => result.value as string);
+
+    if (successfulData.length > 0) {
+      return `\n\nExtracted information from attached images:\n${successfulData.join("\n---\n")}\n\nUse this information if relevant when answering.`;
+    }
+  } catch (error) {
+    logger.error({ message: "Failed to process attached images", error });
+  }
+
+  return "";
+}
+
 export async function handleGetAllChatMessages(c: Context) {
   const userId = c.get("user");
   const { chatId } = c.get("param");
@@ -124,6 +143,8 @@ export async function handleGetAllChatMessages(c: Context) {
     select: {
       id: true,
       title: true,
+      ragStatus: true,
+      isRag: true,
       messages: {
         orderBy: { createdAt: "asc" },
         select: {

backend/src/controllers/upload.controllers.ts

@@ -1,5 +1,7 @@
 import { type Context } from "hono";
 import cloudinary from "../config/cloudinary.config.js";
+import prisma from "../config/prisma.config.js";
+import { fileIngestQueue } from "../config/queue.config.js";
 
 export async function handleFileSignature(c: Context) {
   const timestamp = Math.floor(Date.now() / 1000);
@@ -17,3 +19,54 @@ export async function handleFileSignature(c: Context) {
     200,
   );
 }
+
+export async function handleStartRag(c: Context) {
+  const userId = c.get("user");
+  const body = await c.req.json();
+  const { fileUrl, fileName, fileType, chatId } = body;
+
+  if (!fileUrl) {
+    return c.json({ error: "Missing fileUrl" }, 400);
+  }
+
+  if (fileType && fileType.includes("image")) {
+    return c.json({ error: "Image files are not supported for RAG parsing" }, 400);
+  }
+
+  let targetChatId = chatId;
+
+  if (!targetChatId) {
+    let title = "Document Chat";
+    if (fileName) {
+      const lastDot = fileName.lastIndexOf(".");
+      title = lastDot !== -1 ? fileName.substring(0, lastDot) : fileName;
+    }
+
+    const newChat = await prisma.chat.create({
+      data: {
+        userId,
+        title,
+        ragStatus: "PROCESSING",
+      }
+    });
+    targetChatId = newChat.id;
+  } else {
+    await prisma.chat.update({
+      where: { id: targetChatId, userId },
+      data: {
+        ragStatus: "PROCESSING"
+      }
+    });
+  }
+
+  const tempFileId = crypto.randomUUID();
+
+  await fileIngestQueue.add("ingest", {
+    fileUrl,
+    userId,
+    chatId: targetChatId,
+    fileId: tempFileId,
+  });
+
+  return c.json({ success: true, chatId: targetChatId }, 200);
+}

…end/src/controllers/clerk.controllers.ts …d/src/controllers/webHook.controllers.tsbackend/src/controllers/clerk.controllers.ts renamed to backend/src/controllers/webHook.controllers.ts backend/src/controllers/clerk.controllers.ts renamed to backend/src/controllers/webHook.controllers.ts

@@ -36,3 +36,28 @@ export async function handleClerkWebHook(c: Context) {
   }
   return c.json({ ok: true }, 200);
 }
+
+export async function handleRagWebhook(c: Context) {
+  const secret = c.req.header("x-internal-secret");
+  if (secret !== process.env.INTERNAL_SECRET) {
+    return c.json({ error: "Unauthorized" }, 401);
+  }
+
+  const { chatId, status } = await c.req.json();
+
+  if (!chatId || !status) {
+    return c.json({ error: "Missing required fields" }, 400);
+  }
+
+  const dataToUpdate: any = { ragStatus: status };
+  if (status === "COMPLETED") {
+    dataToUpdate.isRag = true;
+  }
+
+  await prisma.chat.update({
+    where: { id: chatId },
+    data: dataToUpdate,
+  });
+
+  return c.json({ success: true }, 200);
+}