signature, imap fetch refactoring & unittest (#2510)

Author: malek10xdev

backend/src/services/cache/EmailSignatureCache.ts

@@ -7,6 +7,7 @@ export interface EmailSignatureWithMetadata {
   userId: string;
   signature: string;
   email: string;
+  messageId: string;
   firstSeenDate: string;
   lastSeenDate: string;
 }
@@ -24,6 +25,7 @@ export default interface EmailSignatureCache {
     userId: string,
     email: string,
     signature: string,
+    messageId: string,
     messageDate: string,
     miningId: string
   ): Promise<void>;

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

@@ -14,6 +14,7 @@ export default class RedisEmailSignatureCache implements EmailSignatureCache {
     userId: string,
     email: string,
     signature: string,
+    messageId: string,
     messageDate: string,
     miningId: string
   ): Promise<void> {
@@ -33,6 +34,7 @@ export default class RedisEmailSignatureCache implements EmailSignatureCache {
       signature,
       firstSeenDate,
       lastSeenDate: messageDateISO,
+      messageId,
       userId,
       email
     });
@@ -77,6 +79,7 @@ export default class RedisEmailSignatureCache implements EmailSignatureCache {
         signature: data.signature,
         firstSeenDate: data.firstSeenDate,
         lastSeenDate: data.lastSeenDate,
+        messageId: data.messageId,
         userId: data.userId,
         email: data.email
       };

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

@@ -244,7 +244,10 @@ export class SignatureLLM implements ExtractSignature {
 
       return (data as OpenRouterResponse).choices?.[0]?.message?.content;
     } catch (err) {
-      this.logger.error('SignatureExtractionLLM error:', { error: err });
+      this.logger.error(
+        `SignatureExtractionLLM error: ${(err as Error).message}`,
+        { error: err }
+      );
       return null;
     }
   }

backend/src/workers/email-signature/handler.ts

@@ -7,7 +7,11 @@ import planer from 'planer';
 import EmailSignatureCache from '../../services/cache/EmailSignatureCache';
 import { Contact } from '../../db/types';
 import logger from '../../utils/logger';
-import { isUsefulSignatureContent, pushNotificationDB } from './utils';
+import {
+  isUsefulSignatureContent,
+  pushNotificationDB,
+  upsertSignaturesDB
+} from './utils';
 import { ExtractSignature } from '../../services/signature/types';
 import { DomainStatusVerificationFunction } from '../../services/extractors/engines/EmailMessage';
 import { CleanQuotedForwardedReplies } from '../../utils/helpers/emailParsers';
@@ -79,11 +83,13 @@ export class EmailSignatureProcessor {
     contacts: Partial<Contact>[] | null;
   }> {
     const { userId, miningId, data: payload } = data;
-    const { from, messageDate } = payload.header ?? {};
+    const { from, messageDate, rawHeader } = payload.header ?? {};
 
     const shouldProcess = await this.isWorthProcessing(data);
 
     if (shouldProcess) {
+      const [messageId] = rawHeader['message-id'];
+
       this.logging.debug('Processing new signature', {
         userId,
         miningId,
@@ -96,6 +102,7 @@ export class EmailSignatureProcessor {
         miningId,
         from?.address,
         payload.body,
+        messageId,
         messageDate
       );
     }
@@ -110,24 +117,36 @@ export class EmailSignatureProcessor {
 
     if (extracted.length) {
       try {
+        const signatures = extracted.map(
+          ([personEmail, messageId, rawSignature, extractedSignature]) => ({
+            userId,
+            personEmail,
+            messageId,
+            rawSignature,
+            extractedSignature,
+            details: { miningId }
+          })
+        );
+
+        await upsertSignaturesDB(this.supabase, signatures);
+
         await pushNotificationDB(this.supabase, {
           userId,
           type: 'signature',
           details: {
-            extracted,
             signatures: extracted.length
           }
         });
       } catch (err) {
         this.logging.error(
-          `Error when pushing notifications: ${(err as Error).message}`,
+          `Error when inserting signatures/notifications: ${(err as Error).message}`,
           err
         );
       }
     }
     return {
       finished: true,
-      contacts: extracted
+      contacts: extracted.map(([, , , contact]) => contact)
     };
   }
 
@@ -136,6 +155,7 @@ export class EmailSignatureProcessor {
     miningId: string,
     email: string,
     body: string,
+    messageId: string,
     messageDate: string
   ): Promise<void> {
     const signature = this.extractSignature(body);
@@ -158,7 +178,14 @@ export class EmailSignatureProcessor {
       return;
     }
 
-    await this.cache.set(userId, email, signature, messageDate, miningId);
+    await this.cache.set(
+      userId,
+      email,
+      signature,
+      messageId,
+      messageDate,
+      miningId
+    );
     this.logging.info('Cached new signature', {
       email,
       miningId,
@@ -170,7 +197,7 @@ export class EmailSignatureProcessor {
   private async handleBatchUpdate(
     userId: string,
     miningId: string
-  ): Promise<Partial<Contact>[]> {
+  ): Promise<[string, string, string, Partial<Contact>][]> {
     this.logging.debug('handleBatchUpdate()', { userId, miningId });
 
     const all = await this.cache.getAllFromMining(miningId);
@@ -180,25 +207,31 @@ export class EmailSignatureProcessor {
       return [];
     }
 
-    const contacts: (Partial<Contact> | undefined)[] = await Promise.all(
-      all.map(async ({ email, signature }) => {
-        try {
-          const contact = await this.extractContact(userId, email, signature);
-          if (contact) {
-            await this.upsertContact(contact);
-            return contact;
+    const contacts: ([string, string, string, Partial<Contact>] | undefined)[] =
+      await Promise.all(
+        all.map(async ({ email, signature, messageId }) => {
+          try {
+            const contact = await this.extractContact(userId, email, signature);
+            if (contact) {
+              await this.upsertContact(contact);
+              return [email, messageId, signature, contact];
+            }
+            return undefined;
+          } catch (err) {
+            this.logging.error('Error on extract/insert contact', err);
+            return undefined;
           }
-          return undefined;
-        } catch (err) {
-          this.logging.error('Error on extract/insert contact', err);
-          return undefined;
-        }
-      })
-    );
+        })
+      );
 
     await this.cache.clearCachedSignature(miningId);
 
-    const successfulContacts = contacts.filter(Boolean) as Contact[];
+    const successfulContacts = contacts.filter((c) => c && contacts.length) as [
+      string,
+      string,
+      string,
+      Partial<Contact>
+    ][];
 
     this.logging.info('Batch complete - cache cleared', {
       miningId,

backend/src/workers/email-signature/utils.ts

@@ -1,5 +1,4 @@
 import { SupabaseClient } from '@supabase/supabase-js';
-import { Contact } from '../../db/types';
 
 export type NotificationType = 'enrich' | 'clean' | 'extract' | 'signature';
 
@@ -8,10 +7,44 @@ export interface NotificationPayload {
   type: NotificationType;
   details: {
     signatures: number;
-    extracted: Partial<Contact>[];
   };
 }
 
+export interface SignaturePayload {
+  userId: string;
+  personEmail: string;
+  messageId: string;
+  rawSignature?: string | null;
+  extractedSignature?: Record<string, string> | null;
+  details?: Record<string, unknown> | null;
+}
+
+/**
+ * Bulk upsert signatures into `private.signatures`.
+ * If (user_id, person_email) already exists, updates all fields.
+ */
+export async function upsertSignaturesDB(
+  supabase: SupabaseClient,
+  signatures: SignaturePayload[]
+) {
+  const rows = signatures.map((s) => ({
+    user_id: s.userId,
+    person_email: s.personEmail,
+    message_id: s.messageId,
+    raw_signature: s.rawSignature,
+    extracted_signature: s.extractedSignature,
+    details: s.details
+  }));
+
+  const { error } = await supabase
+    .schema('private')
+    .from('signatures')
+    .upsert(rows, {
+      onConflict: 'user_id,person_email'
+    });
+  if (error) throw error;
+}
+
 /**
  * Push a new notification to the Supabase `notifications` table.
  */