import { ImapFlow } from "imapflow"; import { simpleParser } from "mailparser"; import { prisma } from "./db"; import { enqueueProcessing } from "./processing-queue"; const ALLOWED_CONTENT_TYPES = [ "application/pdf", "image/jpeg", "image/jpg", "image/png", "image/webp", ]; const MAX_ATTACHMENT_SIZE = 50 * 1024 * 1024; // 50 MB type EmailConfig = { host: string; port: number; user: string; pass: string; tls: boolean; folder: string; processedAction: string; processedFolder: string; }; let client: ImapFlow | null = null; let watching = false; let reconnectTimer: ReturnType | null = null; let currentConfig: EmailConfig | null = null; function log(msg: string, ...args: unknown[]) { console.log(`[email-watcher] ${msg}`, ...args); } function logError(msg: string, ...args: unknown[]) { console.error(`[email-watcher] ${msg}`, ...args); } async function handleMessage(client: ImapFlow, uid: number, config: EmailConfig) { const result = await client.fetchOne(String(uid), { source: true }, { uid: true }); if (!result) return; const rawSource = (result as unknown as { source?: Buffer }).source; if (!rawSource) return; const parsed = await simpleParser(rawSource); const subject = parsed.subject || "(no subject)"; const from = parsed.from?.text || "unknown"; const attachments = (parsed.attachments || []).filter((att) => { const ct = att.contentType?.toLowerCase() || ""; return ALLOWED_CONTENT_TYPES.includes(ct); }); if (attachments.length === 0) { log(`Skipping email from ${from} "${subject}" — no valid attachments`); return; } log(`Processing email from ${from} "${subject}" — ${attachments.length} attachment(s)`); for (const att of attachments) { const fileName = att.filename || `email-attachment-${Date.now()}`; const buffer = att.content; if (buffer.length > MAX_ATTACHMENT_SIZE) { log(`Skipping attachment "${fileName}" — exceeds 50 MB size limit (${(buffer.length / 1024 / 1024).toFixed(1)} MB)`); continue; } const isPdf = att.contentType?.toLowerCase() === "application/pdf"; try { const job = await prisma.processingJob.create({ data: { fileName, filePath: `email/${fileName}`, status: "queued", }, }); enqueueProcessing(job.id, fileName, buffer, isPdf).catch((err) => { logError(`Processing failed for ${fileName}:`, err); }); log(`Queued: ${fileName} (job ${job.id})`); } catch (err) { logError(`Failed to create job for ${fileName}:`, err); } } if (config.processedAction === "mark_read") { await client.messageFlagsAdd(String(uid), ["\\Seen"], { uid: true }); } else if (config.processedAction === "move") { try { await client.messageMove(String(uid), config.processedFolder, { uid: true }); } catch { log(`Could not move to "${config.processedFolder}", marking as read instead`); await client.messageFlagsAdd(String(uid), ["\\Seen"], { uid: true }); } } else if (config.processedAction === "delete") { await client.messageDelete(String(uid), { uid: true }); } } async function pollLoop(config: EmailConfig) { if (!client || !watching) return; try { const lock = await client.getMailboxLock(config.folder); try { const searchResult = await client.search({ seen: false }, { uid: true }); const unseen = Array.isArray(searchResult) ? searchResult : []; if (unseen.length > 0) { log(`Found ${unseen.length} unseen message(s)`); for (const uid of unseen) { if (!watching) break; try { await handleMessage(client, uid, config); } catch (err) { logError(`Error handling message UID ${uid}:`, err); } } } } finally { lock.release(); } } catch (err) { logError("Error during poll:", err); } if (!watching || !client) return; try { await client.idle(); } catch { // IDLE interrupted or connection lost — reconnect handles it } if (watching) { setImmediate(() => pollLoop(config)); } } async function connect(config: EmailConfig) { if (client) { try { await client.logout(); } catch {} client = null; } client = new ImapFlow({ host: config.host, port: config.port, secure: config.tls, auth: { user: config.user, pass: config.pass }, logger: false, emitLogs: false, }); client.on("error", (err: Error) => { logError("Connection error:", err.message); if (watching) scheduleReconnect(config); }); client.on("close", () => { log("Connection closed"); if (watching) scheduleReconnect(config); }); await client.connect(); log(`Connected to ${config.host}:${config.port} as ${config.user}`); pollLoop(config); } function scheduleReconnect(config: EmailConfig) { if (reconnectTimer) return; const delay = 10_000; log(`Reconnecting in ${delay / 1000}s...`); reconnectTimer = setTimeout(async () => { reconnectTimer = null; if (!watching) return; try { await connect(config); } catch (err) { logError("Reconnect failed:", err); scheduleReconnect(config); } }, delay); } export async function startEmailWatching(config: EmailConfig): Promise { if (watching) await stopEmailWatching(); watching = true; currentConfig = config; await prisma.appSettings.upsert({ where: { id: "singleton" }, update: { emailWatching: true }, create: { id: "singleton", emailWatching: true }, }); await connect(config); } export async function stopEmailWatching(): Promise { watching = false; currentConfig = null; if (reconnectTimer) { clearTimeout(reconnectTimer); reconnectTimer = null; } if (client) { try { await client.logout(); } catch {} client = null; } await prisma.appSettings.upsert({ where: { id: "singleton" }, update: { emailWatching: false }, create: { id: "singleton", emailWatching: false }, }); log("Stopped"); } export function isEmailWatching(): boolean { return watching; } export function getEmailWatchStatus() { return { watching, connected: client !== null && client.usable !== false, config: currentConfig ? { host: currentConfig.host, user: currentConfig.user, folder: currentConfig.folder } : null, }; } export async function scanInbox(): Promise<{ processed: number; skipped: number; error?: string }> { const settings = await prisma.appSettings.findUnique({ where: { id: "singleton" } }); if (!settings?.emailImapHost || !settings?.emailImapUser || !settings?.emailImapPass) { return { processed: 0, skipped: 0, error: "IMAP not configured" }; } const config: EmailConfig = { host: settings.emailImapHost, port: settings.emailImapPort, user: settings.emailImapUser, pass: settings.emailImapPass, tls: settings.emailImapTls, folder: settings.emailFolder, processedAction: settings.emailProcessed, processedFolder: settings.emailProcessedFolder, }; const scanClient = new ImapFlow({ host: config.host, port: config.port, secure: config.tls, auth: { user: config.user, pass: config.pass }, logger: false, emitLogs: false, }); let processed = 0; let skipped = 0; try { await scanClient.connect(); log(`[scan] Connected for manual inbox scan`); const lock = await scanClient.getMailboxLock(config.folder); try { const searchResult = await scanClient.search({ seen: false }, { uid: true }); const unseen = Array.isArray(searchResult) ? searchResult : []; log(`[scan] Found ${unseen.length} unseen message(s)`); for (const uid of unseen) { try { await handleMessage(scanClient, uid, config); processed++; } catch (err) { logError(`[scan] Error handling message UID ${uid}:`, err); skipped++; } } } finally { lock.release(); } await scanClient.logout(); } catch (err) { const message = err instanceof Error ? err.message : "Scan failed"; logError("[scan] Error:", message); try { await scanClient.logout(); } catch {} return { processed, skipped, error: message }; } log(`[scan] Complete: ${processed} processed, ${skipped} skipped`); return { processed, skipped }; } export async function testEmailConnection(config: Omit): Promise<{ ok: boolean; folders?: string[]; error?: string }> { let testClient: ImapFlow | null = null; try { testClient = new ImapFlow({ host: config.host, port: config.port, secure: config.tls, auth: { user: config.user, pass: config.pass }, logger: false, emitLogs: false, }); await testClient.connect(); const folders: string[] = []; const list = await testClient.list(); for (const folder of list) { folders.push(folder.path); } await testClient.logout(); return { ok: true, folders }; } catch (err) { const message = err instanceof Error ? err.message : "Connection failed"; return { ok: false, error: message }; } finally { if (testClient) { try { await testClient.logout(); } catch {} } } }