wip: email sync via imap with status tracking and auto cron

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-06 04:57:30 +00:00
co-authored by Claude Opus 4.6
parent 5925ac49a1
commit 170bd6d41b
20 changed files with 1929 additions and 373 deletions
+171
View File
@@ -0,0 +1,171 @@
import { createRouter } from '../../create-router';
import { BAD_REQUEST, NOT_FOUND } from '../../custom-errors';
import {
getEmailAccounts,
getEmailAccount,
createEmailAccount,
deleteEmailAccount,
getUserIntegration,
updateEmailAccountStatus,
} from 'officerdb';
import { validateImapConnection } from './imap-validate';
import * as sidecar from '../../sidecar-client';
type CreateAccountBody = {
provider: string;
email: string;
displayName?: string;
imapHost: string;
imapPort: number;
imapSecure: boolean;
authType: string;
credentials: Record<string, unknown>;
};
type ValidateBody = {
imapHost: string;
imapPort: number;
imapSecure: boolean;
authType: string;
email: string;
credentials: Record<string, unknown>;
};
export const accountsRouter = createRouter();
accountsRouter.get('/', async (ctx) => {
const user = ctx.get('user');
const accounts = await getEmailAccounts(user.id);
return ctx.json(
accounts.map((a) => ({
id: a.id,
provider: a.provider,
email: a.email,
displayName: a.displayName,
enabled: a.enabled,
status: a.status,
createdAt: a.createdAt,
})),
);
});
accountsRouter.post('/', async (ctx) => {
const user = ctx.get('user');
const body = ctx.get('body') as CreateAccountBody;
if (!body.provider || !body.email || !body.imapHost || !body.imapPort || !body.authType) {
throw BAD_REQUEST('Missing required fields');
}
const authResult = await resolveAuth(user.id, body.authType, body.email, body.credentials);
if (!authResult.ok) throw BAD_REQUEST(authResult.error);
const validation = await validateImapConnection({
host: body.imapHost,
port: body.imapPort,
secure: body.imapSecure,
user: body.email,
...authResult.auth,
});
if (!validation.ok) throw BAD_REQUEST(`IMAP connection failed: ${validation.error}`);
const account = await createEmailAccount({
userId: user.id,
provider: body.provider,
email: body.email,
displayName: body.displayName,
imapHost: body.imapHost,
imapPort: body.imapPort,
imapSecure: body.imapSecure,
authType: body.authType,
credentials: body.credentials,
});
return ctx.json({ id: account.id, provider: account.provider, email: account.email }, 201);
});
accountsRouter.delete('/:id', async (ctx) => {
const user = ctx.get('user');
const id = Number(ctx.req.param('id'));
const deleted = await deleteEmailAccount(id, user.id);
if (!deleted) throw NOT_FOUND('Account not found');
return ctx.json({ ok: true });
});
accountsRouter.post('/:id/sync', async (ctx) => {
const user = ctx.get('user');
const id = Number(ctx.req.param('id'));
const account = await getEmailAccount(id);
if (!account || account.userId !== user.id) throw NOT_FOUND('Account not found');
if (account.status === 'queued') throw BAD_REQUEST('Sync is already queued');
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');
// Set status immediately so the UI reflects the queued state
await updateEmailAccountStatus(id, 'queued');
const job = await sidecar.enqueueJob({
lane: 'email',
type: 'email-sync',
userId: user.email,
meta: { emailAccountId: id },
});
return ctx.json({ ok: true, jobId: job.id }, 201);
});
accountsRouter.post('/validate', async (ctx) => {
const user = ctx.get('user');
const body = ctx.get('body') as ValidateBody;
if (!body.imapHost || !body.imapPort || !body.authType || !body.email) {
throw BAD_REQUEST('Missing required fields');
}
const authResult = await resolveAuth(user.id, body.authType, body.email, body.credentials);
if (!authResult.ok) return ctx.json({ ok: false, error: authResult.error });
const result = await validateImapConnection({
host: body.imapHost,
port: body.imapPort,
secure: body.imapSecure,
user: body.email,
...authResult.auth,
});
return ctx.json(result);
});
type AuthResult = { ok: true; auth: { pass?: string; accessToken?: string } } | { ok: false; error: string };
async function resolveAuth(
userId: number,
authType: string,
email: string,
credentials: Record<string, unknown>,
): Promise<AuthResult> {
if (authType === 'oauth') {
const integrationId = credentials.userIntegrationId as number | undefined;
if (!integrationId) return { ok: false, error: 'Missing userIntegrationId for OAuth' };
const integration = await getUserIntegration(userId, 'google');
if (!integration) return { ok: false, error: 'Google integration not found' };
const config = integration.config as Record<string, unknown>;
const accessToken = config.accessToken as string | undefined;
if (!accessToken) return { ok: false, error: 'No access token available — reconnect Google account' };
return { ok: true, auth: { accessToken } };
}
if (authType === 'password') {
const pass = credentials.password as string | undefined;
if (!pass) return { ok: false, error: 'Missing password' };
return { ok: true, auth: { pass } };
}
return { ok: false, error: `Unknown auth type: ${authType}` };
}
+30 -13
View File
@@ -4,9 +4,12 @@ import type { EmailMessage } from 'types';
import { createRouter } from '../../create-router';
import { DATA_PATH } from '@@/data-path';
import { openEmailDb, rowToSummary, getSyncMeta } from './email-db';
import { accountsRouter } from './accounts';
export const emailRouter = createRouter();
emailRouter.route('/accounts', accountsRouter);
emailRouter.get('/messages', async (ctx) => {
const email = ctx.get('user').email;
const page = Number(ctx.req.query('page') ?? '1');
@@ -18,7 +21,9 @@ emailRouter.get('/messages', async (ctx) => {
const db = openEmailDb(email);
try {
const rows = db.query(`SELECT * FROM emails WHERE ${folderWhere} ORDER BY date DESC LIMIT ? OFFSET ?`).all(limit, offset) as Record<string, unknown>[];
const rows = db
.query(`SELECT * FROM emails WHERE ${folderWhere} ORDER BY date DESC LIMIT ? OFFSET ?`)
.all(limit, offset) as Record<string, unknown>[];
const countRow = db.query(`SELECT COUNT(*) as total FROM emails WHERE ${folderWhere}`).get() as { total: number };
const messages = rows.map(rowToSummary);
return ctx.json({ messages, total: countRow.total });
@@ -36,11 +41,11 @@ emailRouter.get('/messages/:id', async (ctx) => {
const row = db.query('SELECT * FROM emails WHERE id = ? AND deleted = 0').get(id) as Record<string, unknown> | null;
if (!row) return ctx.text('Not found', 404);
const attachmentRows = db.query('SELECT * FROM attachments WHERE email_id = ? ORDER BY idx').all(id) as Array<Record<string, unknown>>;
const attachmentRows = db.query('SELECT * FROM attachments WHERE email_id = ? ORDER BY idx').all(id) as Array<
Record<string, unknown>
>;
const from = row.from_name
? `${row.from_name} <${row.from_address}>`
: (row.from_address as string);
const from = row.from_name ? `${row.from_name} <${row.from_address}>` : (row.from_address as string);
const message: EmailMessage = {
id: row.id as string,
@@ -75,7 +80,10 @@ emailRouter.post('/messages/:id/attachments/:index/extract', async (ctx) => {
const db = openEmailDb(email);
try {
const row = db.query('SELECT filename, content FROM attachments WHERE email_id = ? AND idx = ?').get(id, index) as { filename: string; content: string | null } | null;
const row = db.query('SELECT filename, content FROM attachments WHERE email_id = ? AND idx = ?').get(id, index) as {
filename: string;
content: string | null;
} | null;
if (!row || !row.content) return ctx.text('Attachment not found', 404);
const fileName = row.filename ?? 'unknown';
@@ -144,9 +152,18 @@ emailRouter.get('/stats', async (ctx) => {
const db = openEmailDb(email);
try {
const total = (db.query(`SELECT COUNT(*) as count FROM emails WHERE ${folderWhere}`).get() as { count: number }).count;
const byDomain = db.query(`SELECT from_domain, COUNT(*) as count FROM emails WHERE ${folderWhere} GROUP BY from_domain ORDER BY count DESC LIMIT 20`).all() as Array<{ from_domain: string; count: number }>;
const bySender = db.query(`SELECT from_address, from_name, COUNT(*) as count FROM emails WHERE ${folderWhere} GROUP BY from_address ORDER BY count DESC LIMIT 20`).all() as Array<{ from_address: string; from_name: string; count: number }>;
const total = (db.query(`SELECT COUNT(*) as count FROM emails WHERE ${folderWhere}`).get() as { count: number })
.count;
const byDomain = db
.query(
`SELECT from_domain, COUNT(*) as count FROM emails WHERE ${folderWhere} GROUP BY from_domain ORDER BY count DESC LIMIT 20`,
)
.all() as Array<{ from_domain: string; count: number }>;
const bySender = db
.query(
`SELECT from_address, from_name, COUNT(*) as count FROM emails WHERE ${folderWhere} GROUP BY from_address ORDER BY count DESC LIMIT 20`,
)
.all() as Array<{ from_address: string; from_name: string; count: number }>;
return ctx.json({ total, byDomain, bySender });
} finally {
@@ -159,7 +176,9 @@ emailRouter.get('/labels', async (ctx) => {
const db = openEmailDb(email);
try {
const rows = db.query('SELECT labels FROM emails WHERE deleted = 0 AND labels IS NOT NULL').all() as Array<{ labels: string }>;
const rows = db.query('SELECT labels FROM emails WHERE deleted = 0 AND labels IS NOT NULL').all() as Array<{
labels: string;
}>;
const counts = new Map<string, number>();
for (const row of rows) {
@@ -169,9 +188,7 @@ emailRouter.get('/labels', async (ctx) => {
}
}
const labels = [...counts.entries()]
.map(([label, count]) => ({ label, count }))
.sort((a, b) => b.count - a.count);
const labels = [...counts.entries()].map(([label, count]) => ({ label, count })).sort((a, b) => b.count - a.count);
return ctx.json({ labels });
} finally {
+54
View File
@@ -0,0 +1,54 @@
type ValidateImapParams = {
host: string;
port: number;
secure: boolean;
user: string;
pass?: string;
accessToken?: string;
};
type ValidateImapResult = { ok: true; folderCount: number } | { ok: false; error: string };
export async function validateImapConnection(params: ValidateImapParams): Promise<ValidateImapResult> {
const { ImapFlow } = await import('imapflow');
const auth: { user: string; pass?: string; accessToken?: string } = { user: params.user };
if (params.accessToken) {
auth.accessToken = params.accessToken;
} else if (params.pass) {
auth.pass = params.pass;
} else {
return { ok: false, error: 'No authentication credentials provided' };
}
const client = new ImapFlow({
host: params.host,
port: params.port,
secure: params.secure,
auth,
logger: false,
greetingTimeout: 60_000,
socketTimeout: 60_000,
});
try {
const result = await Promise.race([
(async () => {
await client.connect();
const folders = await client.list();
await client.logout();
return { ok: true as const, folderCount: folders.length };
})(),
new Promise<ValidateImapResult>((_, reject) =>
setTimeout(() => reject(new Error('Connection timed out')), 90_000),
),
]);
return result;
} catch (err) {
try {
client.close();
} catch {}
const message = err instanceof Error ? err.message : 'Connection failed';
return { ok: false, error: message };
}
}
@@ -0,0 +1,34 @@
import { getServerIntegration } from 'officerdb';
import { PermanentError } from '../../queue/types';
type TokenRefreshResult = { accessToken: string; expiresAt: number };
export async function refreshGoogleAccessToken(refreshToken: string): Promise<TokenRefreshResult> {
const serverGoogle = await getServerIntegration('google');
const serverConfig = serverGoogle?.config as Record<string, unknown> | undefined;
if (!serverConfig?.clientId || !serverConfig?.clientSecret) {
throw new PermanentError('Google OAuth not configured on server');
}
const response = await fetch('https://oauth2.googleapis.com/token', {
method: 'POST',
headers: { 'Content-Type': 'application/x-www-form-urlencoded' },
body: new URLSearchParams({
client_id: serverConfig.clientId as string,
client_secret: serverConfig.clientSecret as string,
refresh_token: refreshToken,
grant_type: 'refresh_token',
}),
});
if (!response.ok) {
const text = await response.text();
throw new Error(`Google token refresh failed: ${text}`);
}
const data = (await response.json()) as { access_token: string; expires_in: number };
return {
accessToken: data.access_token,
expiresAt: Date.now() + data.expires_in * 1000,
};
}
@@ -259,11 +259,16 @@ export const googleCallbackHandler = async (ctx: any) => {
const serverIntegration = await getServerIntegration('google');
// Merge with existing config to preserve fields like imapAppPassword
const existingIntegration = await getUserIntegration(dbUser.id, 'google');
const existingConfig = (existingIntegration?.config as Record<string, unknown>) ?? {};
await upsertUserIntegration({
userId: dbUser.id,
provider: 'google',
serverIntegrationId: serverIntegration?.id,
config: {
...existingConfig,
accessToken: tokens.access_token,
refreshToken: tokens.refresh_token,
expiresAt: Date.now() + tokens.expires_in * 1000,
+356
View File
@@ -0,0 +1,356 @@
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 { refreshGoogleAccessToken } from '../../api/integrations/google-auth';
import {
getEmailAccount,
getUserById,
getUserIntegration,
upsertUserIntegration,
getServerIntegration,
updateEmailAccountStatus,
updateEmailAccountSyncMeta,
getDockPaths,
setDockPaths,
} from 'officerdb';
// ── 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();
}
// ── 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>;
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, // 30 min — large mailboxes need time
});
// 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');
// 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
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;
}
// Skip if no new messages
if (lastUid > 0 && uidNext <= lastUid + 1) continue;
foldersToSync.push({ folder, lastUid });
}
console.log(`[email-sync] ${foldersToSync.length} folder(s) to sync`);
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;
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}`);
connState.error = connState.error ?? new Error(errMsg);
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 });
// 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>);
}
}
}
} 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}`);
connState.error = connState.error ?? new Error(errMsg);
} 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` });
// Save sync meta after each folder
await updateEmailAccountSyncMeta(account.id, syncMeta as Record<string, unknown>);
} finally {
lock.release();
}
}
// 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}`);
}
// 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}`);
ctx.meta.saved = saved;
await ctx.updateProgress({ current: 1, total: 1, label: `Done — ${saved} new emails` });
},
},
{
name: 'Finalize',
run: async (ctx) => {
const account = ctx.meta.account as { id: number; userId: number; email: string };
const saved = ctx.meta.saved as number;
// Update account status to synced
await updateEmailAccountStatus(account.id, 'synced');
// Auto-add /email to dock
if (saved > 0) {
try {
const paths = await getDockPaths(account.userId);
if (!paths) {
const defaults = ['/', '/files', '/automation', '/projects', '/dashboards', '/chat'];
await setDockPaths(account.userId, [...defaults, '/email']);
} else if (!paths.includes('/email')) {
await setDockPaths(account.userId, [...paths, '/email']);
}
} catch {
// Non-fatal
}
}
console.log(`[email-sync] ${account.email} status set to synced`);
},
},
],
};
registerHandler(emailSyncHandler);
+488 -112
View File
@@ -4,28 +4,59 @@ 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 } from '../../api/email/email-db';
import { getUserByEmail, getUserIntegration, getDockPaths, setDockPaths } from 'officerdb';
import { openEmailDb, upsertFromRawEml, setSyncMeta, getSyncMeta } from '../../api/email/email-db';
import {
getUserByEmail,
getUserIntegration,
upsertUserIntegration,
getServerIntegration,
getDockPaths,
setDockPaths,
} from 'officerdb';
import { getMaildirPath } from '@@/data-path';
import { refreshGoogleAccessToken } from '@@/api/integrations/google-auth';
// ── Credentials ──
type ImapCredentials = { email: string; appPassword: string };
type GmailCredentials = {
email: string;
userId: number;
appPassword?: string;
accessToken?: string;
refreshToken?: string;
expiresAt?: number;
};
async function loadImapCredentials(userEmail: string): Promise<ImapCredentials> {
async function loadGmailCredentials(userEmail: string): Promise<GmailCredentials> {
const dbUser = await getUserByEmail(userEmail);
if (!dbUser) throw new PermanentError('User not found');
const userGoogle = await getUserIntegration(dbUser.id, 'google');
const config = userGoogle?.config as Record<string, unknown> | undefined;
const gmailEmail = (config?.email as string) ?? userEmail;
// Prefer OAuth tokens, fall back to app password
if (config?.accessToken && config?.refreshToken) {
return {
email: gmailEmail,
userId: dbUser.id,
accessToken: config.accessToken as string,
refreshToken: config.refreshToken as string,
expiresAt: config.expiresAt as number | undefined,
appPassword: config.imapAppPassword as string | undefined,
};
}
if (!config?.imapAppPassword) {
throw new PermanentError('Gmail App Password not configured — set it in Settings → Integrations');
}
const gmailEmail = (config.email as string) ?? userEmail;
return { email: gmailEmail, appPassword: config.imapAppPassword as string };
return { email: gmailEmail, userId: dbUser.id, appPassword: config.imapAppPassword as string };
}
// refreshGoogleAccessToken is imported from @@/api/integrations/google-auth
// ── mbsync config ──
function buildMbsyncConfig(email: string, appPassword: string, maildirPath: string): string {
@@ -48,7 +79,7 @@ SubFolders Verbatim
Channel gmail
Far :gmail-remote:
Near :gmail-local:
Patterns * ![Gmail]/Trash ![Gmail]/Spam ![Gmail]/Bin ![Google Mail]/Trash ![Google Mail]/Spam ![Google Mail]/Bin
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 *
@@ -99,21 +130,40 @@ function messageIdToStableId(raw: string): string | null {
type ImportResult = { saved: number; skipped: number; errors: number };
export async function importMaildir(
maildirPath: string,
emailAccount: string,
db: Database,
onProgress?: (saved: number, skipped: number) => void,
): Promise<ImportResult> {
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
@@ -162,6 +212,12 @@ export async function importMaildir(
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) {
@@ -238,6 +294,220 @@ async function countMaildirFiles(maildirPath: string): Promise<number> {
return count;
}
// ── Gmail IMAP label → our label mapping ──
const GMAIL_LABEL_MAP: Record<string, string> = {
'\\Inbox': 'inbox',
'\\Sent': 'sent',
'\\Drafts': 'draft',
'\\Starred': 'starred',
'\\Important': 'important',
'\\All': 'archive',
'\\Trash': 'trash',
'\\Junk': 'spam',
};
function gmailLabelsToLabels(gmailLabels: Set<string>): string[] {
const labels: string[] = [];
for (const gl of gmailLabels) {
const mapped = GMAIL_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 {
if (folder.flags.has('\\Noselect') || folder.flags.has('\\NonExistent')) return true;
if (folder.specialUse && SKIP_SPECIAL_USE.has(folder.specialUse)) return true;
const norm = normalizeGmailFolder(folder.path);
return norm === 'All Mail' || norm === 'Trash' || norm === 'Spam' || norm === 'Bin';
}
// ── Incremental IMAP sync ──
type IncrementalSyncParams = {
creds: GmailCredentials;
db: Database;
onProgress?: (fetched: number, folder: string) => void;
};
type IncrementalSyncResult = { saved: number; skipped: number; errors: number };
async function incrementalImapSync({ creds, db, onProgress }: IncrementalSyncParams): Promise<IncrementalSyncResult> {
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) {
auth.pass = creds.appPassword;
} else {
throw new PermanentError('No authentication method available for IMAP');
}
const client = new ImapFlow({
host: 'imap.gmail.com',
port: 993,
secure: true,
auth,
logger: false,
});
let saved = 0;
let skipped = 0;
let errors = 0;
try {
await client.connect();
console.log('[gmail-sync] Incremental IMAP connected');
// 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;
const folderPath = folder.path;
const uidValidityKey = `imap_uidvalidity:${folderPath}`;
const lastUidKey = `imap_lastuid:${folderPath}`;
const storedUidValidity = getSyncMeta(db, uidValidityKey);
const storedLastUid = getSyncMeta(db, lastUidKey);
const uidValidity = folder.status?.uidValidity ? String(folder.status.uidValidity) : null;
const uidNext = folder.status?.uidNext ?? 0;
const lastUid =
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`);
}
foldersToSync.push(folder);
}
console.log(`[gmail-sync] ${foldersToSync.length} folder(s) with new messages`);
for (const folder of foldersToSync) {
const folderPath = folder.path;
const uidValidityKey = `imap_uidvalidity:${folderPath}`;
const lastUidKey = `imap_lastuid:${folderPath}`;
const storedUidValidity = getSyncMeta(db, uidValidityKey);
const storedLastUid = getSyncMeta(db, lastUidKey);
// Open folder and fetch new messages
let lock;
try {
lock = await client.getMailboxLock(folderPath);
} catch (err) {
console.log(`[gmail-sync] Skipping folder ${folderPath}: ${err instanceof Error ? err.message : err}`);
continue;
}
try {
const mailbox = client.mailbox;
if (!mailbox) continue;
const uidValidity = String(mailbox.uidValidity);
let lastUid = 0;
if (storedUidValidity === uidValidity && storedLastUid) {
lastUid = parseInt(storedLastUid, 10);
}
const range = lastUid > 0 ? `${lastUid + 1}:*` : '1:*';
let maxUid = lastUid;
let folderFetched = 0;
try {
for await (const msg of client.fetch(range, { source: true, labels: true, uid: true }, { uid: true })) {
if (msg.uid <= lastUid) continue;
maxUid = Math.max(maxUid, msg.uid);
folderFetched++;
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;
}
// 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 labels = folderLabel ? [...new Set([folderLabel, ...gmailLabels])] : gmailLabels;
try {
upsertFromRawEml({ db, id, raw, integration: 'gmail', emailAccount: creds.email, labels });
existingIds.add(id);
saved++;
} catch {
errors++;
}
}
} catch (fetchErr) {
const msg = fetchErr instanceof Error ? fetchErr.message : String(fetchErr);
if (!msg.includes('Nothing to fetch')) {
console.log(`[gmail-sync] Fetch error in ${folderPath}: ${msg}`);
errors++;
}
}
if (maxUid > lastUid) {
setSyncMeta(db, lastUidKey, String(maxUid));
}
setSyncMeta(db, uidValidityKey, uidValidity);
if (folderFetched > 0) {
console.log(`[gmail-sync] ${folderPath}: fetched ${folderFetched}, saved ${saved}, skipped ${skipped}`);
onProgress?.(saved + skipped, folderPath);
}
} finally {
lock.release();
}
}
} finally {
await client.logout().catch(() => {});
}
return { saved, skipped, errors };
}
// ── Handler ──
const gmailSyncHandler: JobHandler = {
@@ -247,119 +517,219 @@ const gmailSyncHandler: JobHandler = {
{
name: 'Verify credentials',
run: async (ctx) => {
const creds = await loadImapCredentials(ctx.job.userId);
const creds = await loadGmailCredentials(ctx.job.userId);
// Determine sync mode: incremental if we have a previous sync
const db = openEmailDb(ctx.job.userId);
try {
const lastSyncAt = getSyncMeta(db, 'last_sync_at');
ctx.meta.isIncremental = !!lastSyncAt;
} finally {
db.close();
}
// For incremental sync with OAuth, refresh token if expired
if (ctx.meta.isIncremental && 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');
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');
await upsertUserIntegration({
userId: creds.userId,
provider: 'google',
serverIntegrationId: serverGoogle?.id,
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;
}
}
}
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');
}
}
},
},
{
name: 'Sync via mbsync',
name: 'Sync emails',
run: async (ctx) => {
const email = ctx.meta.email as string;
const appPassword = ctx.meta.appPassword as string;
const maildirPath = getMaildirPath(ctx.job.userId);
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...' });
// 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 {
let proc: ReturnType<typeof Bun.spawn>;
const db = openEmailDb(ctx.job.userId);
try {
proc = Bun.spawn(['mbsync', '-c', configPath, '-a', '-V'], {
stdout: 'pipe',
stderr: 'pipe',
const result = await incrementalImapSync({
creds,
db,
onProgress: (fetched, folder) => {
ctx.updateProgress({ current: fetched, total: 0, label: `Syncing ${folder}...` });
},
});
} catch (spawnErr) {
const msg = spawnErr instanceof Error ? spawnErr.message : String(spawnErr);
throw new PermanentError(`Failed to start mbsync: ${msg}`);
}
// Periodically count downloaded emails and update progress
let emailCount = 0;
let counting = true;
const countLoop = (async () => {
while (counting) {
await new Promise((r) => setTimeout(r, 3000));
if (!counting) break;
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}`);
}
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);
ctx.updateProgress({
console.log(`[gmail-sync] mbsync completed successfully — ${emailCount} emails`);
await ctx.updateProgress({
current: emailCount,
total: 0,
label: `Downloading${emailCount.toLocaleString()} emails`,
total: emailCount,
label: `Download complete${emailCount.toLocaleString()} emails`,
});
}
})();
// Stream stderr for logging
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) {
// Gmail throttled us — don't retry, just import what we have
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;
// Fall through to import step
} 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(() => {});
}
} finally {
// Always clean up config (contains password)
await unlink(configPath).catch(() => {});
}
},
},
{
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);
@@ -367,12 +737,19 @@ const gmailSyncHandler: JobHandler = {
const db = openEmailDb(ctx.job.userId);
try {
const result = await importMaildir(maildirPath, emailAccount, db, (saved, skipped) => {
ctx.updateProgress({
current: saved + skipped,
total: 0,
label: `Saved ${saved.toLocaleString()}, skipped ${skipped.toLocaleString()}`,
});
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(
@@ -389,7 +766,6 @@ const gmailSyncHandler: JobHandler = {
if (dbUser) {
const paths = await getDockPaths(dbUser.id);
if (!paths) {
// User hasn't customized dock — initialize with defaults + /email
const defaults = ['/', '/files', '/automation', '/projects', '/dashboards', '/chat'];
await setDockPaths(dbUser.id, [...defaults, '/email']);
} else if (!paths.includes('/email')) {
+1
View File
@@ -1 +1,2 @@
import './gmail-sync';
import './email-sync';
+56
View File
@@ -0,0 +1,56 @@
import { getAllSyncedAccounts, getUserById } from 'officerdb';
import * as queueRunner from './queue-runner';
const INTERVAL_MS = 10 * 60 * 1000; // 10 minutes
let timer: ReturnType<typeof setInterval> | null = null;
async function tick() {
try {
const accounts = await getAllSyncedAccounts();
if (accounts.length === 0) return;
const allJobs = await queueRunner.listAllJobs();
const activeEmailSyncIds = new Set(
allJobs
.filter((j) => j.type === 'email-sync' && (j.status === 'queued' || j.status === 'running'))
.map((j) => (j.meta as Record<string, unknown> | undefined)?.emailAccountId),
);
for (const account of accounts) {
if (activeEmailSyncIds.has(account.id)) continue;
const user = await getUserById(account.userId);
if (!user) continue;
try {
await queueRunner.enqueue({
lane: 'email',
type: 'email-sync',
userId: user.email,
meta: { emailAccountId: account.id },
});
console.log(`[email-cron] Enqueued incremental sync for ${account.email}`);
} catch (err) {
console.error(`[email-cron] Failed to enqueue sync for ${account.email}:`, err instanceof Error ? err.message : err);
}
}
} catch (err) {
console.error('[email-cron] Error:', err instanceof Error ? err.message : err);
}
}
export function initEmailCron() {
if (timer) return;
console.log(`[email-cron] Starting email sync cron (every ${INTERVAL_MS / 60_000} min)`);
timer = setInterval(tick, INTERVAL_MS);
// Run first tick after a short delay to let the queue initialize
setTimeout(tick, 30_000);
}
export function stopEmailCron() {
if (timer) {
clearInterval(timer);
timer = null;
}
}
+5
View File
@@ -5,6 +5,7 @@ import { startAnthropicProxy, getProxySecret, ensureProxySecret } from './proxy'
import * as claudeManager from './claude-manager';
import * as piManager from './pi-manager';
import * as queueRunner from './queue-runner';
import { initEmailCron, stopEmailCron } from './email-cron';
const PORT = Number(process.env.SIDECAR_PORT ?? '5100');
const startedAt = Date.now();
@@ -31,6 +32,9 @@ queueRunner.initQueue().catch((err) => {
console.error('[sidecar] failed to initialize queue:', err);
});
// Start email sync cron
// initEmailCron(); // TODO: re-enable after initial sync testing
// ── WebSocket connections ──
const clients = new Set<ServerWebSocket<unknown>>();
@@ -251,6 +255,7 @@ console.log(`[sidecar] listening on 127.0.0.1:${PORT}`);
async function shutdown(signal: string) {
console.log(`[sidecar] ${signal} received, saving state...`);
stopEmailCron();
await flushAndSave();
releaseLock();
process.exit(0);