Refactored imapEmailFetcher to run as an independent process (#2458)

Author: malek10xdev

backend/package-lock.json

@@ -47,7 +47,7 @@
     "google-auth-library": "^9.15.1",
     "html-entities": "^2.6.0",
     "imap": "^0.8.19",
-    "imapflow": "^1.0.195",
+    "imapflow": "^1.0.196",
     "ioredis": "^5.7.0",
     "is-url-http": "^2.3.10",
     "jsonwebtoken": "^9.0.2",
@@ -90,7 +90,7 @@
     "@types/ioredis-mock": "^8.2.5",
     "@types/jsonwebtoken": "^9.0.7",
     "@types/mailparser": "^3.4.6",
-    "@types/node": "^22.10.5",
+    "@types/node": "^22.18.8",
     "@types/pg": "^8.11.10",
     "@types/pg-format": "^1.0.5",
     "@types/quoted-printable": "^1.0.2",

backend/package.json

@@ -13,6 +13,9 @@ const schema = z.object({
   FRONTEND_HOST: z.string().url(),
 
   /* FETCHING */
+  EMAIL_FETCHING_SERVICE_URL: z.string().min(1),
+  EMAIL_FETCHING_SERVICE_API_TOKEN: z.string().min(8),
+  // TODO: CONSISTENT PREFIX FOR FETCHING RELATED VARS
   FETCHING_BATCH_SIZE_TO_SEND: number(),
   FETCHING_CHUNK_SIZE_PER_CONNECTION: number(),
   FETCHING_MAX_CONNECTIONS_PER_FOLDER: number(),

backend/src/config/schema.ts

@@ -1,8 +1,6 @@
 import { User } from '@supabase/supabase-js';
 import { NextFunction, Request, Response } from 'express';
-import { ImapFlow as Connection } from 'imapflow';
 import {
-  ImapMiningSourceCredentials,
   MiningSources,
   OAuthMiningSourceCredentials,
   OAuthMiningSourceProvider
@@ -23,23 +21,6 @@ type NewToken = {
   expires_at: number;
 };
 
-async function getImapConnectionProvider(
-  data: OAuthMiningSourceCredentials | ImapMiningSourceCredentials
-) {
-  if ('accessToken' in data) {
-    const connection = await new ImapConnectionProvider(data.email).withOAuth(
-      data
-    );
-    return connection;
-  }
-  return new ImapConnectionProvider(data.email).withPassword(
-    data.host,
-    data.password,
-    data.tls,
-    data.port
-  );
-}
-
 function getTokenAndProvider(data: OAuthMiningSourceCredentials) {
   const { provider, accessToken, refreshToken, expiresAt } = data;
   const client = provider === 'azure' ? azureOAuth2Client : googleOAuth2Client;
@@ -76,9 +57,6 @@ async function upsertMiningSource(
 export default function initializeImapController(miningSources: MiningSources) {
   return {
     async getImapBoxes(req: Request, res: Response, next: NextFunction) {
-      let imapConnectionProvider: ImapConnectionProvider | null = null;
-      let imapConnection: Connection | null = null;
-
       const { email } = req.body;
 
       const errors = [validateType('email', email, 'string')].filter(Boolean);
@@ -124,12 +102,25 @@ export default function initializeImapController(miningSources: MiningSources) {
           }
         }
 
-        imapConnectionProvider = await getImapConnectionProvider(data);
-        imapConnection = await imapConnectionProvider.acquireConnection();
+        const imapConnection = await ImapConnectionProvider.getSingleConnection(
+          email,
+          'accessToken' in data
+            ? {
+                oauthToken: data.accessToken
+              }
+            : {
+                host: data.host,
+                password: data.password,
+                tls: data.tls,
+                port: data.port
+              }
+        );
 
-        const imapBoxesFetcher = new ImapBoxesFetcher(imapConnectionProvider);
+        const imapBoxesFetcher = new ImapBoxesFetcher(imapConnection, logger);
         const tree: any = await imapBoxesFetcher.getTree(data.email);
 
+        await imapConnection.logout();
+
         logger.info('Mining IMAP tree succeeded.', {
           metadata: {
             user: hashEmail(data.email, userId)
@@ -140,6 +131,10 @@ export default function initializeImapController(miningSources: MiningSources) {
           data: { message: 'IMAP folders fetched successfully!', folders: tree }
         });
       } catch (error: any) {
+        logger.error(`Error during inbox fetch: ${(error as Error).message}`, {
+          error
+        });
+
         if ([502, 503].includes(error?.output?.payload?.statusCode)) {
           return res
             .status(error?.output?.payload?.statusCode)
@@ -151,11 +146,6 @@ export default function initializeImapController(miningSources: MiningSources) {
           return res.status(generatedError.status).send(generatedError);
         }
         return next(generatedError);
-      } finally {
-        if (imapConnection) {
-          await imapConnectionProvider?.releaseConnection(imapConnection);
-        }
-        await imapConnectionProvider?.cleanPool();
       }
     }
   };

backend/src/controllers/imap.controller.ts

@@ -7,8 +7,6 @@ import {
   OAuthMiningSourceProvider
 } from '../db/interfaces/MiningSources';
 import { ContactFormat } from '../services/extractors/engines/FileImport';
-import ImapConnectionProvider from '../services/imap/ImapConnectionProvider';
-import { ImapEmailsFetcherOptions } from '../services/imap/types';
 import TaskManagerFile from '../services/tasks-manager/TaskManagerFile';
 import TasksManager from '../services/tasks-manager/TasksManager';
 import { ImapAuthError } from '../utils/errors';
@@ -241,44 +239,15 @@ export default function initializeMiningController(
         });
       }
 
-      const imapConnectionProvider =
-        'accessToken' in miningSourceCredentials
-          ? await new ImapConnectionProvider(
-              miningSourceCredentials.email
-            ).withOAuth(miningSourceCredentials, {
-              miningSources,
-              userId: user.id
-            })
-          : new ImapConnectionProvider(
-              miningSourceCredentials.email
-            ).withPassword(
-              miningSourceCredentials.host,
-              miningSourceCredentials.password,
-              miningSourceCredentials.tls,
-              miningSourceCredentials.port
-            );
-
-      let imapConnection = null;
-      let miningTask = null;
-
       try {
-        // Connect to validate connection before creating the pool.
-        imapConnection = await imapConnectionProvider.acquireConnection();
-        const imapEmailsFetcherOptions: ImapEmailsFetcherOptions = {
-          imapConnectionProvider,
+        const miningTask = await tasksManager.createTask({
           boxes: sanitizedFolders,
           userId: user.id,
           email: miningSourceCredentials.email,
-          batchSize: ENV.FETCHING_BATCH_SIZE_TO_SEND,
           fetchEmailBody: extractSignatures && ENV.IMAP_FETCH_BODY
-        };
-        miningTask = await tasksManager.createTask(imapEmailsFetcherOptions);
+        });
+        return res.status(201).send({ error: null, data: miningTask });
       } catch (err) {
-        if (imapConnection) {
-          await imapConnectionProvider.releaseConnection(imapConnection);
-        }
-        await imapConnectionProvider.cleanPool();
-
         if (
           err instanceof Error &&
           err.message.toLowerCase().startsWith('invalid credentials')
@@ -298,8 +267,6 @@ export default function initializeMiningController(
         res.status(500);
         return next(new Error(newError.message));
       }
-
-      return res.status(201).send({ error: null, data: miningTask });
     },
 
     async startMiningFile(req: Request, res: Response, next: NextFunction) {

backend/src/controllers/mining.controller.ts

@@ -7,7 +7,6 @@ import PgContacts from './db/pg/PgContacts';
 import PgMiningSources from './db/pg/PgMiningSources';
 import SupabaseUsers from './db/supabase/users';
 import SupabaseAuthResolver from './services/auth/SupabaseAuthResolver';
-import EmailFetcherFactory from './services/factory/EmailFetcherFactory';
 import SSEBroadcasterFactory from './services/factory/SSEBroadcasterFactory';
 import TasksManager from './services/tasks-manager/TasksManager';
 import { flickrBase58IdGenerator } from './services/tasks-manager/utils';
@@ -16,6 +15,8 @@ import redis from './utils/redis';
 import supabaseClient from './utils/supabase';
 import SupabaseTasks from './db/supabase/tasks';
 import TasksManagerFile from './services/tasks-manager/TaskManagerFile';
+import EmailFetcherClient from './services/email-fetching';
+import EmailFetcherFactory from './services/factory/EmailFetcherFactory';
 
 // eslint-disable-next-line no-console
 console.log(
@@ -47,7 +48,11 @@ console.log(
     tasksResolver,
     redis.getSubscriberClient(),
     redis.getClient(),
-    new EmailFetcherFactory(),
+    new EmailFetcherClient(
+      logger,
+      ENV.EMAIL_FETCHING_SERVICE_API_TOKEN,
+      ENV.EMAIL_FETCHING_SERVICE_URL
+    ),
     new SSEBroadcasterFactory(),
     flickrBase58IdGenerator()
   );

backend/src/server.ts

@@ -0,0 +1,66 @@
+// src/clients/EmailFetcherClient.ts
+import { Logger } from 'winston';
+import axios, { AxiosInstance } from 'axios';
+
+export interface FetchStartPayload {
+  userId: string;
+  miningId: string;
+  email: string;
+  boxes: string[];
+  extractSignatures: boolean;
+  contactStream: string;
+  signatureStream: string;
+}
+
+export interface FetchStopPayload {
+  miningId: string;
+  canceled: boolean;
+}
+
+class EmailFetcherClient {
+  private client: AxiosInstance;
+
+  constructor(
+    private readonly logger: Logger,
+    apiToken: string,
+    baseUrl: string
+  ) {
+    this.client = axios.create({
+      baseURL: baseUrl,
+      headers: { 'Content-Type': 'application/json', 'x-api-token': apiToken }
+    });
+  }
+
+  /**
+   * Start IMAP fetch job
+   */
+  async startFetch(payload: FetchStartPayload) {
+    try {
+      const { data } = await this.client.post('api/imap/fetch/start', payload);
+      return data;
+    } catch (error) {
+      this.logger.error('Start fetching request failed', { error, payload });
+      throw error;
+    }
+  }
+
+  /**
+   * Stop IMAP fetch job
+   */
+  async stopFetch(payload: FetchStopPayload) {
+    try {
+      const { data } = await this.client.delete('api/imap/fetch/stop', {
+        data: payload
+      });
+      return data;
+    } catch (error) {
+      this.logger.error('Stop fetching request with failed', {
+        error,
+        payload
+      });
+      throw error;
+    }
+  }
+}
+
+export default EmailFetcherClient;