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.
708 lines
24 KiB
TypeScript
708 lines
24 KiB
TypeScript
import type { Database } from 'bun:sqlite';
|
|
import { createHash } from 'node:crypto';
|
|
import { openEmailDb, upsertFromRawEml, setSyncMeta, getSyncMeta } from './store';
|
|
import { getUserByEmail, getUserIntegration, upsertUserIntegration, getServerIntegration, getDockPaths, setDockPaths } from 'officerdb';
|
|
import { updateEmailAccountStatus, getEmailAccount } from '../db/queries';
|
|
import { refreshGoogleAccessToken } from '@@/api/integrations/google-auth';
|
|
|
|
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 {}
|
|
|
|
// ── Credentials ──
|
|
|
|
export type GmailCredentials = {
|
|
email: string;
|
|
userId: number;
|
|
appPassword?: string;
|
|
accessToken?: string;
|
|
refreshToken?: string;
|
|
expiresAt?: number;
|
|
};
|
|
|
|
export 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');
|
|
}
|
|
|
|
return { email: gmailEmail, userId: dbUser.id, appPassword: config.imapAppPassword as string };
|
|
}
|
|
|
|
// ── 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);
|
|
}
|
|
|
|
// ── Gmail API label → our label mapping ──
|
|
|
|
const GMAIL_API_LABEL_MAP: Record<string, string> = {
|
|
INBOX: 'inbox',
|
|
SENT: 'sent',
|
|
DRAFT: 'draft',
|
|
STARRED: 'starred',
|
|
IMPORTANT: 'important',
|
|
TRASH: 'trash',
|
|
SPAM: 'spam',
|
|
CATEGORY_PROMOTIONS: 'promotions',
|
|
CATEGORY_SOCIAL: 'social',
|
|
CATEGORY_UPDATES: 'updates',
|
|
CATEGORY_FORUMS: 'forums',
|
|
};
|
|
|
|
const SKIP_LABELS = new Set(['TRASH', 'SPAM', 'DRAFT']);
|
|
|
|
function gmailApiLabelsToLabels(labelIds: string[]): string[] {
|
|
const labels: string[] = [];
|
|
for (const id of labelIds) {
|
|
const mapped = GMAIL_API_LABEL_MAP[id];
|
|
if (mapped) {
|
|
labels.push(mapped);
|
|
} else if (!id.startsWith('CATEGORY_') && id !== 'UNREAD' && id !== 'IMPORTANT') {
|
|
// Custom label — use as-is (Gmail API returns label IDs, not names, for custom labels)
|
|
labels.push(id.toLowerCase());
|
|
}
|
|
}
|
|
return labels;
|
|
}
|
|
|
|
function shouldSkipMessage(labelIds: string[]): boolean {
|
|
return labelIds.some((id) => SKIP_LABELS.has(id));
|
|
}
|
|
|
|
// ── Gmail API helpers ──
|
|
|
|
const GMAIL_API = 'https://gmail.googleapis.com/gmail/v1/users/me';
|
|
|
|
export type GmailApiResult = { saved: number; skipped: number; errors: number };
|
|
|
|
async function gmailApiFetch(accessToken: string, path: string, params?: Record<string, string>): Promise<Response> {
|
|
const url = new URL(`${GMAIL_API}${path}`);
|
|
if (params) {
|
|
for (const [k, v] of Object.entries(params)) url.searchParams.set(k, v);
|
|
}
|
|
const res = await fetch(url, { headers: { Authorization: `Bearer ${accessToken}` } });
|
|
return res;
|
|
}
|
|
|
|
async function getGmailProfile(accessToken: string): Promise<{ historyId: string }> {
|
|
const res = await gmailApiFetch(accessToken, '/profile');
|
|
if (!res.ok) throw new Error(`Gmail profile request failed: ${res.status} ${await res.text()}`);
|
|
const data = (await res.json()) as { historyId: string };
|
|
return data;
|
|
}
|
|
|
|
type HistoryMessage = { id: string; threadId: string };
|
|
type HistoryEntry = { messagesAdded?: Array<{ message: HistoryMessage }> };
|
|
type HistoryResponse = { history?: HistoryEntry[]; historyId: string; nextPageToken?: string };
|
|
|
|
async function getHistoryChanges(
|
|
accessToken: string,
|
|
startHistoryId: string,
|
|
): Promise<{ messageIds: string[]; historyId: string }> {
|
|
const messageIds = new Set<string>();
|
|
let pageToken: string | undefined;
|
|
let latestHistoryId = startHistoryId;
|
|
|
|
while (true) {
|
|
const params: Record<string, string> = {
|
|
startHistoryId,
|
|
historyTypes: 'messageAdded',
|
|
maxResults: '500',
|
|
};
|
|
if (pageToken) params.pageToken = pageToken;
|
|
|
|
const res = await gmailApiFetch(accessToken, '/history', params);
|
|
|
|
if (res.status === 404 || res.status === 410) {
|
|
// historyId too old or invalid — caller should fall back to date-based listing
|
|
throw new HistoryExpiredError();
|
|
}
|
|
|
|
if (!res.ok) throw new Error(`Gmail history request failed: ${res.status} ${await res.text()}`);
|
|
|
|
const data = (await res.json()) as HistoryResponse;
|
|
latestHistoryId = data.historyId;
|
|
|
|
if (data.history) {
|
|
for (const entry of data.history) {
|
|
if (entry.messagesAdded) {
|
|
for (const added of entry.messagesAdded) {
|
|
messageIds.add(added.message.id);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
if (!data.nextPageToken) break;
|
|
pageToken = data.nextPageToken;
|
|
}
|
|
|
|
return { messageIds: [...messageIds], historyId: latestHistoryId };
|
|
}
|
|
|
|
class HistoryExpiredError extends Error {
|
|
constructor() {
|
|
super('Gmail historyId expired');
|
|
}
|
|
}
|
|
|
|
type MessageListResponse = { messages?: Array<{ id: string }>; nextPageToken?: string };
|
|
|
|
async function listMessagesSince(accessToken: string, sinceDate: string): Promise<string[]> {
|
|
const messageIds: string[] = [];
|
|
let pageToken: string | undefined;
|
|
|
|
while (true) {
|
|
const params: Record<string, string> = {
|
|
q: `after:${sinceDate}`,
|
|
maxResults: '500',
|
|
};
|
|
if (pageToken) params.pageToken = pageToken;
|
|
|
|
const res = await gmailApiFetch(accessToken, '/messages', params);
|
|
if (!res.ok) throw new Error(`Gmail messages.list failed: ${res.status} ${await res.text()}`);
|
|
|
|
const data = (await res.json()) as MessageListResponse;
|
|
if (data.messages) {
|
|
for (const msg of data.messages) messageIds.push(msg.id);
|
|
}
|
|
|
|
if (!data.nextPageToken) break;
|
|
pageToken = data.nextPageToken;
|
|
}
|
|
|
|
return messageIds;
|
|
}
|
|
|
|
type GmailMessage = { id: string; labelIds: string[]; raw: string; historyId: string };
|
|
|
|
async function fetchRawMessage(accessToken: string, messageId: string): Promise<GmailMessage | null> {
|
|
const res = await gmailApiFetch(accessToken, `/messages/${messageId}`, { format: 'raw' });
|
|
if (res.status === 404) return null;
|
|
if (!res.ok) throw new Error(`Gmail messages.get failed: ${res.status} ${await res.text()}`);
|
|
return (await res.json()) as GmailMessage;
|
|
}
|
|
|
|
// ── Gmail API sync (for resyncs) ──
|
|
|
|
export type GmailApiSyncParams = {
|
|
creds: GmailCredentials;
|
|
db: Database;
|
|
onProgress?: (saved: number, total: number) => void;
|
|
};
|
|
|
|
export async function gmailApiSync({ creds, db, onProgress }: GmailApiSyncParams): Promise<GmailApiResult> {
|
|
if (!creds.accessToken) throw new PermanentError('OAuth access token required for Gmail API sync');
|
|
|
|
let saved = 0;
|
|
let skipped = 0;
|
|
let errors = 0;
|
|
|
|
// Load existing IDs for dedup
|
|
const existingIds = new Set<string>();
|
|
const rows = db.query('SELECT id FROM emails').all() as Array<{ id: string }>;
|
|
for (const row of rows) existingIds.add(row.id);
|
|
|
|
const storedHistoryId = getSyncMeta(db, 'gmail_history_id');
|
|
|
|
let messageIds: string[];
|
|
let newHistoryId: string;
|
|
|
|
if (storedHistoryId) {
|
|
// Try history-based sync first
|
|
try {
|
|
const result = await getHistoryChanges(creds.accessToken, storedHistoryId);
|
|
messageIds = result.messageIds;
|
|
newHistoryId = result.historyId;
|
|
console.log(`[gmail-sync] History returned ${messageIds.length} new message(s)`);
|
|
} catch (err) {
|
|
if (err instanceof HistoryExpiredError) {
|
|
// historyId too old — fall back to date-based listing
|
|
const lastSyncDate = getSyncMeta(db, 'last_sync_date');
|
|
const sinceDate = lastSyncDate ?? new Date(Date.now() - 7 * 24 * 60 * 60 * 1000).toISOString().split('T')[0]!;
|
|
console.log(`[gmail-sync] History expired, falling back to messages since ${sinceDate}`);
|
|
messageIds = await listMessagesSince(creds.accessToken, sinceDate);
|
|
const profile = await getGmailProfile(creds.accessToken);
|
|
newHistoryId = profile.historyId;
|
|
console.log(`[gmail-sync] Date query returned ${messageIds.length} message(s)`);
|
|
} else {
|
|
throw err;
|
|
}
|
|
}
|
|
} else {
|
|
// No historyId stored — use date-based listing
|
|
const lastSyncDate = getSyncMeta(db, 'last_sync_date');
|
|
const sinceDate = lastSyncDate ?? new Date(Date.now() - 7 * 24 * 60 * 60 * 1000).toISOString().split('T')[0]!;
|
|
console.log(`[gmail-sync] No historyId, listing messages since ${sinceDate}`);
|
|
messageIds = await listMessagesSince(creds.accessToken, sinceDate);
|
|
const profile = await getGmailProfile(creds.accessToken);
|
|
newHistoryId = profile.historyId;
|
|
console.log(`[gmail-sync] Date query returned ${messageIds.length} message(s)`);
|
|
}
|
|
|
|
// Fetch and store each new message
|
|
for (let i = 0; i < messageIds.length; i++) {
|
|
const gmailId = messageIds[i]!;
|
|
|
|
try {
|
|
const msg = await fetchRawMessage(creds.accessToken, gmailId);
|
|
if (!msg) {
|
|
skipped++;
|
|
continue;
|
|
}
|
|
|
|
if (shouldSkipMessage(msg.labelIds ?? [])) {
|
|
skipped++;
|
|
continue;
|
|
}
|
|
|
|
// Gmail API returns base64url-encoded raw RFC822
|
|
const rawBytes = Buffer.from(msg.raw, 'base64url');
|
|
const raw = rawBytes.toString('utf-8');
|
|
|
|
const id = messageIdToStableId(raw);
|
|
if (!id) {
|
|
errors++;
|
|
continue;
|
|
}
|
|
|
|
if (existingIds.has(id)) {
|
|
skipped++;
|
|
continue;
|
|
}
|
|
|
|
const labels = gmailApiLabelsToLabels(msg.labelIds ?? []);
|
|
|
|
upsertFromRawEml({ db, id, raw, integration: 'gmail', emailAccount: creds.email, labels });
|
|
existingIds.add(id);
|
|
saved++;
|
|
|
|
if ((saved + skipped) % 50 === 0) {
|
|
onProgress?.(saved, messageIds.length);
|
|
}
|
|
} catch (err) {
|
|
console.log(`[gmail-sync] Error fetching message ${gmailId}: ${err instanceof Error ? err.message : err}`);
|
|
errors++;
|
|
}
|
|
}
|
|
|
|
// Persist new historyId
|
|
setSyncMeta(db, 'gmail_history_id', newHistoryId);
|
|
|
|
return { saved, skipped, errors };
|
|
}
|
|
|
|
// ── IMAP sync (for first sync) ──
|
|
|
|
// Folder → label mapping for IMAP
|
|
const GMAIL_PREFIX_RE = /^\[(?:Gmail|Google Mail)\]\//;
|
|
|
|
function normalizeGmailFolder(folder: string): string {
|
|
return folder.replace(GMAIL_PREFIX_RE, '');
|
|
}
|
|
|
|
const SUFFIX_LABEL_MAP: Record<string, string> = {
|
|
'Sent Mail': 'sent',
|
|
Drafts: 'draft',
|
|
Starred: 'starred',
|
|
Important: 'important',
|
|
'All Mail': 'archive',
|
|
Trash: 'trash',
|
|
Bin: 'trash',
|
|
Spam: 'spam',
|
|
};
|
|
|
|
function folderToLabel(folder: string): string | null {
|
|
const suffix = normalizeGmailFolder(folder);
|
|
if (suffix !== folder) {
|
|
if (SUFFIX_LABEL_MAP[suffix]) return SUFFIX_LABEL_MAP[suffix]!;
|
|
return suffix.toLowerCase();
|
|
}
|
|
if (folder === 'INBOX') return 'inbox';
|
|
return folder.toLowerCase();
|
|
}
|
|
|
|
const IMAP_LABEL_MAP: Record<string, string> = {
|
|
'\\Inbox': 'inbox',
|
|
'\\Sent': 'sent',
|
|
'\\Drafts': 'draft',
|
|
'\\Starred': 'starred',
|
|
'\\Important': 'important',
|
|
'\\All': 'archive',
|
|
'\\Trash': 'trash',
|
|
'\\Junk': 'spam',
|
|
};
|
|
|
|
function imapLabelsToLabels(gmailLabels: Set<string>): string[] {
|
|
const labels: string[] = [];
|
|
for (const gl of gmailLabels) {
|
|
const mapped = IMAP_LABEL_MAP[gl];
|
|
if (mapped) {
|
|
labels.push(mapped);
|
|
} else if (!gl.startsWith('\\')) {
|
|
labels.push(gl.toLowerCase());
|
|
}
|
|
}
|
|
if (labels.length > 1 && labels.includes('archive')) {
|
|
return labels.filter((l) => l !== 'archive');
|
|
}
|
|
return labels;
|
|
}
|
|
|
|
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';
|
|
}
|
|
|
|
type ImapSyncParams = {
|
|
creds: GmailCredentials;
|
|
db: Database;
|
|
onProgress?: (fetched: number, folder: string) => void;
|
|
};
|
|
|
|
type SyncResult = { saved: number; skipped: number; errors: number };
|
|
|
|
async function imapFullSync({ creds, db, onProgress }: ImapSyncParams): Promise<SyncResult> {
|
|
const { ImapFlow } = await import('imapflow');
|
|
|
|
const auth: { user: string; pass?: string; accessToken?: string } = { user: creds.email };
|
|
if (creds.appPassword) {
|
|
auth.pass = creds.appPassword;
|
|
} else if (creds.accessToken) {
|
|
auth.accessToken = creds.accessToken;
|
|
} else {
|
|
throw new PermanentError('No authentication method available for IMAP');
|
|
}
|
|
|
|
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] IMAP connected for full sync');
|
|
|
|
const folders = await client.list({ statusQuery: { uidNext: true, uidValidity: true } });
|
|
|
|
const existingIds = new Set<string>();
|
|
const rows = db.query('SELECT id FROM emails').all() as Array<{ id: string }>;
|
|
for (const row of rows) existingIds.add(row.id);
|
|
|
|
const 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;
|
|
|
|
foldersToSync.push(folder);
|
|
}
|
|
|
|
console.log(`[gmail-sync] ${foldersToSync.length} folder(s) to sync`);
|
|
|
|
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);
|
|
|
|
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;
|
|
}
|
|
|
|
const folderLabel = folderToLabel(folderPath);
|
|
const gmailLabels = msg.labels ? imapLabelsToLabels(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 = {
|
|
type: 'gmail-sync',
|
|
retry: { delayMs: 15 * 60 * 1000, maxRetries: 10 },
|
|
steps: [
|
|
{
|
|
name: 'Verify credentials',
|
|
run: async (ctx: SyncCtx) => {
|
|
const creds = await loadGmailCredentials(ctx.job.userId);
|
|
const emailAccountId = (ctx.meta as Record<string, unknown>).emailAccountId as number | undefined;
|
|
const syncAccount = emailAccountId ? await getEmailAccount(emailAccountId) : undefined;
|
|
|
|
// Determine sync mode — check SQLite first, fall back to PostgreSQL syncMeta
|
|
const db = openEmailDb(ctx.job.userId, syncAccount?.email ?? ctx.job.userId);
|
|
let lastSyncAt: string | null = null;
|
|
try {
|
|
lastSyncAt = getSyncMeta(db, 'last_sync_at');
|
|
if (!lastSyncAt && emailAccountId) {
|
|
const pgAccount = await getEmailAccount(emailAccountId);
|
|
const pgMeta = pgAccount?.syncMeta as Record<string, string> | null;
|
|
if (pgMeta?.last_sync_at) {
|
|
lastSyncAt = pgMeta.last_sync_at;
|
|
// Backfill SQLite so future checks don't need PostgreSQL
|
|
setSyncMeta(db, 'last_sync_at', lastSyncAt);
|
|
if (pgMeta.last_sync_date) setSyncMeta(db, 'last_sync_date', pgMeta.last_sync_date);
|
|
}
|
|
}
|
|
ctx.meta.isFirstSync = !lastSyncAt;
|
|
} finally {
|
|
db.close();
|
|
}
|
|
|
|
if (emailAccountId) {
|
|
await updateEmailAccountStatus(emailAccountId, 'syncing');
|
|
ctx.meta.emailAccountId = emailAccountId;
|
|
}
|
|
|
|
// For API resyncs, refresh OAuth token if expired
|
|
if (!ctx.meta.isFirstSync && creds.accessToken && creds.refreshToken) {
|
|
const tokenExpired = !creds.expiresAt || creds.expiresAt < Date.now() + 60_000;
|
|
if (tokenExpired) {
|
|
console.log('[gmail-sync] Refreshing OAuth token');
|
|
try {
|
|
const refreshed = await refreshGoogleAccessToken(creds.refreshToken);
|
|
creds.accessToken = refreshed.accessToken;
|
|
creds.expiresAt = refreshed.expiresAt;
|
|
|
|
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) {
|
|
throw new PermanentError(`OAuth refresh failed — reconnect Google in Settings: ${err}`);
|
|
}
|
|
}
|
|
}
|
|
|
|
if (!ctx.meta.isFirstSync && !creds.accessToken) {
|
|
throw new PermanentError('OAuth not configured — connect Google in Settings → Integrations for resyncs');
|
|
}
|
|
|
|
ctx.meta.creds = creds;
|
|
console.log(
|
|
`[gmail-sync] ${ctx.meta.isFirstSync ? 'First sync (IMAP)' : 'Resync (Gmail API)'} — ${creds.accessToken ? 'OAuth' : 'App Password'}`,
|
|
);
|
|
},
|
|
},
|
|
{
|
|
name: 'Sync emails',
|
|
run: async (ctx: SyncCtx) => {
|
|
const creds = ctx.meta.creds as GmailCredentials;
|
|
|
|
const emailAccountId = (ctx.meta as Record<string, unknown>).emailAccountId as number | undefined;
|
|
const syncAccount = emailAccountId ? await getEmailAccount(emailAccountId) : undefined;
|
|
const db = openEmailDb(ctx.job.userId, syncAccount?.email ?? ctx.job.userId);
|
|
try {
|
|
let result: SyncResult;
|
|
|
|
if (ctx.meta.isFirstSync) {
|
|
// First sync: IMAP with app password
|
|
await ctx.updateProgress({ current: 0, total: 0, label: 'Connecting to Gmail (IMAP)...' });
|
|
|
|
result = await imapFullSync({
|
|
creds,
|
|
db,
|
|
onProgress: (fetched, folder) => {
|
|
ctx.updateProgress({ current: fetched, total: 0, label: `Syncing ${folder}...` });
|
|
},
|
|
});
|
|
|
|
// Seed historyId for future API syncs
|
|
if (creds.accessToken) {
|
|
try {
|
|
const profile = await getGmailProfile(creds.accessToken);
|
|
setSyncMeta(db, 'gmail_history_id', profile.historyId);
|
|
console.log(`[gmail-sync] Seeded historyId: ${profile.historyId}`);
|
|
} catch (err) {
|
|
console.log(`[gmail-sync] Could not seed historyId: ${err}`);
|
|
}
|
|
}
|
|
} else {
|
|
// Resync: Gmail API with OAuth
|
|
await ctx.updateProgress({ current: 0, total: 0, label: 'Checking for new emails...' });
|
|
|
|
result = await gmailApiSync({
|
|
creds,
|
|
db,
|
|
onProgress: (saved, total) => {
|
|
ctx.updateProgress({ current: saved, total, label: `Fetching new emails (${saved}/${total})...` });
|
|
},
|
|
});
|
|
}
|
|
|
|
console.log(`[gmail-sync] Done: saved ${result.saved}, skipped ${result.skipped}, errors ${result.errors}`);
|
|
|
|
setSyncMeta(db, 'last_sync_date', new Date().toISOString().split('T')[0]!);
|
|
setSyncMeta(db, 'last_sync_at', new Date().toISOString());
|
|
|
|
ctx.meta.syncResult = result;
|
|
await ctx.updateProgress({ current: 1, total: 1, label: `Done — ${result.saved} new emails` });
|
|
} finally {
|
|
db.close();
|
|
}
|
|
},
|
|
},
|
|
{
|
|
name: 'Finalize',
|
|
run: async (ctx: SyncCtx) => {
|
|
const result = ctx.meta.syncResult as SyncResult | undefined;
|
|
const emailAccountId = ctx.meta.emailAccountId as number | undefined;
|
|
|
|
if (emailAccountId) {
|
|
await updateEmailAccountStatus(emailAccountId, 'synced');
|
|
}
|
|
|
|
if (result && result.saved > 0) {
|
|
try {
|
|
const dbUser = await getUserByEmail(ctx.job.userId);
|
|
if (dbUser) {
|
|
const paths = await getDockPaths(dbUser.id);
|
|
if (!paths) {
|
|
const defaults = ['/', '/files', '/automation', '/dashboards', '/chat'];
|
|
await setDockPaths(dbUser.id, [...defaults, '/email']);
|
|
} else if (!paths.includes('/email')) {
|
|
await setDockPaths(dbUser.id, [...paths, '/email']);
|
|
}
|
|
}
|
|
} catch {
|
|
// Non-fatal
|
|
}
|
|
}
|
|
},
|
|
},
|
|
],
|
|
};
|
|
|
|
/** The handler's steps, for the sidecar's own runner. */
|
|
export const gmailSyncSteps: SyncStep[] = gmailSyncHandler.steps as SyncStep[];
|