email: real-time IMAP IDLE watchers (push on new mail); cron kept as backstop

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

src/servers/api/email/resync.ts

@@ -271,7 +271,19 @@ type ResyncParams = {
   userId: number;
 };
 
-export async function performResync({ accountId, userEmail, userId }: ResyncParams): Promise<ResyncResult> {
+// 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<number, Promise<ResyncResult>>();
+
+export function performResync(params: ResyncParams): Promise<ResyncResult> {
+  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<ResyncResult> {
   const account = await getEmailAccount(accountId);
   if (!account || account.userId !== userId) throw new Error('Account not found');
 

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<ReturnType<typeof getAllSyncedAccounts>>[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<typeof setTimeout> | null;
+};
+
+const RECONNECT_BASE = 5_000;
+const RECONNECT_MAX = 5 * 60_000;
+const RECONCILE_MS = 5 * 60_000;
+
+const watchers = new Map<number, Watcher>();
+let reconcileTimer: ReturnType<typeof setInterval> | 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<string, unknown> | 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<void> {
+  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<void> {
+  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<void> {
+  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);
+}

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);