officerdev/platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

email sync: API passes full config at enqueue, auto-reconnect on IMAP drops

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
pastilhas
2026-03-06 07:02:51 +00:00
co-authored by Claude Opus 4.6
parent 170bd6d41b
commit 4ececbe748
4 changed files with 336 additions and 221 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/servers/api/email/accounts.ts
+54 -2
View File
@@ -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<number>();
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<string, unknown> | 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<string, unknown>);
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);
src/servers/queue/handlers/email-sync.ts
+147 -122
View File
@@ -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<string, unknown>;
};
imapAuth: { user: string; pass?: string; accessToken?: string };
saved?: number;
};
// ── Stable ID from Message-Id header ──
function messageIdToStableId(raw: string): string | null {
@@ -54,84 +73,94 @@ 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<string, unknown> | 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<string, unknown> | 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<string, unknown>;
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<string, unknown>;
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<string, unknown>;
const meta = ctx.meta as unknown as EmailSyncMeta;
const { account, userEmail } = meta;
// Resolve auth (refreshes OAuth token if needed)
const imapAuth = await resolveImapAuth(meta);
// 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<string, unknown>;
const isIncremental = !!syncMeta.last_sync_at;
await updateEmailAccountStatus(account.id, 'syncing');
console.log(`[email-sync] ${isIncremental ? 'Incremental' : 'Initial'} sync for ${account.email}`);
const MAX_RECONNECTS = 10;
const RECONNECT_DELAY_MS = 5_000;
let reconnects = 0;
let saved = 0;
let skipped = 0;
let errors = 0;
let allDone = false;
const db = openEmailDb(userEmail);
// Load existing IDs for dedup (once, shared across reconnections)
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);
try {
while (!allDone && reconnects <= MAX_RECONNECTS) {
if (reconnects > 0) {
console.log(`[email-sync] Reconnecting (${reconnects}/${MAX_RECONNECTS}) after ${RECONNECT_DELAY_MS / 1000}s...`);
await ctx.updateProgress({ current: saved + skipped, total: 0, label: `Reconnecting (${reconnects}/${MAX_RECONNECTS})...` });
await new Promise((r) => setTimeout(r, RECONNECT_DELAY_MS));
// Re-read syncMeta from DB to get latest saved UIDs
const updated = await getEmailAccount(account.id);
if (updated?.syncMeta) {
Object.assign(syncMeta, updated.syncMeta as Record<string, unknown>);
}
} else {
await ctx.updateProgress({ current: 0, total: 0, label: 'Connecting...' });
}
const client = new ImapFlow({
host: account.imapHost,
@@ -139,34 +168,22 @@ const emailSyncHandler: JobHandler = {
secure: account.imapSecure,
auth: imapAuth,
logger: false,
socketTimeout: 30 * 60 * 1000, // 30 min — large mailboxes need time
socketTimeout: 30 * 60 * 1000,
});
// 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;
});
const db = openEmailDb(userEmail);
let saved = 0;
let skipped = 0;
let errors = 0;
try {
await client.connect();
console.log('[email-sync] IMAP connected');
console.log(`[email-sync] IMAP connected${reconnects > 0 ? ` (reconnect ${reconnects})` : ''}`);
// 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<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 syncable folders and count total messages to fetch
// Filter to folders that still need syncing
const foldersToSync: Array<{ folder: FolderInfo; lastUid: number }> = [];
for (const folder of folders) {
if (shouldSkipFolder(folder)) continue;
@@ -185,19 +202,26 @@ const emailSyncHandler: JobHandler = {
lastUid = 0;
}
// Skip if no new messages
if (lastUid > 0 && uidNext <= lastUid + 1) continue;
foldersToSync.push({ folder, lastUid });
}
if (foldersToSync.length === 0) {
console.log('[email-sync] All folders synced');
allDone = true;
await client.logout().catch(() => {});
break;
}
console.log(`[email-sync] ${foldersToSync.length} folder(s) to sync`);
let connectionLost = false;
for (let fi = 0; fi < foldersToSync.length; fi++) {
const { folder, lastUid } = foldersToSync[fi]!;
// If connection is already dead, stop trying more folders
if (connState.error) break;
if (connState.error) { connectionLost = true; break; }
console.log(`[email-sync] Syncing folder ${fi + 1}/${foldersToSync.length}: ${folder.path}`);
@@ -208,7 +232,7 @@ const emailSyncHandler: JobHandler = {
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);
connectionLost = true;
break;
}
console.log(`[email-sync] Skipping folder ${folder.path}: ${errMsg}`);
@@ -234,22 +258,13 @@ const emailSyncHandler: JobHandler = {
maxUid = Math.max(maxUid, msg.uid);
if (!msg.source) {
errors++;
continue;
}
if (!msg.source) { errors++; continue; }
const raw = msg.source.toString('utf-8');
const id = messageIdToStableId(raw);
if (!id) {
errors++;
continue;
}
if (!id) { errors++; continue; }
if (existingIds.has(id)) {
skipped++;
continue;
}
if (existingIds.has(id)) { skipped++; continue; }
const labels = [label];
try {
@@ -265,7 +280,6 @@ const emailSyncHandler: JobHandler = {
console.log(`[email-sync] ${progressLabel}`);
await ctx.updateProgress({ current: saved + skipped, total: 0, label: progressLabel });
// 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<string, unknown>);
@@ -278,7 +292,7 @@ const emailSyncHandler: JobHandler = {
// 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);
connectionLost = true;
} else {
console.log(`[email-sync] Fetch error in ${folder.path}: ${errMsg}`);
errors++;
@@ -292,33 +306,46 @@ const emailSyncHandler: JobHandler = {
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<string, unknown>);
} finally {
lock.release();
}
if (connectionLost) break;
}
// 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}`);
await client.logout().catch(() => {});
if (connectionLost) {
console.log(`[email-sync] Connection lost after saving ${saved} emails — will reconnect`);
reconnects++;
continue;
}
allDone = true;
} catch (err) {
await client.logout().catch(() => {});
if (reconnects < MAX_RECONNECTS) {
console.log(`[email-sync] Error: ${err instanceof Error ? err.message : String(err)} — will reconnect`);
reconnects++;
continue;
}
throw err;
}
}
if (!allDone) {
throw new Error(`IMAP sync incomplete after ${MAX_RECONNECTS} reconnection attempts`);
}
// Mark sync complete
syncMeta.last_sync_at = new Date().toISOString();
await updateEmailAccountSyncMeta(account.id, syncMeta as Record<string, unknown>);
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`);
},
},
],
src/servers/sidecar/email-cron.ts
+33 -2
View File
@@ -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<string, unknown> = { user: account.email };
if (account.authType === 'oauth') {
const integration = await getUserIntegration(account.userId, 'google');
const config = integration?.config as Record<string, unknown> | 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<string, unknown>;
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) {
src/servers/sidecar/queue-runner.ts
+7
View File
@@ -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();