replace gmail resync with Gmail REST API, remove mbsync dependency

First sync still uses IMAP with app password. Subsequent syncs use
Gmail API history.list + messages.get with OAuth for faster, more
reliable incremental sync. Dispatch gmail-sync handler for gmail
accounts instead of generic email-sync. Show sync button for synced
accounts.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-12 07:24:44 +00:00
co-authored by Claude Opus 4.6
parent ee1b816d66
commit 24ac7ac796
5 changed files with 382 additions and 489 deletions
-33
View File
@@ -1,33 +0,0 @@
import { join } from 'node:path';
import { unlinkSync } from 'node:fs';
import { openEmailDb } from '../src/servers/api/email/email-db';
import { importMaildir } from '../src/servers/queue/handlers/gmail-sync';
const DATA_PATH = process.env.DATA_PATH ?? join(process.cwd(), 'data');
const email = process.argv[2];
if (!email) {
console.error('Usage: bun scripts/reimport-gmail.ts <email>');
process.exit(1);
}
const maildirPath = join(DATA_PATH, email, 'Gmail', 'Maildir');
const dbPath = join(DATA_PATH, email, 'emails.db');
// Delete existing DB for a fresh import
try {
unlinkSync(dbPath);
console.log(`Deleted ${dbPath}`);
} catch {
console.log('No existing DB to delete');
}
const db = openEmailDb(email);
console.log(`Importing from ${maildirPath}...`);
const result = await importMaildir({ maildirPath, emailAccount: email, db, onProgress: (saved, skipped) => {
process.stdout.write(`\r saved ${saved}, skipped ${skipped}`);
}});
console.log(`\nDone: saved ${result.saved}, skipped ${result.skipped}, errors ${result.errors}`);
db.close();
@@ -210,7 +210,7 @@ export const EmailAccounts = () => {
{account.status === 'queued' && (
<span className="text-xs text-duck-dark/50 dark:text-foreground/50">Waiting...</span>
)}
{account.status === 'connected' && (
{(account.status === 'connected' || account.status === 'synced') && (
<Button
type="button"
variant="outline"
@@ -224,7 +224,7 @@ export const EmailAccounts = () => {
) : (
<RefreshCw className="h-3 w-3" />
)}
Initial Sync
Sync
</Button>
)}
<button
+4 -2
View File
@@ -47,7 +47,7 @@ accountsRouter.get('/', async (ctx) => {
const jobs = await listAllJobs();
activeJobAccountIds = new Set(
jobs
.filter((j) => j.type === 'email-sync' && (j.status === 'queued' || j.status === 'running'))
.filter((j) => (j.type === 'email-sync' || j.type === 'gmail-sync') && (j.status === 'queued' || j.status === 'running'))
.map((j) => (j.meta as Record<string, unknown> | undefined)?.emailAccountId as number)
.filter(Boolean),
);
@@ -148,9 +148,11 @@ accountsRouter.post('/:id/sync', async (ctx) => {
// Set status immediately so the UI reflects the queued state
await updateEmailAccountStatus(id, 'queued');
const jobType = account.provider === 'gmail' ? 'gmail-sync' : 'email-sync';
const job = await enqueueJob({
lane: 'email',
type: 'email-sync',
type: jobType,
userId: user.email,
meta: {
emailAccountId: id,
-1
View File
@@ -69,7 +69,6 @@ export const getAttachmentsDir = (email: string, sessionId: string) => join(DATA
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 => {
+376 -451
View File
@@ -1,7 +1,5 @@
import type { Database } from 'bun:sqlite';
import { createHash } from 'node:crypto';
import { join } from 'node:path';
import { readdir, readFile, writeFile, unlink, mkdir, stat } from 'node:fs/promises';
import { type JobHandler, PermanentError } from '../types';
import { registerHandler } from '../handler-registry';
import { openEmailDb, upsertFromRawEml, setSyncMeta, getSyncMeta } from '../../api/email/email-db';
@@ -12,8 +10,9 @@ import {
getServerIntegration,
getDockPaths,
setDockPaths,
updateEmailAccountStatus,
getEmailAccount,
} from 'officerdb';
import { getMaildirPath } from '@@/data-path';
import { refreshGoogleAccessToken } from '@@/api/integrations/google-auth';
// ── Credentials ──
@@ -55,40 +54,281 @@ async function loadGmailCredentials(userEmail: string): Promise<GmailCredentials
return { email: gmailEmail, userId: dbUser.id, appPassword: config.imapAppPassword as string };
}
// refreshGoogleAccessToken is imported from @@/api/integrations/google-auth
// ── Stable ID from Message-Id header ──
// ── mbsync config ──
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
IMAPStore gmail-remote
Account gmail
MaildirStore gmail-local
Path ${maildirPath}/
Inbox ${maildirPath}/INBOX
SubFolders Verbatim
Channel gmail
Far :gmail-remote:
Near :gmail-local:
Patterns * ![Gmail]/Trash ![Gmail]/Spam ![Gmail]/Bin !"[Gmail]/All Mail" ![Google Mail]/Trash ![Google Mail]/Spam ![Google Mail]/Bin !"[Google Mail]/All Mail"
Create Near
Expunge None
SyncState *
`;
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);
}
// ── Folder → label mapping ──
// ── Gmail API label → our label mapping ──
// Strips [Gmail]/ or [Google Mail]/ prefix, returns the suffix (e.g. "Sent Mail")
const GMAIL_API_LABEL_MAP: Record<string, string> = {
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';
type GmailApiResult = { saved: number; skipped: number; errors: number };
async function gmailApiFetch(accessToken: string, path: string, params?: Record<string, string>): Promise<Response> {
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<string>();
let pageToken: string | undefined;
let latestHistoryId = startHistoryId;
while (true) {
const params: Record<string, string> = {
startHistoryId,
historyTypes: 'messageAdded',
maxResults: '500',
};
if (pageToken) params.pageToken = pageToken;
const res = await gmailApiFetch(accessToken, '/history', params);
if (res.status === 404) {
// No history — nothing new
return { messageIds: [], historyId: latestHistoryId };
}
if (res.status === 410) {
// historyId too old — 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<string[]> {
const messageIds: string[] = [];
let pageToken: string | undefined;
while (true) {
const params: Record<string, string> = {
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<GmailMessage | null> {
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) ──
type GmailApiSyncParams = {
creds: GmailCredentials;
db: Database;
onProgress?: (saved: number, total: number) => void;
};
async function gmailApiSync({ creds, db, onProgress }: GmailApiSyncParams): Promise<GmailApiResult> {
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<string>();
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 {
@@ -108,195 +348,15 @@ const SUFFIX_LABEL_MAP: Record<string, string> = {
function folderToLabel(folder: string): string | null {
const suffix = normalizeGmailFolder(folder);
// If it had a Gmail prefix, check against known suffixes
if (suffix !== folder) {
if (SUFFIX_LABEL_MAP[suffix]) return SUFFIX_LABEL_MAP[suffix]!;
return suffix.toLowerCase();
}
// Non-Gmail folders (INBOX, custom labels)
if (folder === 'INBOX') return 'inbox';
return folder.toLowerCase();
}
// ── 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);
}
// ── Maildir import ──
type ImportResult = { saved: number; skipped: number; errors: number };
type ImportMaildirParams = {
maildirPath: string;
emailAccount: string;
db: Database;
lastSyncAt?: string | null;
onProgress?: (saved: number, skipped: number) => void;
};
export async function importMaildir({
maildirPath,
emailAccount,
db,
lastSyncAt,
onProgress,
}: ImportMaildirParams): Promise<ImportResult> {
let saved = 0;
let skipped = 0;
let errors = 0;
// For incremental syncs, skip files older than last sync (with 60s buffer for clock skew)
const mtimeCutoff = lastSyncAt ? new Date(lastSyncAt).getTime() - 60_000 : 0;
const isIncremental = mtimeCutoff > 0;
// 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);
if (isIncremental) {
console.log(
`[gmail-sync] Incremental import — only reading files newer than ${new Date(mtimeCutoff).toISOString()}`,
);
}
// 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
// Discover Maildir folders — handles nested [Gmail]/ and [Google Mail]/ structures
let topEntries: string[];
try {
topEntries = await readdir(maildirPath);
} catch {
console.log('[gmail-sync] No Maildir folders found');
return { saved, skipped, errors };
}
const folders: string[] = [];
for (const entry of topEntries) {
if (entry === '[Gmail]' || entry === '[Google Mail]') {
// This is [Gmail] or [Google Mail] — subfolders are one level deeper
try {
const subs = await readdir(join(maildirPath, entry));
for (const sub of subs) folders.push(`${entry}/${sub}`);
} catch {
// empty
}
} else {
folders.push(entry);
}
}
// Process All Mail last so specific-folder labels take priority
const isAllMail = (f: string) => normalizeGmailFolder(f) === 'All Mail' && GMAIL_PREFIX_RE.test(f);
folders.sort((a, b) => (isAllMail(a) ? 1 : 0) - (isAllMail(b) ? 1 : 0));
for (const folder of folders) {
const label = folderToLabel(folder);
if (label === null) continue;
for (const subdir of ['cur', 'new']) {
const dirPath = join(maildirPath, folder, subdir);
let files: string[];
try {
files = await readdir(dirPath);
} catch {
continue;
}
for (const file of files) {
const filePath = join(dirPath, file);
try {
// Skip files older than last sync for incremental imports
if (isIncremental) {
const fileStat = await stat(filePath);
if (fileStat.mtimeMs < mtimeCutoff) continue;
}
const raw = await readFile(filePath, 'utf-8');
const id = messageIdToStableId(raw);
if (!id) {
errors++;
continue;
}
// Track labels — only add "archive" for messages not seen in any specific folder
const labels = messageIdLabels.get(id) ?? new Set<string>();
if (label !== 'archive' || labels.size === 0) {
labels.add(label);
}
messageIdLabels.set(id, labels);
// Keep first file path for importing
if (!messageFiles.has(id)) {
messageFiles.set(id, filePath);
}
} catch {
errors++;
}
}
}
}
// 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;
}
// ── Gmail IMAP label → our label mapping ──
const GMAIL_LABEL_MAP: Record<string, string> = {
const IMAP_LABEL_MAP: Record<string, string> = {
'\\Inbox': 'inbox',
'\\Sent': 'sent',
'\\Drafts': 'draft',
@@ -307,26 +367,22 @@ const GMAIL_LABEL_MAP: Record<string, string> = {
'\\Junk': 'spam',
};
function gmailLabelsToLabels(gmailLabels: Set<string>): string[] {
function imapLabelsToLabels(gmailLabels: Set<string>): string[] {
const labels: string[] = [];
for (const gl of gmailLabels) {
const mapped = GMAIL_LABEL_MAP[gl];
const mapped = IMAP_LABEL_MAP[gl];
if (mapped) {
labels.push(mapped);
} else if (!gl.startsWith('\\')) {
// Custom label — lowercase it
labels.push(gl.toLowerCase());
}
}
// If only "archive" and no specific folder, keep it; otherwise drop "archive"
if (labels.length > 1 && labels.includes('archive')) {
return labels.filter((l) => l !== 'archive');
}
return labels;
}
// ── IMAP special-use folders to skip ──
const SKIP_SPECIAL_USE = new Set(['\\Trash', '\\Junk', '\\All']);
function shouldSkipFolder(folder: { specialUse?: string; path: string; flags: Set<string> }): boolean {
@@ -336,25 +392,22 @@ function shouldSkipFolder(folder: { specialUse?: string; path: string; flags: Se
return norm === 'All Mail' || norm === 'Trash' || norm === 'Spam' || norm === 'Bin';
}
// ── Incremental IMAP sync ──
type IncrementalSyncParams = {
type ImapSyncParams = {
creds: GmailCredentials;
db: Database;
onProgress?: (fetched: number, folder: string) => void;
};
type IncrementalSyncResult = { saved: number; skipped: number; errors: number };
type SyncResult = { saved: number; skipped: number; errors: number };
async function incrementalImapSync({ creds, db, onProgress }: IncrementalSyncParams): Promise<IncrementalSyncResult> {
async function imapFullSync({ creds, db, onProgress }: ImapSyncParams): Promise<SyncResult> {
const { ImapFlow } = await import('imapflow');
// Determine auth method: prefer OAuth, fall back to app password
const auth: { user: string; pass?: string; accessToken?: string } = { user: creds.email };
if (creds.accessToken) {
auth.accessToken = creds.accessToken;
} else if (creds.appPassword) {
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');
}
@@ -373,17 +426,14 @@ async function incrementalImapSync({ creds, db, onProgress }: IncrementalSyncPar
try {
await client.connect();
console.log('[gmail-sync] Incremental IMAP connected');
console.log('[gmail-sync] IMAP connected for full sync');
// Get all folders with status in a single LIST command
const folders = await client.list({ statusQuery: { uidNext: true, uidValidity: true } });
// Load existing IDs for 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);
// Filter to only folders with new messages
const foldersToSync: typeof folders = [];
for (const folder of folders) {
if (shouldSkipFolder(folder)) continue;
@@ -400,17 +450,12 @@ async function incrementalImapSync({ creds, db, onProgress }: IncrementalSyncPar
storedUidValidity && uidValidity === storedUidValidity && storedLastUid ? parseInt(storedLastUid, 10) : 0;
if (uidValidity) setSyncMeta(db, uidValidityKey, uidValidity);
if (lastUid > 0 && uidNext <= lastUid + 1) continue; // no new messages
if (storedUidValidity && uidValidity && storedUidValidity !== uidValidity) {
console.log(`[gmail-sync] UIDVALIDITY changed for ${folderPath} — will re-scan`);
}
if (lastUid > 0 && uidNext <= lastUid + 1) continue;
foldersToSync.push(folder);
}
console.log(`[gmail-sync] ${foldersToSync.length} folder(s) with new messages`);
console.log(`[gmail-sync] ${foldersToSync.length} folder(s) to sync`);
for (const folder of foldersToSync) {
const folderPath = folder.path;
@@ -419,7 +464,6 @@ async function incrementalImapSync({ creds, db, onProgress }: IncrementalSyncPar
const storedUidValidity = getSyncMeta(db, uidValidityKey);
const storedLastUid = getSyncMeta(db, lastUidKey);
// Open folder and fetch new messages
let lock;
try {
lock = await client.getMailboxLock(folderPath);
@@ -466,10 +510,8 @@ async function incrementalImapSync({ creds, db, onProgress }: IncrementalSyncPar
continue;
}
// Gmail X-GM-EXT-1 labels don't include folder membership (e.g. \Inbox),
// so always use the folder we're fetching from as the base label
const folderLabel = folderToLabel(folderPath);
const gmailLabels = msg.labels ? gmailLabelsToLabels(msg.labels) : [];
const gmailLabels = msg.labels ? imapLabelsToLabels(msg.labels) : [];
const labels = folderLabel ? [...new Set([folderLabel, ...gmailLabels])] : gmailLabels;
try {
@@ -518,27 +560,43 @@ const gmailSyncHandler: JobHandler = {
name: 'Verify credentials',
run: async (ctx) => {
const creds = await loadGmailCredentials(ctx.job.userId);
const emailAccountId = (ctx.meta as Record<string, unknown>).emailAccountId as number | undefined;
// Determine sync mode: incremental if we have a previous sync
// Determine sync mode — check SQLite first, fall back to PostgreSQL syncMeta
const db = openEmailDb(ctx.job.userId);
let lastSyncAt: string | null = null;
try {
const lastSyncAt = getSyncMeta(db, 'last_sync_at');
ctx.meta.isIncremental = !!lastSyncAt;
lastSyncAt = getSyncMeta(db, 'last_sync_at');
if (!lastSyncAt && emailAccountId) {
const pgAccount = await getEmailAccount(emailAccountId);
const pgMeta = pgAccount?.syncMeta as Record<string, string> | 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();
}
// For incremental sync with OAuth, refresh token if expired
if (ctx.meta.isIncremental && creds.accessToken && creds.refreshToken) {
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 for incremental sync');
console.log('[gmail-sync] Refreshing OAuth token');
try {
const refreshed = await refreshGoogleAccessToken(creds.refreshToken);
creds.accessToken = refreshed.accessToken;
creds.expiresAt = refreshed.expiresAt;
// Persist refreshed token
const userGoogle = await getUserIntegration(creds.userId, 'google');
const existingConfig = (userGoogle?.config as Record<string, unknown>) ?? {};
const serverGoogle = await getServerIntegration('google');
@@ -549,39 +607,35 @@ const gmailSyncHandler: JobHandler = {
config: { ...existingConfig, accessToken: refreshed.accessToken, expiresAt: refreshed.expiresAt },
});
} catch (err) {
console.log(`[gmail-sync] OAuth refresh failed, will fall back to app password: ${err}`);
// Clear OAuth so we fall back to app password
creds.accessToken = undefined;
creds.refreshToken = undefined;
throw new PermanentError(`OAuth refresh failed — reconnect Google in Settings: ${err}`);
}
}
}
ctx.meta.creds = creds;
ctx.meta.email = creds.email;
ctx.meta.appPassword = creds.appPassword;
if (ctx.meta.isIncremental) {
console.log(`[gmail-sync] Incremental sync mode (${creds.accessToken ? 'OAuth' : 'App Password'})`);
} else {
console.log('[gmail-sync] Full sync mode (mbsync)');
if (!creds.appPassword) {
throw new PermanentError('Gmail App Password required for initial sync');
}
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) => {
if (ctx.meta.isIncremental) {
// ── Incremental: direct IMAP via imapflow ──
const creds = ctx.meta.creds as GmailCredentials;
await ctx.updateProgress({ current: 0, total: 0, label: 'Connecting to Gmail...' });
const creds = ctx.meta.creds as GmailCredentials;
const db = openEmailDb(ctx.job.userId);
try {
const result = await incrementalImapSync({
const db = openEmailDb(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) => {
@@ -589,200 +643,71 @@ const gmailSyncHandler: JobHandler = {
},
});
console.log(
`[gmail-sync] Incremental 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();
}
} else {
// ── Full sync: mbsync ──
const email = ctx.meta.email as string;
const appPassword = ctx.meta.appPassword as string;
const maildirPath = getMaildirPath(ctx.job.userId);
await mkdir(maildirPath, { recursive: true });
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 {
let proc: ReturnType<typeof Bun.spawn>;
try {
proc = Bun.spawn(['mbsync', '-c', configPath, '-a'], {
stdout: 'pipe',
stderr: 'pipe',
});
} catch (spawnErr) {
const msg = spawnErr instanceof Error ? spawnErr.message : String(spawnErr);
throw new PermanentError(`Failed to start mbsync: ${msg}`);
// 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...' });
let emailCount = 0;
let counting = true;
const countLoop = (async () => {
while (counting) {
await new Promise((r) => setTimeout(r, 3000));
if (!counting) break;
emailCount = await countMaildirFiles(maildirPath);
ctx.updateProgress({
current: emailCount,
total: 0,
label: `Downloading — ${emailCount.toLocaleString()} emails`,
});
}
})();
let stderrBuf = '';
const reader = (proc.stderr as ReadableStream<Uint8Array>).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) console.log(`[gmail-sync] mbsync: ${trimmed}`);
}
}
})();
const exitCode = await proc.exited;
counting = false;
await readLoop;
await countLoop;
if (exitCode !== 0) {
const isOverquota = stderrBuf.includes('OVERQUOTA');
const isAuthFail =
stderrBuf.includes('AUTHENTICATIONFAILED') || stderrBuf.includes('Invalid credentials');
if (isOverquota) {
const emailCount = await countMaildirFiles(maildirPath);
console.log(`[gmail-sync] Gmail OVERQUOTA — proceeding to import ${emailCount} downloaded emails`);
ctx.meta.gmailSyncPartial = true;
ctx.meta.gmailSyncEmailCount = emailCount;
} else {
console.error(`[gmail-sync] mbsync failed`);
if (isAuthFail) {
const emailCount = await countMaildirFiles(maildirPath);
ctx.meta.gmailSyncRecoverable = true;
ctx.meta.gmailSyncEmailCount = emailCount;
ctx.meta.gmailSyncIsAuthFail = true;
throw new PermanentError(`Authentication failed — check your App Password`);
}
throw new Error(`mbsync exited with code ${exitCode}: ${stderrBuf.slice(-500)}`);
}
} else {
emailCount = await countMaildirFiles(maildirPath);
console.log(`[gmail-sync] mbsync completed successfully — ${emailCount} emails`);
await ctx.updateProgress({
current: emailCount,
total: emailCount,
label: `Download complete — ${emailCount.toLocaleString()} emails`,
});
}
} finally {
await unlink(configPath).catch(() => {});
result = await gmailApiSync({
creds,
db,
onProgress: (saved, total) => {
ctx.updateProgress({ current: saved, total, label: `Fetching new emails (${saved}/${total})...` });
},
});
}
}
},
},
{
name: 'Import to database',
run: async (ctx) => {
// Incremental sync already imported in the previous step
if (ctx.meta.isIncremental) {
const result = ctx.meta.syncResult as IncrementalSyncResult | undefined;
// Auto-add /email to dock
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', '/projects', '/dashboards', '/chat'];
await setDockPaths(dbUser.id, [...defaults, '/email']);
} else if (!paths.includes('/email')) {
await setDockPaths(dbUser.id, [...paths, '/email']);
}
}
} catch {
// Non-fatal
}
}
await ctx.updateProgress({ current: 1, total: 1, label: `Done — ${result?.saved ?? 0} new emails` });
return;
}
// Full sync: import from Maildir
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 {
const lastSyncAt = getSyncMeta(db, 'last_sync_at');
const result = await importMaildir({
maildirPath,
emailAccount,
db,
lastSyncAt,
onProgress: (saved, skipped) => {
ctx.updateProgress({
current: saved + skipped,
total: 0,
label: `Saved ${saved.toLocaleString()}, skipped ${skipped.toLocaleString()}`,
});
},
});
console.log(
`[gmail-sync] Import done: saved ${result.saved}, skipped ${result.skipped}, errors ${result.errors}`,
`[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());
// Auto-add /email to dock if not already present
if (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', '/projects', '/dashboards', '/chat'];
await setDockPaths(dbUser.id, [...defaults, '/email']);
} else if (!paths.includes('/email')) {
await setDockPaths(dbUser.id, [...paths, '/email']);
}
}
} catch {
// Non-fatal — dock update is best-effort
}
}
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) => {
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', '/projects', '/dashboards', '/chat'];
await setDockPaths(dbUser.id, [...defaults, '/email']);
} else if (!paths.includes('/email')) {
await setDockPaths(dbUser.id, [...paths, '/email']);
}
}
} catch {
// Non-fatal
}
}
},
},
],
};