add: fetch using concurrent connection & better sse management (#2368)

Author: malek10xdev

.env.master.dev

@@ -21,10 +21,13 @@ FRONTEND_HOST = http://localhost:8082   # ( REQUIRED )
 LEADMINER_API_PORT = 8081   # ( REQUIRED )
 LEADMINER_API_HOST = http://localhost:8081 # ( REQUIRED )
 
-LEADMINER_FETCH_BATCH_SIZE = 100          # ( REQUIRED ) Sends notification every x processed items to the frontend
 LEADMINER_API_HASH_SECRET = change_me     # ( REQUIRED ) Used for hashing secrets
 LEADMINER_MINING_ID_GENERATOR_LENGTH = 10 # ( REQUIRED ) Length of the task id
 
+FETCHING_BATCH_SIZE_TO_SEND = 100          # ( REQUIRED ) Sends notification every x processed items to the frontend
+FETCHING_CHUNK_SIZE_PER_CONNECTION = 10000
+FETCHING_MAX_CONNECTIONS_PER_FOLDER = 5
+
 ## LOGGING ##
 # GRAFANA_LOKI_HOST =             # Use Loki transport for logging (Default is console)
 LEADMINER_API_LOG_LEVEL = debug   # Logging level (debug, info, notice, warning...)

.env.master.prod

@@ -19,10 +19,13 @@ FRONTEND_HOST =             # ( REQUIRED )
 LEADMINER_API_PORT = 8081   # ( REQUIRED )
 LEADMINER_API_HOST =        # ( REQUIRED )
 
-LEADMINER_FETCH_BATCH_SIZE = 100          # ( REQUIRED ) Sends notification every x processed items to the frontend
 LEADMINER_API_HASH_SECRET = change_me     # ( REQUIRED ) Used for hashing secrets
 LEADMINER_MINING_ID_GENERATOR_LENGTH = 10 # ( REQUIRED ) Length of the task id
 
+FETCHING_BATCH_SIZE_TO_SEND = 100          # ( REQUIRED ) Sends notification every x processed items to the frontend
+FETCHING_CHUNK_SIZE_PER_CONNECTION = 10000
+FETCHING_MAX_CONNECTIONS_PER_FOLDER = 5
+
 ## LOGGING ##
 GRAFANA_LOKI_HOST =             # Use Loki transport for logging (Default is console)
 LEADMINER_API_LOG_LEVEL = debug # Logging level (debug, info, notice, warning...)

backend/package-lock.json

@@ -56,6 +56,7 @@
     "mailparser": "^3.7.4",
     "nanoid": "^3.3.11",
     "nodemon": "^3.1.10",
+    "p-limit": "^7.1.0",
     "pg": "^8.16.1",
     "pg-format": "^1.0.4",
     "planer": "^1.2.0",

backend/package.json

@@ -10,9 +10,13 @@ const schema = z.object({
   LEADMINER_API_HOST: z.string().url(),
   LEADMINER_API_HASH_SECRET: z.string().min(1),
   LEADMINER_MINING_ID_GENERATOR_LENGTH: number(),
-  LEADMINER_FETCH_BATCH_SIZE: number(),
   FRONTEND_HOST: z.string().url(),
 
+  /* FETCHING */
+  FETCHING_BATCH_SIZE_TO_SEND: number(),
+  FETCHING_CHUNK_SIZE_PER_CONNECTION: number(),
+  FETCHING_MAX_CONNECTIONS_PER_FOLDER: number(),
+
   /* IMAP */
   IMAP_AUTH_TIMEOUT: number(),
   IMAP_CONNECTION_TIMEOUT: number(),

backend/src/config/schema.ts

@@ -135,7 +135,7 @@ export default function initializeMiningController(
 
       const sanitizedHost = sanitizeImapInput(host);
       const sanitizedEmail = sanitizeImapInput(email);
-      const sanitizedPassword = sanitizeImapInput(password);
+      const sanitizedPassword = password;
 
       try {
         // Validate & Get the valid IMAP login connection before creating the pool.
@@ -262,7 +262,7 @@ export default function initializeMiningController(
           boxes: sanitizedFolders,
           userId: user.id,
           email: miningSourceCredentials.email,
-          batchSize: ENV.LEADMINER_FETCH_BATCH_SIZE,
+          batchSize: ENV.FETCHING_BATCH_SIZE_TO_SEND,
           fetchEmailBody: ENV.IMAP_FETCH_BODY
         };
         miningTask = await tasksManager.createTask(imapEmailsFetcherOptions);

backend/src/controllers/mining.controller.ts

@@ -25,26 +25,20 @@ export default function initializeStreamController(
         logger.debug(`Attaching sse connection for taskId ${taskId}`);
         manager.attachSSE(taskId, { req, res });
       } catch (error) {
+        res.setHeader('Content-Type', 'text/event-stream');
+        res.setHeader('Cache-Control', 'no-cache');
+        res.setHeader('Connection', 'keep-alive');
+
         res.status(404);
-        res.write('id: 0\n');
+        res.write('id: 404-not-found\n');
         res.write('event: close\n');
         res.write(`data: ${JSON.stringify((error as Error).message)}\n\n`);
         res.flushHeaders();
         res.end();
       }
 
       req.on('close', async () => {
-        try {
-          logger.debug(
-            `Closing sse connection and deleting task for taskId ${taskId}`
-          );
-          await manager.deleteTask(taskId, null);
-        } catch (error) {
-          logger.error(
-            `Error when disconnecting from the stream with miningId ${taskId}`,
-            error
-          );
-        }
+        logger.warn(`SSE Connection lost for mining task with id: ${taskId}`);
       });
     }
   };