email/sidecar/email-idle.ts

import type { ImapFlow } from 'imapflow';
import { getUserById } from 'officerdb';
import { getAllSyncedAccounts } from '../db/queries';
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<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;
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<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`);
      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<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(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);
}