import { createHash } from 'node:crypto'; import { openEmailDb, upsertFromRawEml } from './store'; import { readSyncMeta, writeSyncMeta } from './sync-meta'; import { refreshGoogleAccessToken } from '../../api/integrations/google-auth'; import { getEmailAccount, getUserIntegration, upsertUserIntegration, getServerIntegration, updateEmailAccountStatus, getDockPaths, setDockPaths, } from 'officerdb'; import type { SyncStep, SyncCtx } from './sync-runner'; // Thrown by the handlers for failures that a retry cannot fix (missing credentials, deleted account). // The queue used it to skip retrying; here it is an ordinary error whose message reaches the sync state. export class PermanentError extends Error {} // ── Types for job meta (passed by the API server at enqueue time) ── type EmailSyncMeta = { emailAccountId: number; userEmail: string; account: { id: number; userId: number; email: string; imapHost: string; imapPort: number; imapSecure: boolean; provider: string; authType: string; credentials: Record; }; imapAuth: { user: string; pass?: string; accessToken?: string }; saved?: number; }; // ── Stable ID from Message-Id header ── function messageIdToStableId(raw: string): string | null { const match = raw.match(/^Message-Id:\s*\s]+)>?/im); if (!match?.[1]) return null; return createHash('sha1').update(match[1]).digest('hex').slice(0, 16); } // ── IMAP folder → label mapping ── const SPECIAL_USE_LABEL_MAP: Record = { '\\Inbox': 'inbox', '\\Sent': 'sent', '\\Drafts': 'draft', '\\Flagged': 'starred', '\\Trash': 'trash', '\\Junk': 'spam', '\\All': 'archive', }; const SKIP_SPECIAL_USE = new Set(['\\Trash', '\\Junk', '\\All']); type FolderInfo = { specialUse?: string; path: string; flags: Set; status?: { uidNext?: number; uidValidity?: number }; }; function shouldSkipFolder(folder: FolderInfo): boolean { if (folder.flags.has('\\Noselect') || folder.flags.has('\\NonExistent')) return true; if (folder.specialUse && SKIP_SPECIAL_USE.has(folder.specialUse)) return true; return false; } function folderToLabel(folder: FolderInfo): string { if (folder.specialUse && SPECIAL_USE_LABEL_MAP[folder.specialUse]) { return SPECIAL_USE_LABEL_MAP[folder.specialUse]!; } if (folder.path === 'INBOX') return 'inbox'; return folder.path.toLowerCase(); } // ── OAuth token refresh ── async function resolveImapAuth(meta: EmailSyncMeta): Promise<{ user: string; pass?: string; accessToken?: string }> { if (meta.account.authType !== 'oauth') return meta.imapAuth; // Refresh OAuth token if expired const userGoogle = await getUserIntegration(meta.account.userId, 'google'); const config = userGoogle?.config as Record | undefined; if (!config?.refreshToken) { throw new PermanentError('Google OAuth not configured — reconnect your Google account'); } const expiresAt = config.expiresAt as number | undefined; const tokenExpired = !expiresAt || expiresAt < Date.now() + 60_000; if (!tokenExpired && config.accessToken) { return { user: meta.account.email, accessToken: config.accessToken as string }; } console.log('[email-sync] Refreshing OAuth token'); const refreshed = await refreshGoogleAccessToken(config.refreshToken as string); const serverGoogle = await getServerIntegration('google'); await upsertUserIntegration({ userId: meta.account.userId, provider: 'google', serverIntegrationId: serverGoogle?.id, config: { ...config, accessToken: refreshed.accessToken, expiresAt: refreshed.expiresAt }, }); return { user: meta.account.email, accessToken: refreshed.accessToken }; } // ── Handler ── const emailSyncHandler = { type: 'email-sync', retry: { delayMs: 15 * 60 * 1000, maxRetries: 5 }, steps: [ { name: 'Sync emails', run: async (ctx: SyncCtx) => { const { ImapFlow } = await import('imapflow'); const meta = ctx.meta as unknown as EmailSyncMeta; const { account, userEmail } = meta; // Resolve auth (refreshes OAuth token if needed) const imapAuth = await resolveImapAuth(meta); // Always re-read the position rather than trusting job meta, which can be stale after a retry. const freshAccount = await getEmailAccount(account.id); if (!freshAccount) throw new PermanentError(`Email account ${account.id} not found`); // Opened here rather than further down because the position now lives inside it. Postgres is // passed only as the one-time backfill for accounts that predate the move — see sync-meta.ts. const db = openEmailDb(userEmail, account.email); const syncMeta: Record = readSyncMeta(db, freshAccount.syncMeta as Record); const isIncremental = !!syncMeta.last_sync_at; await updateEmailAccountStatus(account.id, 'syncing'); console.log(`[email-sync] ${isIncremental ? 'Incremental' : 'Initial'} sync for ${account.email}`); const MAX_RECONNECTS = 10; const RECONNECT_DELAY_MS = 5_000; let reconnects = 0; let saved = 0; let skipped = 0; let errors = 0; let allDone = false; // Load existing IDs for dedup (once, shared across reconnections) const existingIds = new Set(); const rows = db.query('SELECT id FROM emails').all() as Array<{ id: string }>; for (const row of rows) existingIds.add(row.id); try { while (!allDone && reconnects <= MAX_RECONNECTS) { if (reconnects > 0) { console.log( `[email-sync] Reconnecting (${reconnects}/${MAX_RECONNECTS}) after ${RECONNECT_DELAY_MS / 1000}s...`, ); await ctx.updateProgress({ current: saved + skipped, total: 0, label: `Reconnecting (${reconnects}/${MAX_RECONNECTS})...`, }); await new Promise((r) => setTimeout(r, RECONNECT_DELAY_MS)); // Re-read syncMeta from DB to get latest saved UIDs const updated = await getEmailAccount(account.id); if (updated?.syncMeta) { Object.assign(syncMeta, updated.syncMeta as Record); } } else { await ctx.updateProgress({ current: 0, total: 0, label: 'Connecting...' }); } const client = new ImapFlow({ host: account.imapHost, port: account.imapPort, secure: account.imapSecure, auth: imapAuth, logger: false, socketTimeout: 30 * 60 * 1000, }); const connState = { error: null as Error | null }; client.on('error', (err: Error) => { console.log(`[email-sync] IMAP connection error: ${err.message}`); connState.error = err; }); try { await client.connect(); console.log(`[email-sync] IMAP connected${reconnects > 0 ? ` (reconnect ${reconnects})` : ''}`); const folders = (await client.list({ statusQuery: { uidNext: true, uidValidity: true }, })) as FolderInfo[]; // Filter to folders that still need syncing const foldersToSync: Array<{ folder: FolderInfo; lastUid: number }> = []; for (const folder of folders) { if (shouldSkipFolder(folder)) continue; const uidValidityKey = `imap_uidvalidity:${folder.path}`; const lastUidKey = `imap_lastuid:${folder.path}`; const storedUidValidity = syncMeta[uidValidityKey] as string | undefined; const storedLastUid = syncMeta[lastUidKey] as string | undefined; const uidValidity = folder.status?.uidValidity ? String(folder.status.uidValidity) : null; const uidNext = folder.status?.uidNext ?? 0; let lastUid = storedUidValidity && uidValidity === storedUidValidity && storedLastUid ? parseInt(storedLastUid, 10) : 0; if (storedUidValidity && uidValidity && storedUidValidity !== uidValidity) { console.log(`[email-sync] UIDVALIDITY changed for ${folder.path} — will re-scan`); lastUid = 0; } if (lastUid > 0 && uidNext <= lastUid + 1) continue; foldersToSync.push({ folder, lastUid }); } if (foldersToSync.length === 0) { console.log('[email-sync] All folders synced'); allDone = true; await client.logout().catch(() => {}); break; } console.log(`[email-sync] ${foldersToSync.length} folder(s) to sync`); let connectionLost = false; for (let fi = 0; fi < foldersToSync.length; fi++) { const { folder, lastUid } = foldersToSync[fi]!; if (connState.error) { connectionLost = true; break; } console.log(`[email-sync] Syncing folder ${fi + 1}/${foldersToSync.length}: ${folder.path}`); let lock; try { lock = await client.getMailboxLock(folder.path); } catch (err) { const errMsg = err instanceof Error ? err.message : String(err); if (errMsg.includes('not available') || errMsg.includes('timeout') || errMsg.includes('closed')) { console.log(`[email-sync] Connection lost at folder ${folder.path}`); connectionLost = true; break; } console.log(`[email-sync] Skipping folder ${folder.path}: ${errMsg}`); continue; } try { const mailbox = client.mailbox; if (!mailbox) continue; const uidValidity = String(mailbox.uidValidity); const effectiveLastUid = syncMeta[`imap_uidvalidity:${folder.path}`] === uidValidity ? lastUid : 0; let maxUid = effectiveLastUid; const label = folderToLabel(folder); const range = effectiveLastUid > 0 ? `${effectiveLastUid + 1}:*` : '1:*'; try { const fetchOpts: Record = { source: true, uid: true }; if (account.provider === 'gmail') fetchOpts.labels = true; for await (const msg of client.fetch(range, fetchOpts, { uid: true })) { if (msg.uid <= effectiveLastUid) continue; maxUid = Math.max(maxUid, msg.uid); if (!msg.source) { errors++; continue; } const raw = msg.source.toString('utf-8'); const id = messageIdToStableId(raw); if (!id) { errors++; continue; } if (existingIds.has(id)) { skipped++; continue; } const labels = [label]; try { upsertFromRawEml({ db, id, raw, integration: account.provider, emailAccount: account.email, labels, }); existingIds.add(id); saved++; } catch { errors++; } if ((saved + skipped) % 50 === 0) { const progressLabel = `${folder.path} — ${saved.toLocaleString()} saved, ${skipped.toLocaleString()} skipped`; console.log(`[email-sync] ${progressLabel}`); await ctx.updateProgress({ current: saved + skipped, total: 0, label: progressLabel }); if (maxUid > effectiveLastUid) { syncMeta[`imap_lastuid:${folder.path}`] = String(maxUid); writeSyncMeta(db, syncMeta); } } } } catch (fetchErr) { const errMsg = fetchErr instanceof Error ? fetchErr.message : String(fetchErr); if (errMsg.includes('Nothing to fetch')) { // No messages in range — normal } else if ( errMsg.includes('not available') || errMsg.includes('timeout') || errMsg.includes('closed') ) { console.log(`[email-sync] Connection lost during fetch in ${folder.path}: ${errMsg}`); connectionLost = true; } else { console.log(`[email-sync] Fetch error in ${folder.path}: ${errMsg}`); errors++; } } syncMeta[`imap_uidvalidity:${folder.path}`] = uidValidity; if (maxUid > effectiveLastUid) { syncMeta[`imap_lastuid:${folder.path}`] = String(maxUid); } console.log(`[email-sync] ${folder.path} done: ${saved} saved, ${skipped} skipped, ${errors} errors`); await ctx.updateProgress({ current: saved + skipped, total: 0, label: `${folder.path}: ${saved} saved`, }); writeSyncMeta(db, syncMeta); } finally { lock.release(); } if (connectionLost) break; } await client.logout().catch(() => {}); if (connectionLost) { console.log(`[email-sync] Connection lost after saving ${saved} emails — will reconnect`); reconnects++; continue; } allDone = true; } catch (err) { await client.logout().catch(() => {}); if (reconnects < MAX_RECONNECTS) { console.log(`[email-sync] Error: ${err instanceof Error ? err.message : String(err)} — will reconnect`); reconnects++; continue; } throw err; } } if (!allDone) { throw new Error(`IMAP sync incomplete after ${MAX_RECONNECTS} reconnection attempts`); } // Mark sync complete. Last, and only after every message above is stored: if this never runs, // the position stays behind and those messages are fetched again. Re-fetching is wasteful; // advancing past mail that was never stored would lose it silently. syncMeta.last_sync_at = new Date().toISOString(); writeSyncMeta(db, syncMeta); } finally { db.close(); } console.log( `[email-sync] Done: saved ${saved}, skipped ${skipped}, errors ${errors}, reconnects ${reconnects}`, ); ctx.meta.saved = saved; await ctx.updateProgress({ current: 1, total: 1, label: `Done — ${saved} new emails` }); }, }, { name: 'Finalize', run: async (ctx: SyncCtx) => { const meta = ctx.meta as unknown as EmailSyncMeta; const saved = meta.saved ?? 0; await updateEmailAccountStatus(meta.account.id, 'synced'); if (saved > 0) { try { const paths = await getDockPaths(meta.account.userId); if (!paths) { const defaults = ['/', '/files', '/automation', '/dashboards', '/chat']; await setDockPaths(meta.account.userId, [...defaults, '/email']); } else if (!paths.includes('/email')) { await setDockPaths(meta.account.userId, [...paths, '/email']); } } catch { // Non-fatal } } console.log(`[email-sync] ${meta.account.email} status set to synced`); }, }, ], }; /** The handler's steps, for the sidecar's own runner. */ export const imapSyncSteps: SyncStep[] = emailSyncHandler.steps as SyncStep[];