port-nimara-client-portal/server/utils/email-sync.ts

import { getMinioClient } from './minio';
import { getIMAPPool } from './imap-pool';
import type { ParsedMail } from 'mailparser';
import { simpleParser } from 'mailparser';

// Email sync service for MinIO-based email management
export interface EmailSyncMetadata {
  lastSyncTime: string;
  totalEmails: number;
  lastError?: string;
  syncStatus: 'idle' | 'syncing' | 'error';
}

export interface EmailThreadIndex {
  threads: Array<{
    id: string;
    subject: string;
    participants: string[];
    emailCount: number;
    lastActivity: string;
    hasAttachments: boolean;
  }>;
  lastUpdated: string;
}

// Get or create sync metadata
export async function getSyncMetadata(interestId: string): Promise<EmailSyncMetadata> {
  const client = getMinioClient();
  const objectName = `interest-${interestId}/metadata.json`;
  
  try {
    const stream = await client.getObject('client-emails', objectName);
    let data = '';
    
    await new Promise((resolve, reject) => {
      stream.on('data', (chunk) => { data += chunk; });
      stream.on('end', resolve);
      stream.on('error', reject);
    });
    
    return JSON.parse(data);
  } catch (error: any) {
    // If not found, create default metadata
    if (error.code === 'NoSuchKey') {
      const defaultMetadata: EmailSyncMetadata = {
        lastSyncTime: new Date(0).toISOString(), // Start from beginning
        totalEmails: 0,
        syncStatus: 'idle'
      };
      
      await saveSyncMetadata(interestId, defaultMetadata);
      return defaultMetadata;
    }
    
    throw error;
  }
}

// Save sync metadata
export async function saveSyncMetadata(interestId: string, metadata: EmailSyncMetadata): Promise<void> {
  const client = getMinioClient();
  const objectName = `interest-${interestId}/metadata.json`;
  const buffer = Buffer.from(JSON.stringify(metadata, null, 2));
  
  await client.putObject('client-emails', objectName, buffer, buffer.length, {
    'Content-Type': 'application/json'
  });
}

// Get thread index
export async function getThreadIndex(interestId: string): Promise<EmailThreadIndex> {
  const client = getMinioClient();
  const objectName = `interest-${interestId}/threads/index.json`;
  
  try {
    const stream = await client.getObject('client-emails', objectName);
    let data = '';
    
    await new Promise((resolve, reject) => {
      stream.on('data', (chunk) => { data += chunk; });
      stream.on('end', resolve);
      stream.on('error', reject);
    });
    
    return JSON.parse(data);
  } catch (error: any) {
    if (error.code === 'NoSuchKey') {
      return {
        threads: [],
        lastUpdated: new Date().toISOString()
      };
    }
    throw error;
  }
}

// Save thread index
export async function saveThreadIndex(interestId: string, index: EmailThreadIndex): Promise<void> {
  const client = getMinioClient();
  const objectName = `interest-${interestId}/threads/index.json`;
  const buffer = Buffer.from(JSON.stringify(index, null, 2));
  
  await client.putObject('client-emails', objectName, buffer, buffer.length, {
    'Content-Type': 'application/json'
  });
}

// Sync emails with exponential backoff retry
export async function syncEmailsWithRetry(
  sessionId: string,
  userEmail: string,
  clientEmail: string,
  interestId: string,
  maxRetries: number = 3
): Promise<void> {
  let retryCount = 0;
  let lastError: Error | null = null;
  
  while (retryCount <= maxRetries) {
    try {
      await syncEmails(sessionId, userEmail, clientEmail, interestId);
      return; // Success
    } catch (error: any) {
      lastError = error;
      retryCount++;
      
      if (retryCount > maxRetries) {
        throw error;
      }
      
      // Exponential backoff: 1s, 2s, 4s, 8s
      const waitTime = Math.pow(2, retryCount - 1) * 1000;
      console.log(`[EmailSync] Retry ${retryCount}/${maxRetries} after ${waitTime}ms`);
      
      await new Promise(resolve => setTimeout(resolve, waitTime));
    }
  }
  
  throw lastError || new Error('Failed to sync emails after retries');
}

// Main sync function
async function syncEmails(
  sessionId: string,
  userEmail: string,
  clientEmail: string,
  interestId: string
): Promise<void> {
  const metadata = await getSyncMetadata(interestId);
  
  // Update status to syncing
  metadata.syncStatus = 'syncing';
  await saveSyncMetadata(interestId, metadata);
  
  try {
    const pool = getIMAPPool();
    const imap = await pool.getConnection(sessionId);
    
    // Fetch emails newer than last sync
    const lastSyncDate = new Date(metadata.lastSyncTime);
    const newEmails = await fetchNewEmails(imap, userEmail, clientEmail, lastSyncDate);
    
    if (newEmails.length > 0) {
      // Save new emails to MinIO
      for (const email of newEmails) {
        const emailId = email.messageId || `${Date.now()}-${Math.random()}`;
        const objectName = `interest-${interestId}/emails/${emailId}.json`;
        const buffer = Buffer.from(JSON.stringify(email, null, 2));
        
        const client = getMinioClient();
        await client.putObject('client-emails', objectName, buffer, buffer.length, {
          'Content-Type': 'application/json'
        });
      }
      
      // Update thread index
      await updateThreadIndex(interestId, newEmails);
      
      // Update metadata
      metadata.lastSyncTime = new Date().toISOString();
      metadata.totalEmails += newEmails.length;
      metadata.syncStatus = 'idle';
      await saveSyncMetadata(interestId, metadata);
    }
  } catch (error: any) {
    // Update metadata with error
    metadata.syncStatus = 'error';
    metadata.lastError = error.message;
    await saveSyncMetadata(interestId, metadata);
    
    throw error;
  }
}

// Fetch new emails from IMAP
async function fetchNewEmails(
  imap: any,
  userEmail: string,
  clientEmail: string,
  since: Date
): Promise<any[]> {
  return new Promise((resolve, reject) => {
    const emails: any[] = [];
    
    imap.openBox('INBOX', true, (err: any) => {
      if (err) {
        reject(err);
        return;
      }
      
      // Search for emails newer than last sync
      const searchCriteria = [
        ['SINCE', since.toISOString().split('T')[0]],
        ['OR', 
          ['FROM', clientEmail],
          ['TO', clientEmail]
        ]
      ];
      
      imap.search(searchCriteria, (err: any, results: number[]) => {
        if (err) {
          reject(err);
          return;
        }
        
        if (results.length === 0) {
          resolve([]);
          return;
        }
        
        const fetch = imap.fetch(results, {
          bodies: '',
          struct: true,
          envelope: true
        });
        
        fetch.on('message', (msg: any) => {
          msg.on('body', (stream: any) => {
            simpleParser(stream, (err: any, parsed: ParsedMail) => {
              if (!err && parsed) {
                // Handle from/to addresses which can be single or array
                const fromText = Array.isArray(parsed.from) 
                  ? parsed.from.map(addr => addr.text).join(', ')
                  : parsed.from?.text || '';
                  
                const toText = Array.isArray(parsed.to)
                  ? parsed.to.map(addr => addr.text).join(', ')
                  : parsed.to?.text || '';
                
                emails.push({
                  id: parsed.messageId,
                  from: fromText,
                  to: toText,
                  subject: parsed.subject,
                  body: parsed.text,
                  html: parsed.html,
                  timestamp: parsed.date?.toISOString(),
                  attachments: parsed.attachments?.map(att => ({
                    filename: att.filename,
                    contentType: att.contentType,
                    size: att.size
                  }))
                });
              }
            });
          });
        });
        
        fetch.once('end', () => {
          resolve(emails);
        });
        
        fetch.once('error', (err: any) => {
          reject(err);
        });
      });
    });
  });
}

// Update thread index with new emails
async function updateThreadIndex(interestId: string, newEmails: any[]): Promise<void> {
  const index = await getThreadIndex(interestId);
  
  // Group emails by thread (simplified - by subject)
  for (const email of newEmails) {
    const normalizedSubject = email.subject
      ?.replace(/^(Re:|Fwd:|Fw:|RE:|FW:|FWD:)\s*/gi, '')
      .trim() || 'No Subject';
    
    let thread = index.threads.find(t => 
      t.subject.replace(/^(Re:|Fwd:|Fw:|RE:|FW:|FWD:)\s*/gi, '').trim() === normalizedSubject
    );
    
    if (!thread) {
      thread = {
        id: `thread-${Date.now()}-${Math.random()}`,
        subject: email.subject || 'No Subject',
        participants: [],
        emailCount: 0,
        lastActivity: email.timestamp,
        hasAttachments: false
      };
      index.threads.push(thread);
    }
    
    // Update thread
    thread.emailCount++;
    thread.lastActivity = email.timestamp;
    if (email.attachments?.length > 0) {
      thread.hasAttachments = true;
    }
    
    // Add participants
    const from = email.from?.match(/<(.+)>/)?.[1] || email.from;
    if (from && !thread.participants.includes(from)) {
      thread.participants.push(from);
    }
  }
  
  // Sort threads by last activity
  index.threads.sort((a, b) => 
    new Date(b.lastActivity).getTime() - new Date(a.lastActivity).getTime()
  );
  
  index.lastUpdated = new Date().toISOString();
  await saveThreadIndex(interestId, index);
}

// Get emails from MinIO cache
export async function getCachedEmails(interestId: string): Promise<any[]> {
  const client = getMinioClient();
  const emails: any[] = [];
  
  try {
    const stream = client.listObjectsV2('client-emails', `interest-${interestId}/emails/`, true);
    const files: any[] = [];
    
    await new Promise((resolve, reject) => {
      stream.on('data', (obj) => {
        if (obj && obj.name && obj.name.endsWith('.json')) {
          files.push(obj.name);
        }
      });
      stream.on('error', reject);
      stream.on('end', resolve);
    });
    
    // Load each email
    for (const fileName of files) {
      try {
        const objStream = await client.getObject('client-emails', fileName);
        let data = '';
        
        await new Promise((resolve, reject) => {
          objStream.on('data', (chunk) => { data += chunk; });
          objStream.on('end', resolve);
          objStream.on('error', reject);
        });
        
        emails.push(JSON.parse(data));
      } catch (error) {
        console.error(`Failed to load email ${fileName}:`, error);
      }
    }
  } catch (error) {
    console.error('Failed to list cached emails:', error);
  }
  
  return emails;
}