From 5a0b9ce70fe1ec155fde1ce9d92011a70b2cbfd4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Padez?= Date: Thu, 23 Jul 2026 17:22:46 +0000 Subject: [PATCH] email: real-time IMAP IDLE watchers (push on new mail); cron kept as backstop Co-Authored-By: Claude Opus 4.8 --- src/servers/api/email/resync.ts | 14 ++- src/servers/sidecar/email/email-idle.ts | 157 ++++++++++++++++++++++++ src/servers/sidecar/email/index.ts | 4 + 3 files changed, 174 insertions(+), 1 deletion(-) create mode 100644 src/servers/sidecar/email/email-idle.ts diff --git a/src/servers/api/email/resync.ts b/src/servers/api/email/resync.ts index 5438a815..33f369e7 100644 --- a/src/servers/api/email/resync.ts +++ b/src/servers/api/email/resync.ts @@ -271,7 +271,19 @@ type ResyncParams = { userId: number; }; -export async function performResync({ accountId, userEmail, userId }: ResyncParams): Promise { +// Coalesce concurrent resyncs of the same account within this process (IDLE + cron + manual can all +// fire) — a caller arriving mid-resync just awaits the one already running. +const resyncInFlight = new Map>(); + +export function performResync(params: ResyncParams): Promise { + const running = resyncInFlight.get(params.accountId); + if (running) return running; + const p = doResync(params).finally(() => resyncInFlight.delete(params.accountId)); + resyncInFlight.set(params.accountId, p); + return p; +} + +async function doResync({ accountId, userEmail, userId }: ResyncParams): Promise { const account = await getEmailAccount(accountId); if (!account || account.userId !== userId) throw new Error('Account not found'); diff --git a/src/servers/sidecar/email/email-idle.ts b/src/servers/sidecar/email/email-idle.ts new file mode 100644 index 00000000..8103305d --- /dev/null +++ b/src/servers/sidecar/email/email-idle.ts @@ -0,0 +1,157 @@ +import type { ImapFlow } from 'imapflow'; +import { getAllSyncedAccounts, getUserById } from 'officerdb'; +import { performResync } from '../../api/email/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; + +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`); + } 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(): void { + if (reconcileTimer) return; + 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); +} diff --git a/src/servers/sidecar/email/index.ts b/src/servers/sidecar/email/index.ts index e461ddcb..cbd01f09 100644 --- a/src/servers/sidecar/email/index.ts +++ b/src/servers/sidecar/email/index.ts @@ -1,6 +1,7 @@ import type { SidecarEvent } from '../protocol'; import type { Job, EnqueueParams } from '../../queue/types'; import { initEmailCron, stopEmailCron } from './email-cron'; +import { initEmailIdle, stopEmailIdle } from './email-idle'; import { createSidecarConnector } from '../connect'; const API_URL = process.env.API_URL ?? `ws://127.0.0.1:${process.env.PORT ?? '5000'}`; @@ -84,6 +85,8 @@ const connection = createSidecarConnector({ onConnected() { // Start email cron once connected (so queue commands can reach API server) initEmailCron(); + // Real-time push via IMAP IDLE; the cron above is the slow backstop. + initEmailIdle(); }, }); @@ -91,6 +94,7 @@ const connection = createSidecarConnector({ function shutdown(signal: string) { console.log(`[email] ${signal} received, shutting down...`); + stopEmailIdle(); stopEmailCron(); connection.destroy(); process.exit(0);