import type { ImapFlow } from 'imapflow'; import { getAllSyncedAccounts, getUserById } from 'officerdb'; import { performResync } from './resync'; import { getValidGoogleAccessToken } from '../../api/integrations/google-auth'; // Real-time email via IMAP IDLE: one persistent connection per account. imapflow auto-enters IDLE // while a mailbox is open and pushes an `exists` event the instant new mail lands — we then run the // same incremental resync the cron uses. The cron stays on as a slow backstop for missed events. type Account = Awaited>[number]; type Watcher = { account: Account; client: ImapFlow | null; closing: boolean; syncing: boolean; pending: boolean; // an exists event arrived mid-sync → run once more backoff: number; reconnectTimer: ReturnType | null; }; const RECONNECT_BASE = 5_000; const RECONNECT_MAX = 5 * 60_000; const RECONCILE_MS = 5 * 60_000; const watchers = new Map(); let reconcileTimer: ReturnType | null = null; let notifyNewMail: ((userEmail: string) => void) | null = null; async function resolveAuth(account: Account): Promise<{ user: string; pass?: string; accessToken?: string } | null> { if (account.authType === 'oauth') { const token = await getValidGoogleAccessToken(account.userId).catch(() => null); return token ? { user: account.email, accessToken: token } : null; } const pass = (account.credentials as Record | null)?.password as string | undefined; return pass ? { user: account.email, pass } : null; } // Coalesced incremental sync — a new event during a sync queues exactly one follow-up. async function runSync(w: Watcher): Promise { if (w.syncing) { w.pending = true; return; } w.syncing = true; try { const user = await getUserById(w.account.userId); if (!user) return; const result = await performResync({ accountId: w.account.id, userEmail: user.email, userId: user.id }); if (result.saved > 0) { console.log(`[email-idle] ${w.account.email}: ${result.saved} new`); notifyNewMail?.(user.email); } } catch (err) { console.error(`[email-idle] sync failed for ${w.account.email}:`, err instanceof Error ? err.message : err); } finally { w.syncing = false; if (w.pending) { w.pending = false; void runSync(w); } } } function scheduleReconnect(w: Watcher): void { if (w.closing || w.reconnectTimer) return; const delay = w.backoff; w.backoff = Math.min(w.backoff * 2, RECONNECT_MAX); w.reconnectTimer = setTimeout(() => { w.reconnectTimer = null; void connect(w); }, delay); } async function connect(w: Watcher): Promise { if (w.closing) return; const auth = await resolveAuth(w.account); if (!auth) { scheduleReconnect(w); return; } const { ImapFlow } = await import('imapflow'); const client = new ImapFlow({ host: w.account.imapHost, port: w.account.imapPort, secure: w.account.imapSecure, auth, logger: false, emitLogs: false, }); w.client = client; client.on('exists', () => void runSync(w)); client.on('error', () => { /* a 'close' event follows and handles the reconnect */ }); client.on('close', () => { w.client = null; scheduleReconnect(w); }); try { await client.connect(); await client.mailboxOpen('INBOX'); // imapflow now auto-IDLEs and emits `exists` on new mail w.backoff = RECONNECT_BASE; // reset after a clean connect console.log(`[email-idle] watching ${w.account.email}`); void runSync(w); // catch up on anything that arrived while disconnected } catch (err) { console.error(`[email-idle] connect failed for ${w.account.email}:`, err instanceof Error ? err.message : err); w.client = null; scheduleReconnect(w); } } function startWatcher(account: Account): void { if (watchers.has(account.id)) return; const w: Watcher = { account, client: null, closing: false, syncing: false, pending: false, backoff: RECONNECT_BASE, reconnectTimer: null }; watchers.set(account.id, w); void connect(w); } function stopWatcher(id: number): void { const w = watchers.get(id); if (!w) return; w.closing = true; if (w.reconnectTimer) clearTimeout(w.reconnectTimer); w.client?.logout().catch(() => {}); watchers.delete(id); } // Keep watchers in sync with the account list (added/removed/credential changes). async function reconcile(): Promise { try { const accounts = await getAllSyncedAccounts(); const ids = new Set(accounts.map((a) => a.id)); for (const a of accounts) { const w = watchers.get(a.id); if (!w) startWatcher(a); else w.account = a; // refresh host/creds for the next reconnect } for (const id of [...watchers.keys()]) if (!ids.has(id)) stopWatcher(id); } catch (err) { console.error('[email-idle] reconcile error:', err instanceof Error ? err.message : err); } } export function initEmailIdle(onNewMail?: (userEmail: string) => void): void { if (reconcileTimer) return; notifyNewMail = onNewMail ?? null; console.log('[email-idle] starting IMAP IDLE watchers'); setTimeout(() => void reconcile(), 5_000); reconcileTimer = setInterval(() => void reconcile(), RECONCILE_MS); } export function stopEmailIdle(): void { if (reconcileTimer) { clearInterval(reconcileTimer); reconcileTimer = null; } for (const id of [...watchers.keys()]) stopWatcher(id); }