feat: add queue for controlling concurrency & multi worker support for signature streams (#2611)

Author: malek10xdev

backend/src/controllers/imap.controller.ts

@@ -94,7 +94,10 @@ export default function initializeImapController(miningSources: MiningSources) {
             await upsertMiningSource(
               miningSources,
               userId,
-              newToken,
+              {
+                ...newToken,
+                refresh_token: newToken.refresh_token ?? refreshToken
+              },
               provider,
               data.email
             );

backend/src/emailSignatureWorker.ts

@@ -50,7 +50,7 @@ if (ENV.SIGNATURE_OPENROUTER_API_KEY) {
   });
 }
 
-const { processStreamData } = initializeEmailSignatureProcessor(
+const { processStreamData, handler } = initializeEmailSignatureProcessor(
   supabaseClient,
   new Signature(logger, signatureEngines),
   emailSignatureCache,
@@ -78,7 +78,8 @@ const emailsStreamConsumer = new EmailSignatureConsumer(
   ENV.REDIS_EMAIL_SIGNATURE_CONSUMER_BATCH_SIZE,
   processStreamData,
   redisClient,
-  logger
+  logger,
+  handler
 );
 
 (async () => {

backend/src/services/cache/EmailSignatureCache.ts

@@ -12,6 +12,10 @@ export interface EmailSignatureWithMetadata {
   lastSeenDate: string;
 }
 
+export interface SignatureProgress {
+  received: number;
+}
+
 export default interface EmailSignatureCache {
   /**
    * Set or update a signature for an email address and associate it with a mining operation
@@ -47,4 +51,24 @@ export default interface EmailSignatureCache {
    * @param miningId - The ID of the mining operation to clear
    */
   clearCachedSignature(miningId: string): Promise<void>;
+
+  /**
+   * Increments the received message count for a mining operation
+   * @param miningId - The ID of the mining operation
+   * @returns The new received count
+   */
+  incrementReceived(miningId: string): Promise<number>;
+
+  /**
+   * Gets current progress state without modifying it
+   * @param miningId - The ID of the mining operation
+   * @returns Current progress state
+   */
+  getProgress(miningId: string): Promise<SignatureProgress>;
+
+  /**
+   * Clears all progress tracking keys for a mining operation
+   * @param miningId - The ID of the mining operation
+   */
+  clearProgress(miningId: string): Promise<void>;
 }

backend/src/services/cache/redis/RedisEmailSignatureCache.ts

@@ -1,6 +1,7 @@
 import Redis from 'ioredis';
 import EmailSignatureCache, {
-  EmailSignatureWithMetadata
+  EmailSignatureWithMetadata,
+  SignatureProgress
 } from '../EmailSignatureCache';
 
 export default class RedisEmailSignatureCache implements EmailSignatureCache {
@@ -102,4 +103,35 @@ export default class RedisEmailSignatureCache implements EmailSignatureCache {
       await this.client.del(miningKey);
     }
   }
+
+  /**
+   * Increments the received message count for a mining operation
+   * Uses Redis INCR for atomicity
+   */
+  public async incrementReceived(miningId: string): Promise<number> {
+    const key = `mining:${miningId}:received`;
+    const count = await this.client.incr(key);
+    return count;
+  }
+
+  /**
+   * Gets current progress state without modifying it
+   */
+  public async getProgress(miningId: string): Promise<SignatureProgress> {
+    const [received] = await this.client.mget(`mining:${miningId}:received`);
+
+    return {
+      received: parseInt(received || '0', 10)
+    };
+  }
+
+  /**
+   * Clears all progress tracking keys for a mining operation
+   */
+  public async clearProgress(miningId: string): Promise<void> {
+    await this.client.del(
+      `mining:${miningId}:received`,
+      `mining:${miningId}:total`
+    );
+  }
 }

backend/src/services/email-status/constants.ts

@@ -1,2 +1,2 @@
-const TIMEOUT = 10_000; // Default API request timeout
-export default TIMEOUT;
+export const TIMEOUT = 10_000; // Default API request timeout
+export const ZEROBOUNCE_TIMEOUT = 30_000; // Default API request timeout

backend/src/services/email-status/mailercheck/client.ts

@@ -2,7 +2,7 @@ import axios, { AxiosInstance } from 'axios';
 import { Logger } from 'winston';
 import { logError } from '../../../utils/axios';
 import { IRateLimiter } from '../../rate-limiter';
-import TIMEOUT from '../constants';
+import { TIMEOUT } from '../constants';
 
 export default class MailerCheckClient {
   private static readonly baseURL = 'https://app.mailercheck.com/api/';

backend/src/services/email-status/reacher/client.ts

@@ -2,7 +2,7 @@ import axios, { AxiosError, AxiosInstance } from 'axios';
 import { Logger } from 'winston';
 import { logError } from '../../../utils/axios';
 import { IRateLimiter } from '../../rate-limiter';
-import TIMEOUT from '../constants';
+import { TIMEOUT } from '../constants';
 
 interface BulkSubmitResponse {
   job_id: string;

backend/src/services/email-status/zerobounce/client.ts

@@ -3,7 +3,7 @@ import axios, { AxiosInstance } from 'axios';
 import assert from 'node:assert';
 import { logError } from '../../../utils/axios';
 import { IRateLimiter } from '../../rate-limiter';
-import TIMEOUT from '../constants';
+import { ZEROBOUNCE_TIMEOUT } from '../constants';
 
 /**
  * Configuration options for ZerobounceClient.
@@ -124,7 +124,7 @@ export default class ZerobounceClient {
   ) {
     this.api = axios.create({
       baseURL: config.url ?? ZerobounceClient.baseURL,
-      timeout: TIMEOUT
+      timeout: ZEROBOUNCE_TIMEOUT
     });
   }
 

backend/src/services/rate-limiter/index.ts

@@ -54,6 +54,15 @@ export class TokenBucketRateLimiter implements IRateLimiter {
       executeEvenly,
       distribution
     });
+
+    logger.debug('Token bucket rate limiter initialized', {
+      intervalSeconds,
+      requests,
+      executeEvenly,
+      distribution,
+      uniqueKey
+    });
+
     this.limiter = new RateLimiterQueue(baseLimiter);
   }
 

backend/src/services/signature/llm/index.ts

@@ -172,6 +172,8 @@ type OpenRouterResponse = {
 export class SignatureLLM implements ExtractSignature {
   LLM_ENDPOINT = 'https://openrouter.ai/api/v1/chat/completions';
 
+  MAX_OUTPUT_TOKENS = 1000;
+
   private active = true;
 
   constructor(
@@ -210,13 +212,17 @@ export class SignatureLLM implements ExtractSignature {
           content: SignaturePrompt.buildUserPrompt(email, signature)
         }
       ],
-      response_format: SignaturePrompt.response_format
+      response_format: SignaturePrompt.response_format,
+      max_tokens: this.MAX_OUTPUT_TOKENS
     });
   }
 
   private handleResponseError(error: OpenRouterError['error']) {
     if ([402, 502, 503].includes(error.code)) {
       this.active = false;
+      this.logger.warn(
+        'LLM engine down — check credits balance or expired/invalid API token'
+      );
     }
     throw new Error(error.message);
   }
@@ -279,9 +285,15 @@ export class SignatureLLM implements ExtractSignature {
     try {
       const content = await this.sendPrompt(email, signature);
 
-      this.logger.debug(`extract signature content: ${content}`);
+      if (!content || content.toLowerCase() === 'null') {
+        this.logger.debug('Received empty or null response. skipping parse');
+        return null;
+      }
 
-      if (!content || content.toLowerCase() === 'null') return null;
+      if (content.length > this.MAX_OUTPUT_TOKENS) {
+        this.logger.debug('Model response was too large, skipping parse');
+        return null;
+      }
 
       const parsed = JSON.parse(content);
       const person = Array.isArray(parsed) ? parsed[0] : parsed;