import { ImapFlow } from "imapflow"; import { simpleParser } from "mailparser"; import { prisma } from "./db"; import { processFile } from "./ocr"; const ALLOWED_CONTENT_TYPES = [ "application/pdf", "image/jpeg", "image/jpg", "image/png", "image/webp", ]; 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; const isPdf = att.contentType?.toLowerCase() === "application/pdf"; try { const job = await prisma.processingJob.create({ data: { fileName, filePath: `email/${fileName}`, status: "queued", }, }); processFile(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 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 {} } } }