use queue for fetching ranges (#2388)

* fix: use estimated time instead of remaining time

* fix: add sse related header

* fix: handle errors & increase timeout, decrease body size

* fix: use env var for body fetch && remove type any

* use queue with concurrent connections

* add: try/catch block

* add: code refactoring + bug fixes

* fix: check the number of  connections allowed

* fix: ignore email coming from the user doing the mining

* add: remove `contact@` from role and add  `webmaster@` `questions@`

* fix: reduce log frequency by logging signatures pushed to extraction

* fix: update mining source on every signin

* fix: restore old logic

* add: toggle to enable/disable signature extraction

* fix: log error do not throw

* fix: update tagging unit test

* fix: clean, minor refactoring

* fix: update param name

* add: static method to get single connection

* add event listener for pool errors

* remove console.log

* fix: deepsource issues

* fix: formatting

Author: malek10xdev

backend/package-lock.json

@@ -56,8 +56,9 @@
     "mailparser": "^3.7.4",
     "nanoid": "^3.3.11",
     "nodemon": "^3.1.10",
-    "pg": "^8.16.3",
     "p-limit": "^7.1.0",
+    "p-queue": "^8.1.0",
+    "pg": "^8.16.3",
     "pg-format": "^1.0.4",
     "planer": "^1.2.0",
     "qs": "^6.14.0",

backend/package.json

@@ -200,18 +200,21 @@ export default function initializeMiningController(
       const user = res.locals.user as User;
 
       const {
+        extractSignatures,
         miningSource: { email },
         boxes: folders
       }: {
         miningSource: {
           email: string;
         };
         boxes: string[];
+        extractSignatures: boolean;
       } = req.body;
 
       const errors = [
         validateType('email', email, 'string'),
-        validateType('boxes', folders, 'string[]')
+        validateType('boxes', folders, 'string[]'),
+        validateType('extractSignatures', extractSignatures, 'boolean')
       ].filter(Boolean);
 
       if (errors.length) {
@@ -232,7 +235,7 @@ export default function initializeMiningController(
         );
 
       if (!miningSourceCredentials) {
-        return res.status(400).json({
+        return res.status(401).json({
           message: "This mining source isn't registered for this user"
         });
       }
@@ -263,7 +266,7 @@ export default function initializeMiningController(
           userId: user.id,
           email: miningSourceCredentials.email,
           batchSize: ENV.FETCHING_BATCH_SIZE_TO_SEND,
-          fetchEmailBody: ENV.IMAP_FETCH_BODY
+          fetchEmailBody: extractSignatures && ENV.IMAP_FETCH_BODY
         };
         miningTask = await tasksManager.createTask(imapEmailsFetcherOptions);
       } catch (err) {

backend/src/controllers/mining.controller.ts

@@ -1,9 +1,15 @@
 import { Pool } from 'pg';
 import ENV from '../../config';
+import logger from '../../utils/logger';
 
 const pool = new Pool({
   connectionString: ENV.PG_CONNECTION_STRING,
   max: 10
 });
 
+pool.on('error', (error, client) => {
+  logger.error('Error raised by pg-pool: ', error);
+  client.release();
+});
+
 export default pool;

backend/src/db/pg/index.ts

@@ -33,6 +33,60 @@ class ImapConnectionProvider {
     this.poolIsInitialized = false;
   }
 
+  /**
+   * Creates a single IMAP connection to test credentials.
+   * @param email - Email to connect with
+   * @param options - Optional: host, password, port, tls, or OAuth token
+   * @returns Promise<Connection> - A connected ImapFlow instance
+   */
+  static async getSingleConnection(
+    email: string,
+    options?: {
+      host?: string;
+      password?: string;
+      tls?: boolean;
+      port?: number;
+      oauthToken?: string;
+    }
+  ): Promise<Connection> {
+    assert(
+      options?.password || options?.oauthToken,
+      'Either password or OAuth token and host must be provided'
+    );
+    const imapConfig: Partial<ImapFlowOptions> = {
+      auth: options?.oauthToken
+        ? { user: email, accessToken: options.oauthToken }
+        : { user: email },
+      logger: false,
+      socketTimeout: 3600000, // Timeout after one hour
+      connectionTimeout: ENV.IMAP_CONNECTION_TIMEOUT,
+      greetingTimeout: ENV.IMAP_AUTH_TIMEOUT,
+      secure: true
+    };
+
+    if (!options?.host || !options?.port) {
+      const { host, port, tls } = await getOAuthImapConfigByEmail(email);
+      imapConfig.host = host;
+      imapConfig.port = port;
+      imapConfig.secure = tls;
+    }
+
+    const connection = new Connection(imapConfig as ImapFlowOptions);
+
+    // Optional logging
+    connection.on('error', (err) => {
+      logger.error('ImapFlow connection error:', err);
+    });
+
+    try {
+      await connection.connect();
+      return connection;
+    } catch (err) {
+      await connection.logout();
+      throw err;
+    }
+  }
+
   /**
    * Builds the configuration for connecting to Google using OAuth.
    * @param accessToken - OAuth access token
@@ -102,7 +156,6 @@ class ImapConnectionProvider {
 
     connection.on('error', (err) => {
       logger.error('ImapFlow connection error:', err);
-      throw err;
     });
 
     try {
@@ -177,8 +230,15 @@ class ImapConnectionProvider {
         }
       },
       destroy: async (connection) => {
-        await connection.logout();
-        logger.debug('[ImapConnectionProvider]: Imap connection destroyed');
+        try {
+          await connection.logout();
+          logger.debug('[ImapConnectionProvider]: Imap connection destroyed');
+        } catch (err) {
+          logger.error(
+            '[ImapConnectionProvider]: Error destroying connection',
+            err
+          );
+        }
       }
     };