chunk data and queue job worker

Author: kyawzinthant-coding

README.MD

@@ -1,5 +1,5 @@
 - [x] Upload Multiple Files
-- [ ] implement queue worker
+- [x] implement queue worker
 - [x] read pdf and chunk the text
 - [ ] combine those knowledge
 - [ ] store in vector db

package.json

@@ -6,6 +6,7 @@
   "scripts": {
     "dev": "nodemon src/index.ts",
     "build": "tsc --project tsconfig.prod.json",
+    "work": "nodemon src/jobs/workers/filesWorker.ts",
     "start": "node dist/src/index.js"
   },
   "keywords": [],

src/controllers/pdf/pdf-upload.ts

@@ -1,7 +1,11 @@
-import { Request, Response, NextFunction } from "express";
+import { Request, Response } from "express";
 import { processPdf } from "../../service/process-document";
+import fileQueue from "../../jobs/queues/filesQueue";
 
-export const pdfUploadFilesController = (req: Request, res: Response): any => {
+export const pdfUploadFilesController = async (
+  req: Request,
+  res: Response
+): Promise<any> => {
   if (!req.files || (req.files as Express.Multer.File[]).length === 0) {
     return res.status(400).json("No Files were upload");
   }
@@ -13,7 +17,18 @@ export const pdfUploadFilesController = (req: Request, res: Response): any => {
     path: file.path,
   }));
 
-  processPdf(fileInfo[0].path);
+  try {
+    for (const file of files) {
+      console.log(`Adding job to queue for file: ${file.originalname}`);
+
+      await fileQueue.add("process-pdf", {
+        filePath: file.path,
+        originalName: file.filename,
+      });
+    }
+  } catch (error) {
+    console.log("Queue Job Worker Error", error.message);
+  }
 
   res.status(200).json({
     message: `${files.length} files uploaded successfully!`,

src/jobs/queues/filesQueue.ts

@@ -3,7 +3,7 @@ import Redis from "ioredis";
 
 const connection = new Redis({
   host: process.env.REDIS_HOST,
-  port: parseInt(process.env.REDIS_PORT),
+  port: parseInt(process.env.REDIS_PORT) | 6379,
 });
 
 const fileQueue = new Queue("FileQueue", { connection });

src/jobs/workers/filesWorker.ts

@@ -1,10 +1,13 @@
+import "dotenv/config";
 import { Job, Worker } from "bullmq";
 import Redis from "ioredis";
 import { processPdf } from "../../service/process-document";
 
+console.log(process.env.REDIS_PORT);
+
 const connection = new Redis({
   host: process.env.REDIS_HOST,
-  port: parseInt(process.env.REDIS_PORT),
+  port: parseInt(process.env.REDIS_PORT) || 6379,
   maxRetriesPerRequest: null,
 });
 
@@ -15,9 +18,15 @@ const fileWorker = new Worker(
       `[Worker] Picked up job #${job.id} for file: ${job.data.originalName}`
     );
 
-    await processPdf(job.data.filePath);
+    const chunk = await processPdf(job.data.filePath);
+
+    console.log(` ✅ Chunk Data from ${job.data.originalName}  ${chunk}`);
+
+    console.log(` ✅ [Worker] Finished job #${job.id}`);
 
-    console.log(`[Worker] Finished job #${job.id}`);
+    console.log(
+      "-------------------------------------------------------------------"
+    );
   },
   { connection }
 );

src/service/process-document.ts

@@ -12,8 +12,8 @@ export async function processPdf(filePath: string) {
     const extractedText = pdfData.text;
 
     console.log(`✅ Text extracted successfully.`);
-    console.log(`   - Total Pages: ${pdfData.numpages}`);
-    console.log(`   - Text Length: ${extractedText.length} characters\n`);
+    // console.log(`   - Total Pages: ${pdfData.numpages}`);
+    // console.log(`   - Text Length: ${extractedText.length} characters\n`);
 
     const splitter = new RecursiveCharacterTextSplitter({
       chunkSize: 1000, // Max characters per chunk
@@ -22,18 +22,20 @@ export async function processPdf(filePath: string) {
 
     const chunks = await splitter.splitText(extractedText);
 
-    console.log(`✅ Text chunked successfully.`);
+    console.log(`✅ Text chunked successfully for ${filePath}.`);
     console.log(`   - Number of chunks created: ${chunks.length}\n`);
 
-    for (let i = 0; i < Math.min(3, chunks.length); i++) {
-      console.log(`\n--- Chunk ${i + 1} (Length: ${chunks[i].length}) ---`);
-      console.log(chunks[i]);
-      console.log("-------------------------------------\n");
-    }
+    // for (let i = 0; i < Math.min(1, chunks.length); i++) {
+    //   console.log(`\n--- Chunk ${i + 1} (Length: ${chunks[i].length}) ---`);
+    //   console.log(chunks[i]);
+    //   console.log("-------------------------------------\n");
+    // }
 
-    if (chunks.length > 3) {
-      console.log(`... and ${chunks.length - 3} more chunks.`);
-    }
+    // if (chunks.length > 3) {
+    //   console.log(`... and ${chunks.length - 3} more chunks.`);
+    // }
+
+    return chunks;
   } catch (error) {
     console.error("An error occurred during PDF processing:", error);
   }