import type { Database } from 'bun:sqlite'; import { createHash } from 'node:crypto'; import { openEmailDb, upsertFromRawEml, setSyncMeta, getSyncMeta } from './store'; import { getUserByEmail, getUserIntegration, upsertUserIntegration, getServerIntegration, getDockPaths, setDockPaths } from 'officerdb'; import { updateEmailAccountStatus, getEmailAccount } from '../db/queries'; import { refreshGoogleAccessToken } from '@@/api/integrations/google-auth'; 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 {} // ── Credentials ── export type GmailCredentials = { email: string; userId: number; appPassword?: string; accessToken?: string; refreshToken?: string; expiresAt?: number; }; export async function loadGmailCredentials(userEmail: string): Promise { const dbUser = await getUserByEmail(userEmail); if (!dbUser) throw new PermanentError('User not found'); const userGoogle = await getUserIntegration(dbUser.id, 'google'); const config = userGoogle?.config as Record | undefined; const gmailEmail = (config?.email as string) ?? userEmail; // Prefer OAuth tokens, fall back to app password if (config?.accessToken && config?.refreshToken) { return { email: gmailEmail, userId: dbUser.id, accessToken: config.accessToken as string, refreshToken: config.refreshToken as string, expiresAt: config.expiresAt as number | undefined, appPassword: config.imapAppPassword as string | undefined, }; } if (!config?.imapAppPassword) { throw new PermanentError('Gmail App Password not configured — set it in Settings → Integrations'); } return { email: gmailEmail, userId: dbUser.id, appPassword: config.imapAppPassword as string }; } // ── 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); } // ── Gmail API label → our label mapping ── const GMAIL_API_LABEL_MAP: Record = { INBOX: 'inbox', SENT: 'sent', DRAFT: 'draft', STARRED: 'starred', IMPORTANT: 'important', TRASH: 'trash', SPAM: 'spam', CATEGORY_PROMOTIONS: 'promotions', CATEGORY_SOCIAL: 'social', CATEGORY_UPDATES: 'updates', CATEGORY_FORUMS: 'forums', }; const SKIP_LABELS = new Set(['TRASH', 'SPAM', 'DRAFT']); function gmailApiLabelsToLabels(labelIds: string[]): string[] { const labels: string[] = []; for (const id of labelIds) { const mapped = GMAIL_API_LABEL_MAP[id]; if (mapped) { labels.push(mapped); } else if (!id.startsWith('CATEGORY_') && id !== 'UNREAD' && id !== 'IMPORTANT') { // Custom label — use as-is (Gmail API returns label IDs, not names, for custom labels) labels.push(id.toLowerCase()); } } return labels; } function shouldSkipMessage(labelIds: string[]): boolean { return labelIds.some((id) => SKIP_LABELS.has(id)); } // ── Gmail API helpers ── const GMAIL_API = 'https://gmail.googleapis.com/gmail/v1/users/me'; export type GmailApiResult = { saved: number; skipped: number; errors: number }; async function gmailApiFetch(accessToken: string, path: string, params?: Record): Promise { const url = new URL(`${GMAIL_API}${path}`); if (params) { for (const [k, v] of Object.entries(params)) url.searchParams.set(k, v); } const res = await fetch(url, { headers: { Authorization: `Bearer ${accessToken}` } }); return res; } async function getGmailProfile(accessToken: string): Promise<{ historyId: string }> { const res = await gmailApiFetch(accessToken, '/profile'); if (!res.ok) throw new Error(`Gmail profile request failed: ${res.status} ${await res.text()}`); const data = (await res.json()) as { historyId: string }; return data; } type HistoryMessage = { id: string; threadId: string }; type HistoryEntry = { messagesAdded?: Array<{ message: HistoryMessage }> }; type HistoryResponse = { history?: HistoryEntry[]; historyId: string; nextPageToken?: string }; async function getHistoryChanges( accessToken: string, startHistoryId: string, ): Promise<{ messageIds: string[]; historyId: string }> { const messageIds = new Set(); let pageToken: string | undefined; let latestHistoryId = startHistoryId; while (true) { const params: Record = { startHistoryId, historyTypes: 'messageAdded', maxResults: '500', }; if (pageToken) params.pageToken = pageToken; const res = await gmailApiFetch(accessToken, '/history', params); if (res.status === 404 || res.status === 410) { // historyId too old or invalid — caller should fall back to date-based listing throw new HistoryExpiredError(); } if (!res.ok) throw new Error(`Gmail history request failed: ${res.status} ${await res.text()}`); const data = (await res.json()) as HistoryResponse; latestHistoryId = data.historyId; if (data.history) { for (const entry of data.history) { if (entry.messagesAdded) { for (const added of entry.messagesAdded) { messageIds.add(added.message.id); } } } } if (!data.nextPageToken) break; pageToken = data.nextPageToken; } return { messageIds: [...messageIds], historyId: latestHistoryId }; } class HistoryExpiredError extends Error { constructor() { super('Gmail historyId expired'); } } type MessageListResponse = { messages?: Array<{ id: string }>; nextPageToken?: string }; async function listMessagesSince(accessToken: string, sinceDate: string): Promise { const messageIds: string[] = []; let pageToken: string | undefined; while (true) { const params: Record = { q: `after:${sinceDate}`, maxResults: '500', }; if (pageToken) params.pageToken = pageToken; const res = await gmailApiFetch(accessToken, '/messages', params); if (!res.ok) throw new Error(`Gmail messages.list failed: ${res.status} ${await res.text()}`); const data = (await res.json()) as MessageListResponse; if (data.messages) { for (const msg of data.messages) messageIds.push(msg.id); } if (!data.nextPageToken) break; pageToken = data.nextPageToken; } return messageIds; } type GmailMessage = { id: string; labelIds: string[]; raw: string; historyId: string }; async function fetchRawMessage(accessToken: string, messageId: string): Promise { const res = await gmailApiFetch(accessToken, `/messages/${messageId}`, { format: 'raw' }); if (res.status === 404) return null; if (!res.ok) throw new Error(`Gmail messages.get failed: ${res.status} ${await res.text()}`); return (await res.json()) as GmailMessage; } // ── Gmail API sync (for resyncs) ── export type GmailApiSyncParams = { creds: GmailCredentials; db: Database; onProgress?: (saved: number, total: number) => void; }; export async function gmailApiSync({ creds, db, onProgress }: GmailApiSyncParams): Promise { if (!creds.accessToken) throw new PermanentError('OAuth access token required for Gmail API sync'); let saved = 0; let skipped = 0; let errors = 0; // 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); const storedHistoryId = getSyncMeta(db, 'gmail_history_id'); let messageIds: string[]; let newHistoryId: string; if (storedHistoryId) { // Try history-based sync first try { const result = await getHistoryChanges(creds.accessToken, storedHistoryId); messageIds = result.messageIds; newHistoryId = result.historyId; console.log(`[gmail-sync] History returned ${messageIds.length} new message(s)`); } catch (err) { if (err instanceof HistoryExpiredError) { // historyId too old — fall back to date-based listing const lastSyncDate = getSyncMeta(db, 'last_sync_date'); const sinceDate = lastSyncDate ?? new Date(Date.now() - 7 * 24 * 60 * 60 * 1000).toISOString().split('T')[0]!; console.log(`[gmail-sync] History expired, falling back to messages since ${sinceDate}`); messageIds = await listMessagesSince(creds.accessToken, sinceDate); const profile = await getGmailProfile(creds.accessToken); newHistoryId = profile.historyId; console.log(`[gmail-sync] Date query returned ${messageIds.length} message(s)`); } else { throw err; } } } else { // No historyId stored — use date-based listing const lastSyncDate = getSyncMeta(db, 'last_sync_date'); const sinceDate = lastSyncDate ?? new Date(Date.now() - 7 * 24 * 60 * 60 * 1000).toISOString().split('T')[0]!; console.log(`[gmail-sync] No historyId, listing messages since ${sinceDate}`); messageIds = await listMessagesSince(creds.accessToken, sinceDate); const profile = await getGmailProfile(creds.accessToken); newHistoryId = profile.historyId; console.log(`[gmail-sync] Date query returned ${messageIds.length} message(s)`); } // Fetch and store each new message for (let i = 0; i < messageIds.length; i++) { const gmailId = messageIds[i]!; try { const msg = await fetchRawMessage(creds.accessToken, gmailId); if (!msg) { skipped++; continue; } if (shouldSkipMessage(msg.labelIds ?? [])) { skipped++; continue; } // Gmail API returns base64url-encoded raw RFC822 const rawBytes = Buffer.from(msg.raw, 'base64url'); const raw = rawBytes.toString('utf-8'); const id = messageIdToStableId(raw); if (!id) { errors++; continue; } if (existingIds.has(id)) { skipped++; continue; } const labels = gmailApiLabelsToLabels(msg.labelIds ?? []); upsertFromRawEml({ db, id, raw, integration: 'gmail', emailAccount: creds.email, labels }); existingIds.add(id); saved++; if ((saved + skipped) % 50 === 0) { onProgress?.(saved, messageIds.length); } } catch (err) { console.log(`[gmail-sync] Error fetching message ${gmailId}: ${err instanceof Error ? err.message : err}`); errors++; } } // Persist new historyId setSyncMeta(db, 'gmail_history_id', newHistoryId); return { saved, skipped, errors }; } // ── IMAP sync (for first sync) ── // Folder → label mapping for IMAP const GMAIL_PREFIX_RE = /^\[(?:Gmail|Google Mail)\]\//; function normalizeGmailFolder(folder: string): string { return folder.replace(GMAIL_PREFIX_RE, ''); } const SUFFIX_LABEL_MAP: Record = { 'Sent Mail': 'sent', Drafts: 'draft', Starred: 'starred', Important: 'important', 'All Mail': 'archive', Trash: 'trash', Bin: 'trash', Spam: 'spam', }; function folderToLabel(folder: string): string | null { const suffix = normalizeGmailFolder(folder); if (suffix !== folder) { if (SUFFIX_LABEL_MAP[suffix]) return SUFFIX_LABEL_MAP[suffix]!; return suffix.toLowerCase(); } if (folder === 'INBOX') return 'inbox'; return folder.toLowerCase(); } const IMAP_LABEL_MAP: Record = { '\\Inbox': 'inbox', '\\Sent': 'sent', '\\Drafts': 'draft', '\\Starred': 'starred', '\\Important': 'important', '\\All': 'archive', '\\Trash': 'trash', '\\Junk': 'spam', }; function imapLabelsToLabels(gmailLabels: Set): string[] { const labels: string[] = []; for (const gl of gmailLabels) { const mapped = IMAP_LABEL_MAP[gl]; if (mapped) { labels.push(mapped); } else if (!gl.startsWith('\\')) { labels.push(gl.toLowerCase()); } } if (labels.length > 1 && labels.includes('archive')) { return labels.filter((l) => l !== 'archive'); } return labels; } const SKIP_SPECIAL_USE = new Set(['\\Trash', '\\Junk', '\\All']); function shouldSkipFolder(folder: { specialUse?: string; path: string; flags: Set }): boolean { if (folder.flags.has('\\Noselect') || folder.flags.has('\\NonExistent')) return true; if (folder.specialUse && SKIP_SPECIAL_USE.has(folder.specialUse)) return true; const norm = normalizeGmailFolder(folder.path); return norm === 'All Mail' || norm === 'Trash' || norm === 'Spam' || norm === 'Bin'; } type ImapSyncParams = { creds: GmailCredentials; db: Database; onProgress?: (fetched: number, folder: string) => void; }; type SyncResult = { saved: number; skipped: number; errors: number }; async function imapFullSync({ creds, db, onProgress }: ImapSyncParams): Promise { const { ImapFlow } = await import('imapflow'); const auth: { user: string; pass?: string; accessToken?: string } = { user: creds.email }; if (creds.appPassword) { auth.pass = creds.appPassword; } else if (creds.accessToken) { auth.accessToken = creds.accessToken; } else { throw new PermanentError('No authentication method available for IMAP'); } const client = new ImapFlow({ host: 'imap.gmail.com', port: 993, secure: true, auth, logger: false, }); let saved = 0; let skipped = 0; let errors = 0; try { await client.connect(); console.log('[gmail-sync] IMAP connected for full sync'); const folders = await client.list({ statusQuery: { uidNext: true, uidValidity: true } }); 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); const foldersToSync: typeof folders = []; for (const folder of folders) { if (shouldSkipFolder(folder)) continue; const folderPath = folder.path; const uidValidityKey = `imap_uidvalidity:${folderPath}`; const lastUidKey = `imap_lastuid:${folderPath}`; const storedUidValidity = getSyncMeta(db, uidValidityKey); const storedLastUid = getSyncMeta(db, lastUidKey); const uidValidity = folder.status?.uidValidity ? String(folder.status.uidValidity) : null; const uidNext = folder.status?.uidNext ?? 0; const lastUid = storedUidValidity && uidValidity === storedUidValidity && storedLastUid ? parseInt(storedLastUid, 10) : 0; if (uidValidity) setSyncMeta(db, uidValidityKey, uidValidity); if (lastUid > 0 && uidNext <= lastUid + 1) continue; foldersToSync.push(folder); } console.log(`[gmail-sync] ${foldersToSync.length} folder(s) to sync`); for (const folder of foldersToSync) { const folderPath = folder.path; const uidValidityKey = `imap_uidvalidity:${folderPath}`; const lastUidKey = `imap_lastuid:${folderPath}`; const storedUidValidity = getSyncMeta(db, uidValidityKey); const storedLastUid = getSyncMeta(db, lastUidKey); let lock; try { lock = await client.getMailboxLock(folderPath); } catch (err) { console.log(`[gmail-sync] Skipping folder ${folderPath}: ${err instanceof Error ? err.message : err}`); continue; } try { const mailbox = client.mailbox; if (!mailbox) continue; const uidValidity = String(mailbox.uidValidity); let lastUid = 0; if (storedUidValidity === uidValidity && storedLastUid) { lastUid = parseInt(storedLastUid, 10); } const range = lastUid > 0 ? `${lastUid + 1}:*` : '1:*'; let maxUid = lastUid; let folderFetched = 0; try { for await (const msg of client.fetch(range, { source: true, labels: true, uid: true }, { uid: true })) { if (msg.uid <= lastUid) continue; maxUid = Math.max(maxUid, msg.uid); folderFetched++; 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 folderLabel = folderToLabel(folderPath); const gmailLabels = msg.labels ? imapLabelsToLabels(msg.labels) : []; const labels = folderLabel ? [...new Set([folderLabel, ...gmailLabels])] : gmailLabels; try { upsertFromRawEml({ db, id, raw, integration: 'gmail', emailAccount: creds.email, labels }); existingIds.add(id); saved++; } catch { errors++; } } } catch (fetchErr) { const msg = fetchErr instanceof Error ? fetchErr.message : String(fetchErr); if (!msg.includes('Nothing to fetch')) { console.log(`[gmail-sync] Fetch error in ${folderPath}: ${msg}`); errors++; } } if (maxUid > lastUid) { setSyncMeta(db, lastUidKey, String(maxUid)); } setSyncMeta(db, uidValidityKey, uidValidity); if (folderFetched > 0) { console.log(`[gmail-sync] ${folderPath}: fetched ${folderFetched}, saved ${saved}, skipped ${skipped}`); onProgress?.(saved + skipped, folderPath); } } finally { lock.release(); } } } finally { await client.logout().catch(() => {}); } return { saved, skipped, errors }; } // ── Handler ── const gmailSyncHandler = { type: 'gmail-sync', retry: { delayMs: 15 * 60 * 1000, maxRetries: 10 }, steps: [ { name: 'Verify credentials', run: async (ctx: SyncCtx) => { const creds = await loadGmailCredentials(ctx.job.userId); const emailAccountId = (ctx.meta as Record).emailAccountId as number | undefined; const syncAccount = emailAccountId ? await getEmailAccount(emailAccountId) : undefined; // Determine sync mode — check SQLite first, fall back to PostgreSQL syncMeta const db = openEmailDb(ctx.job.userId, syncAccount?.email ?? ctx.job.userId); let lastSyncAt: string | null = null; try { lastSyncAt = getSyncMeta(db, 'last_sync_at'); if (!lastSyncAt && emailAccountId) { const pgAccount = await getEmailAccount(emailAccountId); const pgMeta = pgAccount?.syncMeta as Record | null; if (pgMeta?.last_sync_at) { lastSyncAt = pgMeta.last_sync_at; // Backfill SQLite so future checks don't need PostgreSQL setSyncMeta(db, 'last_sync_at', lastSyncAt); if (pgMeta.last_sync_date) setSyncMeta(db, 'last_sync_date', pgMeta.last_sync_date); } } ctx.meta.isFirstSync = !lastSyncAt; } finally { db.close(); } if (emailAccountId) { await updateEmailAccountStatus(emailAccountId, 'syncing'); ctx.meta.emailAccountId = emailAccountId; } // For API resyncs, refresh OAuth token if expired if (!ctx.meta.isFirstSync && creds.accessToken && creds.refreshToken) { const tokenExpired = !creds.expiresAt || creds.expiresAt < Date.now() + 60_000; if (tokenExpired) { console.log('[gmail-sync] Refreshing OAuth token'); try { const refreshed = await refreshGoogleAccessToken(creds.refreshToken); creds.accessToken = refreshed.accessToken; creds.expiresAt = refreshed.expiresAt; const userGoogle = await getUserIntegration(creds.userId, 'google'); const existingConfig = (userGoogle?.config as Record) ?? {}; const serverGoogle = await getServerIntegration('google'); await upsertUserIntegration({ userId: creds.userId, provider: 'google', serverIntegrationId: serverGoogle?.id, config: { ...existingConfig, accessToken: refreshed.accessToken, expiresAt: refreshed.expiresAt }, }); } catch (err) { throw new PermanentError(`OAuth refresh failed — reconnect Google in Settings: ${err}`); } } } if (!ctx.meta.isFirstSync && !creds.accessToken) { throw new PermanentError('OAuth not configured — connect Google in Settings → Integrations for resyncs'); } ctx.meta.creds = creds; console.log( `[gmail-sync] ${ctx.meta.isFirstSync ? 'First sync (IMAP)' : 'Resync (Gmail API)'} — ${creds.accessToken ? 'OAuth' : 'App Password'}`, ); }, }, { name: 'Sync emails', run: async (ctx: SyncCtx) => { const creds = ctx.meta.creds as GmailCredentials; const emailAccountId = (ctx.meta as Record).emailAccountId as number | undefined; const syncAccount = emailAccountId ? await getEmailAccount(emailAccountId) : undefined; const db = openEmailDb(ctx.job.userId, syncAccount?.email ?? ctx.job.userId); try { let result: SyncResult; if (ctx.meta.isFirstSync) { // First sync: IMAP with app password await ctx.updateProgress({ current: 0, total: 0, label: 'Connecting to Gmail (IMAP)...' }); result = await imapFullSync({ creds, db, onProgress: (fetched, folder) => { ctx.updateProgress({ current: fetched, total: 0, label: `Syncing ${folder}...` }); }, }); // Seed historyId for future API syncs if (creds.accessToken) { try { const profile = await getGmailProfile(creds.accessToken); setSyncMeta(db, 'gmail_history_id', profile.historyId); console.log(`[gmail-sync] Seeded historyId: ${profile.historyId}`); } catch (err) { console.log(`[gmail-sync] Could not seed historyId: ${err}`); } } } else { // Resync: Gmail API with OAuth await ctx.updateProgress({ current: 0, total: 0, label: 'Checking for new emails...' }); result = await gmailApiSync({ creds, db, onProgress: (saved, total) => { ctx.updateProgress({ current: saved, total, label: `Fetching new emails (${saved}/${total})...` }); }, }); } console.log(`[gmail-sync] Done: saved ${result.saved}, skipped ${result.skipped}, errors ${result.errors}`); setSyncMeta(db, 'last_sync_date', new Date().toISOString().split('T')[0]!); setSyncMeta(db, 'last_sync_at', new Date().toISOString()); ctx.meta.syncResult = result; await ctx.updateProgress({ current: 1, total: 1, label: `Done — ${result.saved} new emails` }); } finally { db.close(); } }, }, { name: 'Finalize', run: async (ctx: SyncCtx) => { const result = ctx.meta.syncResult as SyncResult | undefined; const emailAccountId = ctx.meta.emailAccountId as number | undefined; if (emailAccountId) { await updateEmailAccountStatus(emailAccountId, 'synced'); } if (result && result.saved > 0) { try { const dbUser = await getUserByEmail(ctx.job.userId); if (dbUser) { const paths = await getDockPaths(dbUser.id); if (!paths) { const defaults = ['/', '/files', '/automation', '/dashboards', '/chat']; await setDockPaths(dbUser.id, [...defaults, '/email']); } else if (!paths.includes('/email')) { await setDockPaths(dbUser.id, [...paths, '/email']); } } } catch { // Non-fatal } } }, }, ], }; /** The handler's steps, for the sidecar's own runner. */ export const gmailSyncSteps: SyncStep[] = gmailSyncHandler.steps as SyncStep[];