replace imap gmail sync with mbsync + maildir import, app password UI

- rewrite gmail-sync handler: mbsync downloads to local Maildir, then import to sqlite
- add app password field to google integration config and API
- gmail sync section independent from oauth in settings UI
- live mbsync progress streaming to job status
- recoverable failure email with instructions for overquota/auth errors
- sync meta persisted to job on failure for richer notifications

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

src/servers/api/integrations/integrations.ts

@@ -126,9 +126,28 @@ integrationsRouter.get('/google/status', async (ctx) => {
     connected: !!connConfig?.accessToken,
     email: connConfig?.email ?? null,
     picture: connConfig?.picture ?? null,
+    hasAppPassword: !!connConfig?.imapAppPassword,
   });
 });
 
+integrationsRouter.put('/google/app-password', async (ctx) => {
+  const user = ctx.get('user');
+  const body = ctx.get('body') as { appPassword?: string };
+  if (!body.appPassword) throw BAD_REQUEST('Missing appPassword');
+
+  const connection = await getUserIntegration(user.id, 'google');
+  const connConfig = (connection?.config as Record<string, unknown>) ?? {};
+
+  await upsertUserIntegration({
+    userId: user.id,
+    provider: 'google',
+    serverIntegrationId: connection?.serverIntegrationId ?? undefined,
+    config: { ...connConfig, imapAppPassword: body.appPassword },
+  });
+
+  return ctx.json({ ok: true });
+});
+
 integrationsRouter.delete('/google/connection', async (ctx) => {
   const user = ctx.get('user');
   const connection = await getUserIntegration(user.id, 'google');

src/servers/data-path.ts

@@ -93,6 +93,8 @@ export const getAttachmentsDir = (email: string, provider: 'claude' | 'opencode'
 
 export const getUserEmailDir = (email: string) => join(DATA_PATH, email, 'Gmail', 'emails');
 
+export const getMaildirPath = (email: string) => join(DATA_PATH, email, 'Gmail', 'Maildir');
+
 /** Derive a valid Linux username from a display username or email. */
 export const toShellUsername = (username: string, email: string): string => {
   const raw = username || email.split('@')[0]!;

src/servers/queue/engine.ts

@@ -81,11 +81,16 @@ function kickLane(lane: string) {
   processNextInLane(lane);
 }
 
+function scheduleRetry(lane: string, delayMs: number) {
+  setTimeout(() => kickLane(lane), delayMs);
+}
+
 async function processNextInLane(lane: string) {
   try {
     const jobs = await listAllJobs();
+    const now = Date.now();
     const next = jobs
-      .filter((j) => j.lane === lane && j.status === 'queued')
+      .filter((j) => j.lane === lane && j.status === 'queued' && (!j.retryAt || j.retryAt <= now))
       .sort((a, b) => a.createdAt - b.createdAt)[0];
 
     if (!next) {
@@ -98,7 +103,7 @@ async function processNextInLane(lane: string) {
     console.error(`[queue] Lane ${lane} processing error:`, err);
   } finally {
     const jobs = await listAllJobs();
-    const hasMore = jobs.some((j) => j.lane === lane && j.status === 'queued');
+    const hasMore = jobs.some((j) => j.lane === lane && j.status === 'queued' && (!j.retryAt || j.retryAt <= Date.now()));
     if (hasMore) {
       processNextInLane(lane);
     } else {
@@ -119,8 +124,10 @@ async function runJob(job: Job) {
 
   job.status = 'running';
   job.startedAt = Date.now();
+  job.retryAt = undefined;
   await writeJob(job);
-  console.log(`[queue] Running job ${job.id} (${job.type})`);
+  const isRetry = (job.retries ?? 0) > 0;
+  console.log(`[queue] ${isRetry ? 'Resuming' : 'Running'} job ${job.id} (${job.type})${isRetry ? ` retry ${job.retries}` : ''}`);
 
   const sharedMeta: Record<string, unknown> = { ...(job.meta ?? {}) };
 
@@ -134,6 +141,9 @@ async function runJob(job: Job) {
     const handlerStep = handler.steps[i]!;
     const step = fresh.steps[i]!;
 
+    // Skip already-completed steps on retry
+    if (step.status === 'completed') continue;
+
     fresh.currentStep = i;
     step.status = 'running';
     step.startedAt = Date.now();
@@ -171,9 +181,32 @@ async function runJob(job: Job) {
       step.status = 'failed';
       step.error = errorMessage;
       step.completedAt = Date.now();
+
+      // Check if handler supports retry
+      const retries = (fresh.retries ?? 0) + 1;
+      if (handler.retry && retries <= handler.retry.maxRetries) {
+        // Schedule retry: reset failed step to pending, re-queue
+        step.status = 'pending';
+        step.error = undefined;
+        step.startedAt = undefined;
+        step.completedAt = undefined;
+        step.progress = undefined;
+        fresh.status = 'queued';
+        fresh.error = undefined;
+        fresh.completedAt = undefined;
+        fresh.startedAt = undefined;
+        fresh.retries = retries;
+        fresh.retryAt = Date.now() + handler.retry.delayMs;
+        await writeJob(fresh);
+        console.log(`[queue] Job ${fresh.id} will retry (${retries}/${handler.retry.maxRetries}) in ${handler.retry.delayMs / 1000}s — failed at step "${step.name}": ${errorMessage}`);
+        scheduleRetry(fresh.lane, handler.retry.delayMs);
+        return;
+      }
+
       fresh.status = 'failed';
       fresh.error = `Step "${step.name}" failed: ${errorMessage}`;
       fresh.completedAt = Date.now();
+      fresh.meta = { ...fresh.meta, ...sharedMeta };
       await writeJob(fresh);
       console.error(`[queue] Job ${fresh.id} failed at step "${step.name}":`, errorMessage);
       if (fresh.notify !== false) await notifyFailure(fresh);

src/servers/queue/handlers/gmail-sync.ts

@@ -1,270 +1,311 @@
 import type { Database } from 'bun:sqlite';
-import { ImapFlow } from 'imapflow';
+import { createHash } from 'node:crypto';
+import { join } from 'node:path';
+import { readdir, readFile, writeFile, unlink, mkdir, stat } from 'node:fs/promises';
 import type { JobHandler } from '../types';
 import { registerHandler } from '../handler-registry';
-import { openEmailDb, upsertFromRawEml, getSyncMeta, setSyncMeta } from '../../api/email/email-db';
-import { getServerIntegration, getUserByEmail, getUserIntegration } from 'officerdb';
+import { openEmailDb, upsertFromRawEml, setSyncMeta } from '../../api/email/email-db';
+import { getUserByEmail, getUserIntegration } from 'officerdb';
+import { getMaildirPath } from '@@/data-path';
 
-type GoogleCredentials = {
-  accessToken: string;
-  refreshToken: string;
-  expiresAt: number;
-  clientId: string;
-  clientSecret: string;
-};
+// ── Credentials ──
 
-async function loadCredentials(email: string): Promise<GoogleCredentials> {
-  const serverGoogle = await getServerIntegration('google');
-  const serverConfig = serverGoogle?.config as Record<string, unknown> | undefined;
-  if (!serverConfig?.clientId || !serverConfig?.clientSecret) {
-    throw new Error('Google OAuth not configured — ask your admin to set up credentials');
-  }
+type ImapCredentials = { email: string; appPassword: string };
 
-  const dbUser = await getUserByEmail(email);
+async function loadImapCredentials(userEmail: string): Promise<ImapCredentials> {
+  const dbUser = await getUserByEmail(userEmail);
   if (!dbUser) throw new Error('User not found');
 
   const userGoogle = await getUserIntegration(dbUser.id, 'google');
-  const userConfig = userGoogle?.config as Record<string, unknown> | undefined;
-  if (!userConfig?.accessToken) {
-    throw new Error('Google account not connected — connect in Settings → Integrations');
+  const config = userGoogle?.config as Record<string, unknown> | undefined;
+  if (!config?.imapAppPassword) {
+    throw new Error('Gmail App Password not configured — set it in Settings → Integrations');
   }
 
-  return {
-    accessToken: userConfig.accessToken as string,
-    refreshToken: (userConfig.refreshToken as string) ?? '',
-    expiresAt: (userConfig.expiresAt as number) ?? 0,
-    clientId: serverConfig.clientId as string,
-    clientSecret: serverConfig.clientSecret as string,
-  };
+  const gmailEmail = (config.email as string) ?? userEmail;
+  return { email: gmailEmail, appPassword: config.imapAppPassword as string };
 }
 
-let cachedAccessToken: string | null = null;
-let cachedExpiresAt = 0;
+// ── mbsync config ──
 
-async function getValidAccessToken(creds: GoogleCredentials): Promise<string> {
-  if (cachedAccessToken && cachedExpiresAt > Date.now() + 5 * 60 * 1000) {
-    return cachedAccessToken;
-  }
+function buildMbsyncConfig(email: string, appPassword: string, maildirPath: string): string {
+  return `IMAPAccount gmail
+Host imap.gmail.com
+Port 993
+User ${email}
+Pass "${appPassword}"
+SSLType IMAPS
+AuthMechs LOGIN
 
-  if (creds.expiresAt > Date.now() + 5 * 60 * 1000) {
-    cachedAccessToken = creds.accessToken;
-    cachedExpiresAt = creds.expiresAt;
-    return creds.accessToken;
-  }
+IMAPStore gmail-remote
+Account gmail
 
-  if (!creds.refreshToken) throw new Error('Token expired and no refresh token available');
+MaildirStore gmail-local
+Path ${maildirPath}/
+Inbox ${maildirPath}/INBOX
+SubFolders Verbatim
 
-  const res = await fetch('https://oauth2.googleapis.com/token', {
-    method: 'POST',
-    headers: { 'Content-Type': 'application/x-www-form-urlencoded' },
-    body: new URLSearchParams({
-      client_id: creds.clientId,
-      client_secret: creds.clientSecret,
-      refresh_token: creds.refreshToken,
-      grant_type: 'refresh_token',
-    }),
-  });
-
-  if (!res.ok) {
-    const error = await res.text().catch(() => '');
-    throw new Error(`Token refresh failed (${res.status}): ${error}`);
-  }
-
-  const data = (await res.json()) as { access_token: string; expires_in?: number };
-  cachedAccessToken = data.access_token;
-  cachedExpiresAt = Date.now() + (data.expires_in ?? 3600) * 1000;
-  return data.access_token;
+Channel gmail
+Far :gmail-remote:
+Near :gmail-local:
+Patterns * ![Gmail]/Trash ![Gmail]/Spam
+Create Near
+Expunge None
+SyncState *
+`;
 }
 
-// ── IMAP label mapping ──
+// ── Folder → label mapping ──
 
-const SYSTEM_LABEL_MAP: Record<string, string> = {
-  '\\Inbox': 'inbox',
-  '\\Sent': 'sent',
-  '\\Trash': 'trash',
-  '\\Spam': 'spam',
-  '\\Draft': 'draft',
-  '\\Starred': 'starred',
-  '\\Important': 'important',
+const FOLDER_LABEL_MAP: Record<string, string> = {
+  INBOX: 'inbox',
+  '[Gmail]/Sent Mail': 'sent',
+  '[Gmail]/Drafts': 'draft',
+  '[Gmail]/Starred': 'starred',
+  '[Gmail]/Important': 'important',
 };
 
-function mapImapLabels(labels: Set<string> | undefined): string[] {
-  if (!labels) return [];
-  const mapped: string[] = [];
-  for (const label of labels) {
-    const system = SYSTEM_LABEL_MAP[label];
-    if (system) {
-      mapped.push(system);
-    } else {
-      mapped.push(label.toLowerCase());
-    }
-  }
-  return mapped;
+const SKIP_FOLDERS = new Set(['[Gmail]/All Mail', '[Gmail]/Trash', '[Gmail]/Spam']);
+
+function folderToLabel(folder: string): string | null {
+  if (SKIP_FOLDERS.has(folder)) return null;
+  if (FOLDER_LABEL_MAP[folder]) return FOLDER_LABEL_MAP[folder]!;
+  // Custom labels / other folders: lowercase the folder name
+  return folder.replace(/^\[Gmail\]\//, '').toLowerCase();
 }
 
-// ── IMAP sync ──
+// ── Stable ID from Message-Id header ──
 
-type ImapSyncResult = { saved: number; skipped: number; errors: number };
+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);
+}
 
-// Mailboxes to sync: All Mail has everything except Trash and Spam
-const SYNC_SPECIAL_USE = ['\\All', '\\Trash', '\\Junk'];
+// ── Maildir import ──
 
-async function syncViaImap(
-  accessToken: string,
+type ImportResult = { saved: number; skipped: number; errors: number };
+
+async function importMaildir(
+  maildirPath: string,
   emailAccount: string,
   db: Database,
-  since: Date | null,
   onProgress?: (saved: number, skipped: number) => void,
-): Promise<ImapSyncResult> {
-  const client = new ImapFlow({
-    host: 'imap.gmail.com',
-    port: 993,
-    secure: true,
-    auth: { user: emailAccount, accessToken },
-    logger: false,
-  });
-
-  try {
-    await client.connect();
-  } catch (err) {
-    console.error('[gmail-sync] IMAP connect failed:', err);
-    throw err;
-  }
-
+): Promise<ImportResult> {
   let saved = 0;
   let skipped = 0;
   let errors = 0;
 
-  // Load existing IDs for dedup (once, shared across mailboxes)
+  // Load existing IDs for fast dedup
   const existingIds = new Set<string>();
   const rows = db.query('SELECT id FROM emails').all() as Array<{ id: string }>;
   for (const row of rows) existingIds.add(row.id);
 
+  // First pass: collect all message files with their folders to build label map
+  const messageIdLabels = new Map<string, Set<string>>();
+  const messageFiles = new Map<string, string>(); // id → first file path
+
+  let folders: string[];
   try {
-    // Find mailboxes by specialUse flag (locale-independent)
-    const allMailboxes = await client.list();
-    const toSync: Array<{ path: string; specialUse: string }> = [];
-    for (const mailbox of allMailboxes) {
-      if (mailbox.specialUse && SYNC_SPECIAL_USE.includes(mailbox.specialUse)) {
-        toSync.push({ path: mailbox.path, specialUse: mailbox.specialUse });
-      }
-    }
+    folders = await readdir(maildirPath);
+  } catch {
+    console.log('[gmail-sync] No Maildir folders found');
+    return { saved, skipped, errors };
+  }
 
-    if (toSync.length === 0) {
-      console.error('[gmail-sync] No mailboxes found to sync');
-      return { saved, skipped, errors };
-    }
+  for (const folder of folders) {
+    const label = folderToLabel(folder);
+    if (label === null) continue;
 
-    for (const mailbox of toSync) {
-      console.log(`[gmail-sync] Opening ${mailbox.path} (${mailbox.specialUse})...`);
-      const lock = await client.getMailboxLock(mailbox.path);
+    for (const subdir of ['cur', 'new']) {
+      const dirPath = join(maildirPath, folder, subdir);
+      let files: string[];
       try {
-        const searchCriteria = since ? { since } : { all: true };
-        const uids = await client.search(searchCriteria, { uid: true });
+        files = await readdir(dirPath);
+      } catch {
+        continue;
+      }
 
-        if (!uids || uids.length === 0) {
-          console.log(`[gmail-sync] ${mailbox.path}: no messages`);
-          continue;
-        }
-
-        console.log(`[gmail-sync] ${mailbox.path}: ${uids.length} messages`);
-
-        const uidRange = uids.join(',');
-        const messages = client.fetch(uidRange, {
-          source: true,
-          labels: true,
-        }, { uid: true });
-
-        for await (const msg of messages) {
-          try {
-            if (!msg.emailId || !msg.source) continue;
-
-            const gmailId = BigInt(msg.emailId).toString(16);
-
-            if (existingIds.has(gmailId)) {
-              skipped++;
-              if ((saved + skipped) % 500 === 0) onProgress?.(saved, skipped);
-              continue;
-            }
-
-            const rawEmail = msg.source.toString('utf-8');
-            const labels = mapImapLabels(msg.labels);
-
-            upsertFromRawEml({ db, id: gmailId, raw: rawEmail, integration: 'gmail', emailAccount, labels });
-            existingIds.add(gmailId);
-            saved++;
-
-            if ((saved + skipped) % 100 === 0) {
-              console.log(`[gmail-sync] Progress: saved ${saved}, skipped ${skipped}, errors ${errors}`);
-              onProgress?.(saved, skipped);
-            }
-          } catch {
+      for (const file of files) {
+        const filePath = join(dirPath, file);
+        try {
+          const raw = await readFile(filePath, 'utf-8');
+          const id = messageIdToStableId(raw);
+          if (!id) {
             errors++;
+            continue;
           }
+
+          // Track labels
+          const labels = messageIdLabels.get(id) ?? new Set<string>();
+          labels.add(label);
+          messageIdLabels.set(id, labels);
+
+          // Keep first file path for importing
+          if (!messageFiles.has(id)) {
+            messageFiles.set(id, filePath);
+          }
+        } catch {
+          errors++;
         }
-      } finally {
-        lock.release();
       }
     }
-  } finally {
-    await client.logout();
+  }
+
+  // Second pass: import messages that aren't already in DB
+  for (const [id, filePath] of messageFiles) {
+    if (existingIds.has(id)) {
+      skipped++;
+      if ((saved + skipped) % 500 === 0) onProgress?.(saved, skipped);
+      continue;
+    }
+
+    try {
+      const raw = await readFile(filePath, 'utf-8');
+      const labels = Array.from(messageIdLabels.get(id) ?? []);
+
+      upsertFromRawEml({ db, id, raw, integration: 'gmail', emailAccount, labels });
+      existingIds.add(id);
+      saved++;
+
+      if ((saved + skipped) % 100 === 0) {
+        console.log(`[gmail-sync] Import progress: saved ${saved}, skipped ${skipped}, errors ${errors}`);
+        onProgress?.(saved, skipped);
+      }
+    } catch {
+      errors++;
+    }
   }
 
   return { saved, skipped, errors };
 }
 
+// ── Maildir stats ──
+
+async function countMaildirFiles(maildirPath: string): Promise<number> {
+  let count = 0;
+  let folders: string[];
+  try {
+    folders = await readdir(maildirPath);
+  } catch {
+    return 0;
+  }
+  for (const folder of folders) {
+    for (const subdir of ['cur', 'new']) {
+      try {
+        const files = await readdir(join(maildirPath, folder, subdir));
+        count += files.length;
+      } catch {
+        continue;
+      }
+    }
+  }
+  return count;
+}
+
 // ── Handler ──
 
 const gmailSyncHandler: JobHandler = {
   type: 'gmail-sync',
+  retry: { delayMs: 15 * 60 * 1000, maxRetries: 10 },
   steps: [
     {
-      name: 'Verify connection',
+      name: 'Verify credentials',
       run: async (ctx) => {
-        const creds = await loadCredentials(ctx.job.userId);
-        const token = await getValidAccessToken(creds);
-        ctx.meta.accessToken = token;
+        const creds = await loadImapCredentials(ctx.job.userId);
+        ctx.meta.email = creds.email;
+        ctx.meta.appPassword = creds.appPassword;
       },
     },
     {
-      name: 'Sync emails',
+      name: 'Sync via mbsync',
       run: async (ctx) => {
-        const token = ctx.meta.accessToken as string;
-        const year = ctx.meta.year as number | undefined;
+        const email = ctx.meta.email as string;
+        const appPassword = ctx.meta.appPassword as string;
+        const maildirPath = getMaildirPath(ctx.job.userId);
+
+        // Ensure Maildir root exists
+        await mkdir(maildirPath, { recursive: true });
+
+        // Write temp config
+        const configPath = join(maildirPath, '.mbsyncrc');
+        const config = buildMbsyncConfig(email, appPassword, maildirPath);
+        await writeFile(configPath, config, { mode: 0o600 });
+
+        await ctx.updateProgress({ current: 0, total: 0, label: 'Running mbsync...' });
+
+        try {
+          const proc = Bun.spawn(['mbsync', '-c', configPath, '-a', '-V'], {
+            stdout: 'pipe',
+            stderr: 'pipe',
+          });
+
+          // Stream stderr for live progress
+          let lastLine = '';
+          let stderrBuf = '';
+          const reader = proc.stderr.getReader();
+          const decoder = new TextDecoder();
+          const readLoop = (async () => {
+            while (true) {
+              const { done, value } = await reader.read();
+              if (done) break;
+              const chunk = decoder.decode(value, { stream: true });
+              stderrBuf += chunk;
+              const lines = chunk.split('\n');
+              for (const line of lines) {
+                const trimmed = line.trim();
+                if (!trimmed) continue;
+                lastLine = trimmed;
+                console.log(`[gmail-sync] mbsync: ${trimmed}`);
+              }
+              if (lastLine) {
+                ctx.updateProgress({ current: 0, total: 0, label: `mbsync: ${lastLine.slice(0, 80)}` });
+              }
+            }
+          })();
+
+          const exitCode = await proc.exited;
+          await readLoop;
+
+          if (exitCode !== 0) {
+            console.error(`[gmail-sync] mbsync failed`);
+            const isOverquota = stderrBuf.includes('OVERQUOTA');
+            const isAuthFail = stderrBuf.includes('AUTHENTICATIONFAILED') || stderrBuf.includes('Invalid credentials');
+            if (isOverquota || isAuthFail) {
+              const emailCount = await countMaildirFiles(maildirPath);
+              ctx.meta.gmailSyncRecoverable = true;
+              ctx.meta.gmailSyncEmailCount = emailCount;
+              ctx.meta.gmailSyncIsAuthFail = isAuthFail;
+            }
+            throw new Error(`mbsync exited with code ${exitCode}: ${stderrBuf.slice(-500)}`);
+          }
+
+          console.log('[gmail-sync] mbsync completed successfully');
+        } finally {
+          // Always clean up config (contains password)
+          await unlink(configPath).catch(() => {});
+        }
+      },
+    },
+    {
+      name: 'Import to database',
+      run: async (ctx) => {
+        const emailAccount = ctx.meta.email as string;
+        const maildirPath = getMaildirPath(ctx.job.userId);
+
+        await ctx.updateProgress({ current: 0, total: 0, label: 'Importing emails...' });
 
         const db = openEmailDb(ctx.job.userId);
         try {
-          // Bootstrap sync_meta from existing emails if DB was imported without metadata
-          if (!getSyncMeta(db, 'last_sync_date')) {
-            const newest = db.query('SELECT date FROM emails ORDER BY date DESC LIMIT 1').get() as { date: string } | null;
-            if (newest?.date) {
-              console.log(`[gmail-sync] Bootstrapping last_sync_date from existing DB: ${newest.date}`);
-              setSyncMeta(db, 'last_sync_date', newest.date);
-            }
-          }
-
-          // Determine since date
-          let since: Date | null = null;
-          if (year) {
-            since = new Date(year, 0, 1);
-          } else {
-            const lastSyncDate = getSyncMeta(db, 'last_sync_date');
-            if (lastSyncDate) {
-              since = new Date(lastSyncDate);
-              console.log(`[gmail-sync] Syncing since ${since.toISOString()}`);
-            }
-          }
-
-          await ctx.updateProgress({ current: 0, total: 0, label: 'Connecting via IMAP...' });
-
-          const result = await syncViaImap(token, ctx.job.userId, db, since, (saved, skipped) => {
-            const label = year
-              ? `${year} — Saved ${saved.toLocaleString()}, skipped ${skipped.toLocaleString()}`
-              : `Saved ${saved.toLocaleString()}, skipped ${skipped.toLocaleString()}`;
-            ctx.updateProgress({ current: saved + skipped, total: 0, label });
+          const result = await importMaildir(maildirPath, emailAccount, db, (saved, skipped) => {
+            ctx.updateProgress({
+              current: saved + skipped,
+              total: 0,
+              label: `Saved ${saved.toLocaleString()}, skipped ${skipped.toLocaleString()}`,
+            });
           });
 
-          console.log(`[gmail-sync] Done: saved ${result.saved}, skipped ${result.skipped}, errors ${result.errors}`);
+          console.log(`[gmail-sync] Import 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());