email: the sidecar schedules its own syncs
Stage 2, and the end of the inversion. The two sync handlers (1,093 lines) ran in the platform's queue, which meant the sidecar reached back over its registration socket to ask the platform to enqueue work, and the credentials travelled through Postgres job metadata to get there. Option (A) from the plan: they run here now, and the Jobs screen is left to the things it actually describes. The handlers moved almost unedited. Their bodies were already a list of steps taking a context, so sync-runner.ts synthesizes that context and runs them; what went away is the JobHandler wrapper and the registration. `job.userId` is the OWNER'S EMAIL rather than a numeric id — the queue's naming — and it resolves the mail store path, so it is called out in the type. That is the same field whose absence made the mailbox read as empty two commits ago; it is set from user.email and checked this time. Deliberately not a queue: one run per account, no persistence, no retry. A failure is picked up by the ten-minute cron like any other, and a sync interrupted by a restart resumes from the stored cursor rather than the beginning. PermanentError survives as a local class — it signalled "do not retry" to the queue and now just carries its message to the sync state. accounts.ts asks the runner whether an account is syncing instead of scanning job rows, and the queue-over-WS shim in index.ts is gone: enqueueViaWs, listJobsViaWs, the pending-response map and the queue branch in the command handler. Nothing but a port crosses that socket now. The three chat channels stop opening the mail store directly. They each carried their own copy of count-rows / enqueue / poll / count-again, coupling three chat bridges to the mail schema — and they enqueued `gmail-sync` unconditionally, the OAuth path, for an app-password account that syncs over IMAP, so the command was already broken. One shared helper calls a new POST /sync-now on the sidecar, which syncs and reports what arrived. queue/handlers/ is now empty; both handlers there were email. The queue is untouched and still serves the Jobs screen. Not moved, and fine where they are: scripts/migrate-emails-to-sqlite.ts and scripts/seed-imap-uids.ts are one-off maintenance scripts that open the store directly. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5
@@ -0,0 +1,385 @@
|
||||
import { createHash } from 'node:crypto';
|
||||
import { openEmailDb, upsertFromRawEml } from './store';
|
||||
import { refreshGoogleAccessToken } from '../../api/integrations/google-auth';
|
||||
import {
|
||||
getEmailAccount,
|
||||
getUserIntegration,
|
||||
upsertUserIntegration,
|
||||
getServerIntegration,
|
||||
updateEmailAccountStatus,
|
||||
updateEmailAccountSyncMeta,
|
||||
getDockPaths,
|
||||
setDockPaths,
|
||||
} from 'officerdb';
|
||||
|
||||
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 {}
|
||||
|
||||
// ── 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 {
|
||||
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);
|
||||
}
|
||||
|
||||
// ── IMAP folder → label mapping ──
|
||||
|
||||
const SPECIAL_USE_LABEL_MAP: Record<string, string> = {
|
||||
'\\Inbox': 'inbox',
|
||||
'\\Sent': 'sent',
|
||||
'\\Drafts': 'draft',
|
||||
'\\Flagged': 'starred',
|
||||
'\\Trash': 'trash',
|
||||
'\\Junk': 'spam',
|
||||
'\\All': 'archive',
|
||||
};
|
||||
|
||||
const SKIP_SPECIAL_USE = new Set(['\\Trash', '\\Junk', '\\All']);
|
||||
|
||||
type FolderInfo = { specialUse?: string; path: string; flags: Set<string>; status?: { uidNext?: number; uidValidity?: number } };
|
||||
|
||||
function shouldSkipFolder(folder: FolderInfo): boolean {
|
||||
if (folder.flags.has('\\Noselect') || folder.flags.has('\\NonExistent')) return true;
|
||||
if (folder.specialUse && SKIP_SPECIAL_USE.has(folder.specialUse)) return true;
|
||||
return false;
|
||||
}
|
||||
|
||||
function folderToLabel(folder: FolderInfo): string {
|
||||
if (folder.specialUse && SPECIAL_USE_LABEL_MAP[folder.specialUse]) {
|
||||
return SPECIAL_USE_LABEL_MAP[folder.specialUse]!;
|
||||
}
|
||||
if (folder.path === 'INBOX') return 'inbox';
|
||||
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 = {
|
||||
type: 'email-sync',
|
||||
retry: { delayMs: 15 * 60 * 1000, maxRetries: 5 },
|
||||
steps: [
|
||||
{
|
||||
name: 'Sync emails',
|
||||
run: async (ctx: SyncCtx) => {
|
||||
const { ImapFlow } = await import('imapflow');
|
||||
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, account.email);
|
||||
|
||||
// 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,
|
||||
port: account.imapPort,
|
||||
secure: account.imapSecure,
|
||||
auth: imapAuth,
|
||||
logger: false,
|
||||
socketTimeout: 30 * 60 * 1000,
|
||||
});
|
||||
|
||||
const connState = { error: null as Error | null };
|
||||
client.on('error', (err: Error) => {
|
||||
console.log(`[email-sync] IMAP connection error: ${err.message}`);
|
||||
connState.error = err;
|
||||
});
|
||||
|
||||
try {
|
||||
await client.connect();
|
||||
console.log(`[email-sync] IMAP connected${reconnects > 0 ? ` (reconnect ${reconnects})` : ''}`);
|
||||
|
||||
const folders = await client.list({ statusQuery: { uidNext: true, uidValidity: true } }) as FolderInfo[];
|
||||
|
||||
// Filter to folders that still need syncing
|
||||
const foldersToSync: Array<{ folder: FolderInfo; lastUid: number }> = [];
|
||||
for (const folder of folders) {
|
||||
if (shouldSkipFolder(folder)) continue;
|
||||
|
||||
const uidValidityKey = `imap_uidvalidity:${folder.path}`;
|
||||
const lastUidKey = `imap_lastuid:${folder.path}`;
|
||||
const storedUidValidity = syncMeta[uidValidityKey] as string | undefined;
|
||||
const storedLastUid = syncMeta[lastUidKey] as string | undefined;
|
||||
|
||||
const uidValidity = folder.status?.uidValidity ? String(folder.status.uidValidity) : null;
|
||||
const uidNext = folder.status?.uidNext ?? 0;
|
||||
let lastUid = storedUidValidity && uidValidity === storedUidValidity && storedLastUid ? parseInt(storedLastUid, 10) : 0;
|
||||
|
||||
if (storedUidValidity && uidValidity && storedUidValidity !== uidValidity) {
|
||||
console.log(`[email-sync] UIDVALIDITY changed for ${folder.path} — will re-scan`);
|
||||
lastUid = 0;
|
||||
}
|
||||
|
||||
if (lastUid > 0 && uidNext <= lastUid + 1) continue;
|
||||
|
||||
foldersToSync.push({ folder, lastUid });
|
||||
}
|
||||
|
||||
if (foldersToSync.length === 0) {
|
||||
console.log('[email-sync] All folders synced');
|
||||
allDone = true;
|
||||
await client.logout().catch(() => {});
|
||||
break;
|
||||
}
|
||||
|
||||
console.log(`[email-sync] ${foldersToSync.length} folder(s) to sync`);
|
||||
|
||||
let connectionLost = false;
|
||||
|
||||
for (let fi = 0; fi < foldersToSync.length; fi++) {
|
||||
const { folder, lastUid } = foldersToSync[fi]!;
|
||||
|
||||
if (connState.error) { connectionLost = true; break; }
|
||||
|
||||
console.log(`[email-sync] Syncing folder ${fi + 1}/${foldersToSync.length}: ${folder.path}`);
|
||||
|
||||
let lock;
|
||||
try {
|
||||
lock = await client.getMailboxLock(folder.path);
|
||||
} catch (err) {
|
||||
const errMsg = err instanceof Error ? err.message : String(err);
|
||||
if (errMsg.includes('not available') || errMsg.includes('timeout') || errMsg.includes('closed')) {
|
||||
console.log(`[email-sync] Connection lost at folder ${folder.path}`);
|
||||
connectionLost = true;
|
||||
break;
|
||||
}
|
||||
console.log(`[email-sync] Skipping folder ${folder.path}: ${errMsg}`);
|
||||
continue;
|
||||
}
|
||||
|
||||
try {
|
||||
const mailbox = client.mailbox;
|
||||
if (!mailbox) continue;
|
||||
|
||||
const uidValidity = String(mailbox.uidValidity);
|
||||
const effectiveLastUid = syncMeta[`imap_uidvalidity:${folder.path}`] === uidValidity ? lastUid : 0;
|
||||
let maxUid = effectiveLastUid;
|
||||
const label = folderToLabel(folder);
|
||||
const range = effectiveLastUid > 0 ? `${effectiveLastUid + 1}:*` : '1:*';
|
||||
|
||||
try {
|
||||
const fetchOpts: Record<string, unknown> = { source: true, uid: true };
|
||||
if (account.provider === 'gmail') fetchOpts.labels = true;
|
||||
|
||||
for await (const msg of client.fetch(range, fetchOpts, { uid: true })) {
|
||||
if (msg.uid <= effectiveLastUid) continue;
|
||||
|
||||
maxUid = Math.max(maxUid, msg.uid);
|
||||
|
||||
if (!msg.source) { errors++; continue; }
|
||||
|
||||
const raw = msg.source.toString('utf-8');
|
||||
const id = messageIdToStableId(raw);
|
||||
if (!id) { errors++; continue; }
|
||||
|
||||
if (existingIds.has(id)) { skipped++; continue; }
|
||||
|
||||
const labels = [label];
|
||||
try {
|
||||
upsertFromRawEml({ db, id, raw, integration: account.provider, emailAccount: account.email, labels });
|
||||
existingIds.add(id);
|
||||
saved++;
|
||||
} catch {
|
||||
errors++;
|
||||
}
|
||||
|
||||
if ((saved + skipped) % 50 === 0) {
|
||||
const progressLabel = `${folder.path} — ${saved.toLocaleString()} saved, ${skipped.toLocaleString()} skipped`;
|
||||
console.log(`[email-sync] ${progressLabel}`);
|
||||
await ctx.updateProgress({ current: saved + skipped, total: 0, label: progressLabel });
|
||||
|
||||
if (maxUid > effectiveLastUid) {
|
||||
syncMeta[`imap_lastuid:${folder.path}`] = String(maxUid);
|
||||
await updateEmailAccountSyncMeta(account.id, syncMeta as Record<string, unknown>);
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch (fetchErr) {
|
||||
const errMsg = fetchErr instanceof Error ? fetchErr.message : String(fetchErr);
|
||||
if (errMsg.includes('Nothing to fetch')) {
|
||||
// No messages in range — normal
|
||||
} else if (errMsg.includes('not available') || errMsg.includes('timeout') || errMsg.includes('closed')) {
|
||||
console.log(`[email-sync] Connection lost during fetch in ${folder.path}: ${errMsg}`);
|
||||
connectionLost = true;
|
||||
} else {
|
||||
console.log(`[email-sync] Fetch error in ${folder.path}: ${errMsg}`);
|
||||
errors++;
|
||||
}
|
||||
}
|
||||
|
||||
syncMeta[`imap_uidvalidity:${folder.path}`] = uidValidity;
|
||||
if (maxUid > effectiveLastUid) {
|
||||
syncMeta[`imap_lastuid:${folder.path}`] = String(maxUid);
|
||||
}
|
||||
|
||||
console.log(`[email-sync] ${folder.path} done: ${saved} saved, ${skipped} skipped, ${errors} errors`);
|
||||
await ctx.updateProgress({ current: saved + skipped, total: 0, label: `${folder.path}: ${saved} saved` });
|
||||
await updateEmailAccountSyncMeta(account.id, syncMeta as Record<string, unknown>);
|
||||
} finally {
|
||||
lock.release();
|
||||
}
|
||||
|
||||
if (connectionLost) break;
|
||||
}
|
||||
|
||||
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>);
|
||||
} finally {
|
||||
db.close();
|
||||
}
|
||||
|
||||
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` });
|
||||
},
|
||||
},
|
||||
{
|
||||
name: 'Finalize',
|
||||
run: async (ctx: SyncCtx) => {
|
||||
const meta = ctx.meta as unknown as EmailSyncMeta;
|
||||
const saved = meta.saved ?? 0;
|
||||
|
||||
await updateEmailAccountStatus(meta.account.id, 'synced');
|
||||
|
||||
if (saved > 0) {
|
||||
try {
|
||||
const paths = await getDockPaths(meta.account.userId);
|
||||
if (!paths) {
|
||||
const defaults = ['/', '/files', '/automation', '/dashboards', '/chat'];
|
||||
await setDockPaths(meta.account.userId, [...defaults, '/email']);
|
||||
} else if (!paths.includes('/email')) {
|
||||
await setDockPaths(meta.account.userId, [...paths, '/email']);
|
||||
}
|
||||
} catch {
|
||||
// Non-fatal
|
||||
}
|
||||
}
|
||||
|
||||
console.log(`[email-sync] ${meta.account.email} status set to synced`);
|
||||
},
|
||||
},
|
||||
],
|
||||
};
|
||||
|
||||
/** The handler's steps, for the sidecar's own runner. */
|
||||
export const imapSyncSteps: SyncStep[] = emailSyncHandler.steps as SyncStep[];
|
||||
Reference in New Issue
Block a user