Claude Opus 5
980a9d3c38
IMAP and Gmail sync into a per-account SQLite store, with the mail read and composed in the browser. moved whole from src/servers/sidecar/email (14 files), src/servers/api/email, officer_db/src/email, and the Screens/Dashboard/Email tree. what changed on the way out: - credentials are encrypted. the column was plaintext jsonb holding an IMAP app password that the SMTP transport read straight out of the row; it is now AES-256-GCM under this plugin's own 'email' secret-store purpose, sealed and opened in queries.ts alone so nothing above it can hold ciphertext by accident. no plaintext fallback on read - a value that does not verify is an error. - the router derives its prefix from mountPrefix() instead of the literal '/api/email', which was only correct for a first-party publisher. - the frontend became panels.ts + layout.ts. the two panel appTypes that were null are now declared, because the platform screen used to override them by panel id and a plugin has no such hook. - account management became a modal. it was a section in the platform's Settings, which a plugin cannot contribute to, and a panel does not work either: PluginScreen renders the workspace locked, so a panel outside the default layout is unreachable. it follows Compose, which had already solved this. - 17 palette classes replaced with semantic tokens. not verified: this has never run. it was switched off in the platform on 2026-08-13 and its table was never created, so nothing has exercised sync, sending or the UI since. the first install is the first test.
428 lines
16 KiB
TypeScript
428 lines
16 KiB
TypeScript
import { createHash } from 'node:crypto';
|
|
import { openEmailDb, upsertFromRawEml } from './store';
|
|
import { readSyncMeta, writeSyncMeta } from './sync-meta';
|
|
import { refreshGoogleAccessToken } from '@@/api/integrations/google-auth';
|
|
import { getUserIntegration, upsertUserIntegration, getServerIntegration, getDockPaths, setDockPaths } from 'officerdb';
|
|
import { getEmailAccount, updateEmailAccountStatus } from '../db/queries';
|
|
|
|
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);
|
|
|
|
// Always re-read the position rather than trusting job meta, which can be stale after a retry.
|
|
const freshAccount = await getEmailAccount(account.id);
|
|
if (!freshAccount) throw new PermanentError(`Email account ${account.id} not found`);
|
|
|
|
// Opened here rather than further down because the position now lives inside it. Postgres is
|
|
// passed only as the one-time backfill for accounts that predate the move — see sync-meta.ts.
|
|
const db = openEmailDb(userEmail, account.email);
|
|
const syncMeta: Record<string, unknown> = readSyncMeta(db, 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;
|
|
|
|
// 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);
|
|
writeSyncMeta(db, syncMeta);
|
|
}
|
|
}
|
|
}
|
|
} 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`,
|
|
});
|
|
writeSyncMeta(db, syncMeta);
|
|
} 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. Last, and only after every message above is stored: if this never runs,
|
|
// the position stays behind and those messages are fetched again. Re-fetching is wasteful;
|
|
// advancing past mail that was never stored would lose it silently.
|
|
syncMeta.last_sync_at = new Date().toISOString();
|
|
writeSyncMeta(db, syncMeta);
|
|
} 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[];
|