From 4ececbe748c676830e85d0997c41f4155ff9825e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Padez?= Date: Fri, 6 Mar 2026 07:02:51 +0000 Subject: [PATCH] email sync: API passes full config at enqueue, auto-reconnect on IMAP drops MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Sidecar/queue runner no longer needs job-specific context. API server resolves account details, IMAP auth, and user email at enqueue time — all persisted in the job file. Handler reads directly from job meta. Removed "Load account" step. Sync step auto-reconnects up to 10 times when Gmail drops the connection, resuming from saved UIDs. Co-Authored-By: Claude Opus 4.6 --- src/servers/api/email/accounts.ts | 56 ++- src/servers/queue/handlers/email-sync.ts | 459 ++++++++++++----------- src/servers/sidecar/email-cron.ts | 35 +- src/servers/sidecar/queue-runner.ts | 7 + 4 files changed, 336 insertions(+), 221 deletions(-) diff --git a/src/servers/api/email/accounts.ts b/src/servers/api/email/accounts.ts index 3aa0b13f..10ef0605 100644 --- a/src/servers/api/email/accounts.ts +++ b/src/servers/api/email/accounts.ts @@ -36,6 +36,39 @@ export const accountsRouter = createRouter(); accountsRouter.get('/', async (ctx) => { const user = ctx.get('user'); const accounts = await getEmailAccounts(user.id); + + // Check for stale syncing/queued accounts with no active job + const staleIds: number[] = []; + const hasActiveAccounts = accounts.some((a) => a.status === 'syncing' || a.status === 'queued'); + let activeJobAccountIds = new Set(); + + if (hasActiveAccounts) { + try { + const jobs = await sidecar.listJobs(); + activeJobAccountIds = new Set( + jobs + .filter((j) => j.type === 'email-sync' && (j.status === 'queued' || j.status === 'running')) + .map((j) => (j.meta as Record | undefined)?.emailAccountId as number) + .filter(Boolean), + ); + } catch { + // Sidecar unavailable — all syncing/queued accounts are stale + } + + for (const a of accounts) { + if ((a.status === 'syncing' || a.status === 'queued') && !activeJobAccountIds.has(a.id)) { + staleIds.push(a.id); + } + } + + // Reset stale accounts in background + if (staleIds.length > 0) { + for (const id of staleIds) { + updateEmailAccountStatus(id, 'connected').catch(() => {}); + } + } + } + return ctx.json( accounts.map((a) => ({ id: a.id, @@ -43,7 +76,7 @@ accountsRouter.get('/', async (ctx) => { email: a.email, displayName: a.displayName, enabled: a.enabled, - status: a.status, + status: staleIds.includes(a.id) ? 'connected' : a.status, createdAt: a.createdAt, })), ); @@ -104,6 +137,10 @@ accountsRouter.post('/:id/sync', async (ctx) => { if (account.status === 'syncing') throw BAD_REQUEST('Account is already syncing'); if (account.status === 'synced') throw BAD_REQUEST('Account is already synced — incremental syncs run automatically'); + // Resolve auth before enqueueing + const authResult = await resolveAuth(user.id, account.authType, account.email, account.credentials as Record); + if (!authResult.ok) throw BAD_REQUEST(authResult.error); + // Set status immediately so the UI reflects the queued state await updateEmailAccountStatus(id, 'queued'); @@ -111,7 +148,22 @@ accountsRouter.post('/:id/sync', async (ctx) => { lane: 'email', type: 'email-sync', userId: user.email, - meta: { emailAccountId: id }, + meta: { + emailAccountId: id, + userEmail: user.email, + account: { + id: account.id, + userId: account.userId, + email: account.email, + imapHost: account.imapHost, + imapPort: account.imapPort, + imapSecure: account.imapSecure, + provider: account.provider, + authType: account.authType, + credentials: account.credentials, + }, + imapAuth: { user: account.email, ...authResult.auth }, + }, }); return ctx.json({ ok: true, jobId: job.id }, 201); diff --git a/src/servers/queue/handlers/email-sync.ts b/src/servers/queue/handlers/email-sync.ts index 0130e9ea..6e11706c 100644 --- a/src/servers/queue/handlers/email-sync.ts +++ b/src/servers/queue/handlers/email-sync.ts @@ -2,11 +2,10 @@ import { createHash } from 'node:crypto'; import type { JobHandler } from '../types'; import { PermanentError } from '../types'; import { registerHandler } from '../handler-registry'; -import { openEmailDb, upsertFromRawEml, setSyncMeta, getSyncMeta } from '../../api/email/email-db'; +import { openEmailDb, upsertFromRawEml } from '../../api/email/email-db'; import { refreshGoogleAccessToken } from '../../api/integrations/google-auth'; import { getEmailAccount, - getUserById, getUserIntegration, upsertUserIntegration, getServerIntegration, @@ -16,6 +15,26 @@ import { setDockPaths, } from 'officerdb'; +// ── 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 { @@ -54,271 +73,279 @@ function folderToLabel(folder: FolderInfo): string { 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: JobHandler = { type: 'email-sync', retry: { delayMs: 15 * 60 * 1000, maxRetries: 5 }, steps: [ - { - name: 'Load account', - run: async (ctx) => { - const emailAccountId = ctx.meta.emailAccountId as number; - if (!emailAccountId) throw new PermanentError('Missing emailAccountId in job meta'); - - const account = await getEmailAccount(emailAccountId); - if (!account) throw new PermanentError(`Email account ${emailAccountId} not found`); - - // Look up user email for openEmailDb - const user = await getUserById(account.userId); - if (!user) throw new PermanentError(`User ${account.userId} not found`); - - ctx.meta.account = account; - ctx.meta.userEmail = user.email; - - // Resolve IMAP credentials - if (account.authType === 'oauth') { - const userGoogle = await getUserIntegration(account.userId, 'google'); - const config = userGoogle?.config as Record | undefined; - if (!config?.refreshToken) { - throw new PermanentError('Google OAuth not configured — reconnect your Google account'); - } - - let accessToken = config.accessToken as string | undefined; - const expiresAt = config.expiresAt as number | undefined; - const tokenExpired = !expiresAt || expiresAt < Date.now() + 60_000; - - if (tokenExpired) { - console.log('[email-sync] Refreshing OAuth token'); - const refreshed = await refreshGoogleAccessToken(config.refreshToken as string); - accessToken = refreshed.accessToken; - - // Persist refreshed token - const serverGoogle = await getServerIntegration('google'); - await upsertUserIntegration({ - userId: account.userId, - provider: 'google', - serverIntegrationId: serverGoogle?.id, - config: { ...config, accessToken: refreshed.accessToken, expiresAt: refreshed.expiresAt }, - }); - } - - ctx.meta.imapAuth = { user: account.email, accessToken }; - } else { - const creds = account.credentials as Record; - const password = creds.password as string | undefined; - if (!password) throw new PermanentError('No password stored for this account'); - ctx.meta.imapAuth = { user: account.email, pass: password }; - } - - // Check if initial or incremental - const syncMeta = account.syncMeta as Record; - ctx.meta.isIncremental = !!syncMeta.last_sync_at; - ctx.meta.syncMeta = syncMeta; - - // Set status to syncing - await updateEmailAccountStatus(emailAccountId, 'syncing'); - - console.log(`[email-sync] ${ctx.meta.isIncremental ? 'Incremental' : 'Initial'} sync for ${account.email}`); - }, - }, { name: 'Sync emails', run: async (ctx) => { const { ImapFlow } = await import('imapflow'); - const account = ctx.meta.account as { id: number; email: string; imapHost: string; imapPort: number; imapSecure: boolean; provider: string }; - const imapAuth = ctx.meta.imapAuth as { user: string; pass?: string; accessToken?: string }; - const userEmail = ctx.meta.userEmail as string; - const syncMeta = ctx.meta.syncMeta as Record; + const meta = ctx.meta as unknown as EmailSyncMeta; + const { account, userEmail } = meta; - await ctx.updateProgress({ current: 0, total: 0, label: 'Connecting...' }); + // Resolve auth (refreshes OAuth token if needed) + const imapAuth = await resolveImapAuth(meta); - const client = new ImapFlow({ - host: account.imapHost, - port: account.imapPort, - secure: account.imapSecure, - auth: imapAuth, - logger: false, - socketTimeout: 30 * 60 * 1000, // 30 min — large mailboxes need time - }); + // Load syncMeta from DB (always fresh, not from job meta) + const freshAccount = await getEmailAccount(account.id); + if (!freshAccount) throw new PermanentError(`Email account ${account.id} not found`); + const syncMeta = (freshAccount.syncMeta ?? {}) as Record; + const isIncremental = !!syncMeta.last_sync_at; - // Prevent unhandled 'error' event from crashing the process - const connState = { error: null as Error | null }; - client.on('error', (err: Error) => { - console.log(`[email-sync] IMAP connection error: ${err.message}`); - connState.error = err; - }); + await updateEmailAccountStatus(account.id, 'syncing'); + console.log(`[email-sync] ${isIncremental ? 'Incremental' : 'Initial'} sync for ${account.email}`); - const db = openEmailDb(userEmail); + 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; + + const db = openEmailDb(userEmail); + + // 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 { - await client.connect(); - console.log('[email-sync] IMAP connected'); + 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)); - // Get all folders with status - const folders = await client.list({ statusQuery: { uidNext: true, uidValidity: true } }) as FolderInfo[]; - - // Load existing IDs for dedup - 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); - - // Filter to syncable folders and count total messages to fetch - 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; + // 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...' }); } - // Skip if no new messages - if (lastUid > 0 && uidNext <= lastUid + 1) continue; + const client = new ImapFlow({ + host: account.imapHost, + port: account.imapPort, + secure: account.imapSecure, + auth: imapAuth, + logger: false, + socketTimeout: 30 * 60 * 1000, + }); - foldersToSync.push({ folder, lastUid }); - } + const connState = { error: null as Error | null }; + client.on('error', (err: Error) => { + console.log(`[email-sync] IMAP connection error: ${err.message}`); + connState.error = err; + }); - console.log(`[email-sync] ${foldersToSync.length} folder(s) to sync`); - - for (let fi = 0; fi < foldersToSync.length; fi++) { - const { folder, lastUid } = foldersToSync[fi]!; - - // If connection is already dead, stop trying more folders - if (connState.error) 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}`); - connState.error = connState.error ?? new Error(errMsg); + 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] Skipping folder ${folder.path}: ${errMsg}`); - continue; - } - try { - const mailbox = client.mailbox; - if (!mailbox) continue; + console.log(`[email-sync] ${foldersToSync.length} folder(s) to sync`); - 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:*'; + let connectionLost = false; - try { - const fetchOpts: Record = { source: true, uid: true }; - if (account.provider === 'gmail') fetchOpts.labels = true; + for (let fi = 0; fi < foldersToSync.length; fi++) { + const { folder, lastUid } = foldersToSync[fi]!; - for await (const msg of client.fetch(range, fetchOpts, { uid: true })) { - if (msg.uid <= effectiveLastUid) continue; + if (connState.error) { connectionLost = true; break; } - maxUid = Math.max(maxUid, msg.uid); + console.log(`[email-sync] Syncing folder ${fi + 1}/${foldersToSync.length}: ${folder.path}`); - if (!msg.source) { - errors++; - continue; + 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; + } - const raw = msg.source.toString('utf-8'); - const id = messageIdToStableId(raw); - if (!id) { - errors++; - continue; - } + try { + const mailbox = client.mailbox; + if (!mailbox) continue; - if (existingIds.has(id)) { - skipped++; - 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:*'; - const labels = [label]; try { - upsertFromRawEml({ db, id, raw, integration: account.provider, emailAccount: account.email, labels }); - existingIds.add(id); - saved++; - } catch { - errors++; - } + const fetchOpts: Record = { source: true, uid: true }; + if (account.provider === 'gmail') fetchOpts.labels = true; - 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 }); + for await (const msg of client.fetch(range, fetchOpts, { uid: true })) { + if (msg.uid <= effectiveLastUid) continue; - // Save progress mid-folder so retries resume from here - if (maxUid > effectiveLastUid) { - syncMeta[`imap_lastuid:${folder.path}`] = String(maxUid); - await updateEmailAccountSyncMeta(account.id, syncMeta as Record); + 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); + await updateEmailAccountSyncMeta(account.id, syncMeta as Record); + } + } + } + } 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` }); + await updateEmailAccountSyncMeta(account.id, syncMeta as Record); + } finally { + lock.release(); } - } 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}`); - connState.error = connState.error ?? new Error(errMsg); - } else { - console.log(`[email-sync] Fetch error in ${folder.path}: ${errMsg}`); - errors++; - } + + if (connectionLost) break; } - syncMeta[`imap_uidvalidity:${folder.path}`] = uidValidity; - if (maxUid > effectiveLastUid) { - syncMeta[`imap_lastuid:${folder.path}`] = String(maxUid); + await client.logout().catch(() => {}); + + if (connectionLost) { + console.log(`[email-sync] Connection lost after saving ${saved} emails — will reconnect`); + reconnects++; + continue; } - 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` }); - - // Save sync meta after each folder - await updateEmailAccountSyncMeta(account.id, syncMeta as Record); - } finally { - lock.release(); + 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 connection died, throw to trigger retry (progress is already saved) - if (connState.error) { - console.log(`[email-sync] Connection lost after saving ${saved} emails — will retry remaining folders`); - throw new Error(`IMAP connection lost: ${connState.error.message}`); + if (!allDone) { + throw new Error(`IMAP sync incomplete after ${MAX_RECONNECTS} reconnection attempts`); } // Mark sync complete syncMeta.last_sync_at = new Date().toISOString(); await updateEmailAccountSyncMeta(account.id, syncMeta as Record); - - await client.logout().catch(() => {}); - } catch (err) { - await client.logout().catch(() => {}); - throw err; } finally { db.close(); } - console.log(`[email-sync] Done: saved ${saved}, skipped ${skipped}, errors ${errors}`); + 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` }); }, @@ -326,28 +353,26 @@ const emailSyncHandler: JobHandler = { { name: 'Finalize', run: async (ctx) => { - const account = ctx.meta.account as { id: number; userId: number; email: string }; - const saved = ctx.meta.saved as number; + const meta = ctx.meta as unknown as EmailSyncMeta; + const saved = meta.saved ?? 0; - // Update account status to synced - await updateEmailAccountStatus(account.id, 'synced'); + await updateEmailAccountStatus(meta.account.id, 'synced'); - // Auto-add /email to dock if (saved > 0) { try { - const paths = await getDockPaths(account.userId); + const paths = await getDockPaths(meta.account.userId); if (!paths) { const defaults = ['/', '/files', '/automation', '/projects', '/dashboards', '/chat']; - await setDockPaths(account.userId, [...defaults, '/email']); + await setDockPaths(meta.account.userId, [...defaults, '/email']); } else if (!paths.includes('/email')) { - await setDockPaths(account.userId, [...paths, '/email']); + await setDockPaths(meta.account.userId, [...paths, '/email']); } } catch { // Non-fatal } } - console.log(`[email-sync] ${account.email} status set to synced`); + console.log(`[email-sync] ${meta.account.email} status set to synced`); }, }, ], diff --git a/src/servers/sidecar/email-cron.ts b/src/servers/sidecar/email-cron.ts index 29213bee..1e6dd86f 100644 --- a/src/servers/sidecar/email-cron.ts +++ b/src/servers/sidecar/email-cron.ts @@ -1,4 +1,4 @@ -import { getAllSyncedAccounts, getUserById } from 'officerdb'; +import { getAllSyncedAccounts, getUserById, getUserIntegration } from 'officerdb'; import * as queueRunner from './queue-runner'; const INTERVAL_MS = 10 * 60 * 1000; // 10 minutes @@ -23,12 +23,43 @@ async function tick() { const user = await getUserById(account.userId); if (!user) continue; + // Resolve IMAP auth + const imapAuth: Record = { user: account.email }; + if (account.authType === 'oauth') { + const integration = await getUserIntegration(account.userId, 'google'); + const config = integration?.config as Record | undefined; + const accessToken = config?.accessToken as string | undefined; + if (!accessToken) { + console.log(`[email-cron] Skipping ${account.email}: no OAuth access token`); + continue; + } + imapAuth.accessToken = accessToken; + } else { + const creds = account.credentials as Record; + imapAuth.pass = creds.password; + } + try { await queueRunner.enqueue({ lane: 'email', type: 'email-sync', userId: user.email, - meta: { emailAccountId: account.id }, + meta: { + emailAccountId: account.id, + userEmail: user.email, + account: { + id: account.id, + userId: account.userId, + email: account.email, + imapHost: account.imapHost, + imapPort: account.imapPort, + imapSecure: account.imapSecure, + provider: account.provider, + authType: account.authType, + credentials: account.credentials, + }, + imapAuth, + }, }); console.log(`[email-cron] Enqueued incremental sync for ${account.email}`); } catch (err) { diff --git a/src/servers/sidecar/queue-runner.ts b/src/servers/sidecar/queue-runner.ts index c0b34755..44ddc064 100644 --- a/src/servers/sidecar/queue-runner.ts +++ b/src/servers/sidecar/queue-runner.ts @@ -73,6 +73,12 @@ async function resumeInterruptedJobs() { console.log(`[sidecar:queue] reset interrupted job ${job.id} back to queued`); lanesToKick.add(job.lane); } else if (job.status === 'queued') { + // Clear retry delay on restart — no reason to wait after a sidecar restart + if (job.retryAt) { + job.retryAt = undefined; + await writeJob(job); + console.log(`[sidecar:queue] cleared retry delay for job ${job.id}`); + } lanesToKick.add(job.lane); } } @@ -187,6 +193,7 @@ async function runJob(job: Job) { await writeJob(fresh); } catch (err) { const errorMessage = err instanceof Error ? err.message : String(err); + console.error(`[sidecar:queue] step "${step.name}" failed: ${errorMessage}`); step.status = 'failed'; step.error = errorMessage; step.completedAt = Date.now();