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

email syncyng

Browse Source
This commit is contained in:
pastilhas
2026-02-26 01:39:38 +00:00
parent 5d4f0114cd
commit 42abb97d7b
22 changed files with 1422 additions and 348 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/servers/queue/handlers/gmail-sync.ts
+261 -196
View File
@@ -1,10 +1,10 @@
import { mkdirSync, readFileSync, readdirSync, writeFileSync } from 'node:fs';
import { join } from 'node:path';
import { homedir } from 'node:os';
import type { EmailSummary } from 'types';
import type { Database } from 'bun:sqlite';
import type { JobHandler } from '../types';
import { registerHandler } from '../handler-registry';
import { DATA_PATH } from '../../data-path';
import { openEmailDb, upsertFromRawEml, getSyncMeta, setSyncMeta, updateEmailLabels } from '../../api/email/email-db';
type GoogleCredentials = {
accessToken: string;
@@ -97,176 +97,58 @@ async function gmailGet(token: string, path: string, params?: Record<string, str
return res.json();
}
function slugify(text: string, maxLen = 60): string {
return text
.toLowerCase()
.replace(/[^a-z0-9]+/g, '-')
.replace(/^-+|-+$/g, '')
.slice(0, maxLen)
.replace(/-+$/, '');
// ── Pre-flight message count ──
async function countMessages(token: string, query?: string): Promise<number> {
let count = 0;
let pageToken: string | undefined;
do {
const params: Record<string, string> = { maxResults: '500' };
if (query) params.q = query;
if (pageToken) params.pageToken = pageToken;
const list = (await gmailGet(token, '/messages', params)) as {
messages?: Array<{ id: string }>;
nextPageToken?: string;
};
count += list.messages?.length ?? 0;
if (!list.messages?.length) break;
pageToken = list.nextPageToken;
} while (pageToken);
return count;
}
function buildEmlFilename(id: string, internalDate: string | undefined, rawEmail: string): string {
const subjectMatch = rawEmail.match(/^Subject:\s*(.+)$/mi);
const subject = subjectMatch?.[1]?.trim() || 'no-subject';
const ts = parseInt(internalDate || '0');
const d = new Date(ts);
const dateStr =
ts > 0
? `${d.getFullYear()}-${String(d.getMonth() + 1).padStart(2, '0')}-${String(d.getDate()).padStart(2, '0')}`
: 'unknown-date';
return `${dateStr}_${slugify(subject)}_${id}.eml`;
}
function decodeMimeWords(text: string): string {
return text.replace(/=\?([^?]+)\?(B|Q)\?([^?]+)\?=/gi, (_, _charset, encoding, encoded) => {
try {
if (encoding.toUpperCase() === 'B') {
return Buffer.from(encoded, 'base64').toString('utf-8');
}
const bytes: number[] = [];
for (let i = 0; i < encoded.length; i++) {
if (encoded[i] === '_') {
bytes.push(0x20);
} else if (encoded[i] === '=' && i + 2 < encoded.length) {
bytes.push(parseInt(encoded.slice(i + 1, i + 3), 16));
i += 2;
} else {
bytes.push(encoded.charCodeAt(i));
}
}
return Buffer.from(bytes).toString('utf-8');
} catch {
return encoded;
}
});
}
function extractHeader(raw: string, name: string): string {
const match = raw.match(new RegExp(`^${name}:\\s*(.+)$`, 'mi'));
return match?.[1]?.trim() ? decodeMimeWords(match[1].trim()) : '';
}
function findHeaderEnd(text: string): number {
const crlf = text.indexOf('\r\n\r\n');
const lf = text.indexOf('\n\n');
if (crlf !== -1) return crlf + 4;
if (lf !== -1) return lf + 2;
return -1;
}
function extractSnippet(raw: string): string {
const idx = findHeaderEnd(raw);
if (idx === -1) return '';
let body = raw.slice(idx);
// If multipart, skip boundary line + part headers to reach actual content
if (body.trimStart().startsWith('--')) {
const afterBoundary = body.slice(body.indexOf('\n') + 1);
const partBodyStart = findHeaderEnd(afterBoundary);
if (partBodyStart !== -1) body = afterBoundary.slice(partBodyStart);
}
// Stop at next MIME boundary
const nextBoundary = body.indexOf('\n--');
if (nextBoundary !== -1) body = body.slice(0, nextBoundary);
return body.replace(/\s+/g, ' ').trim().slice(0, 120);
}
function countAttachments(raw: string): number {
const matches = raw.match(/^Content-Disposition:\s*attachment/gim);
return matches?.length ?? 0;
}
function parseEmlToSummary(raw: string, id: string): EmailSummary {
const dateStr = extractHeader(raw, 'Date');
const date = dateStr ? new Date(dateStr).toISOString() : new Date(0).toISOString();
const attachmentCount = countAttachments(raw);
return {
id,
from: extractHeader(raw, 'From'),
to: extractHeader(raw, 'To'),
subject: extractHeader(raw, 'Subject') || '(no subject)',
date,
snippet: extractSnippet(raw),
...(attachmentCount > 0 ? { attachmentCount } : {}),
};
}
type EmailIndex = { v: number; entries: EmailSummary[] };
const INDEX_VERSION = 3;
export function rebuildIndex(emailsDir: string): EmailSummary[] {
let filenames: string[];
try {
filenames = readdirSync(emailsDir).filter((f) => f.endsWith('.eml'));
} catch {
return [];
}
const indexPath = join(emailsDir, 'index.json');
const existing = new Map<string, EmailSummary>();
try {
const raw = JSON.parse(readFileSync(indexPath, 'utf-8'));
const index = (Array.isArray(raw) ? null : raw) as EmailIndex | null;
if (index?.v === INDEX_VERSION) {
for (const entry of index.entries) {
existing.set(entry.id, entry);
}
}
} catch {
/* no existing index or corrupted */
}
const entries: EmailSummary[] = [];
for (const filename of filenames) {
const id = filename.replace(/\.eml$/, '');
const cached = existing.get(id);
if (cached) {
entries.push(cached);
} else {
try {
const raw = readFileSync(join(emailsDir, filename), 'utf-8');
entries.push(parseEmlToSummary(raw, id));
} catch {
/* skip unreadable files */
}
}
}
entries.sort((a, b) => new Date(b.date).getTime() - new Date(a.date).getTime());
writeFileSync(indexPath, JSON.stringify({ v: INDEX_VERSION, entries }));
return entries;
}
// ── Full sync: Gmail API → SQLite directly ──
type SyncProgress = { saved: number; skipped: number; errors: number; page: number };
type OnProgress = (progress: SyncProgress) => void;
type SyncResult = {
saved: number;
skipped: number;
errors: number;
labelMap: Map<string, string[]>;
maxHistoryId: string | null;
};
async function syncInbox(
token: string,
outputDir: string,
db: Database,
emailAccount: string,
query?: string,
onProgress?: OnProgress,
): Promise<{ saved: number; skipped: number; errors: number }> {
mkdirSync(outputDir, { recursive: true });
): Promise<SyncResult> {
// Dedup via DB
const existingIds = new Set<string>();
try {
for (const file of readdirSync(outputDir)) {
const match = file.match(/_([a-f0-9]+)\.eml$/i);
if (match) existingIds.add(match[1]!);
}
} catch {
/* dir might not exist yet */
}
const rows = db.query('SELECT id FROM emails').all() as Array<{ id: string }>;
for (const row of rows) existingIds.add(row.id);
let saved = 0;
let skipped = 0;
let errors = 0;
let page = 0;
let pageToken: string | undefined;
const labelMap = new Map<string, string[]>();
let maxHistoryId: bigint | null = null;
do {
const params: Record<string, string> = { maxResults: '100' };
@@ -294,12 +176,19 @@ async function syncInbox(
id: string;
internalDate?: string;
raw: string;
labelIds?: string[];
historyId?: string;
};
const rawEmail = Buffer.from(msg.raw, 'base64url').toString('utf-8');
const filename = buildEmlFilename(msg.id, msg.internalDate, rawEmail);
writeFileSync(join(outputDir, filename), rawEmail);
upsertFromRawEml({ db, id, raw: rawEmail, integration: 'gmail', emailAccount, labels: msg.labelIds });
existingIds.add(id);
saved++;
if (msg.labelIds) labelMap.set(id, msg.labelIds);
if (msg.historyId) {
const hid = BigInt(msg.historyId);
if (maxHistoryId === null || hid > maxHistoryId) maxHistoryId = hid;
}
} catch {
errors++;
}
@@ -314,7 +203,121 @@ async function syncInbox(
pageToken = list.nextPageToken;
} while (pageToken);
return { saved, skipped, errors };
return { saved, skipped, errors, labelMap, maxHistoryId: maxHistoryId !== null ? String(maxHistoryId) : null };
}
// ── Gmail History API (incremental sync) ──
type HistoryMessage = { id: string; labelIds?: string[] };
type HistoryRecord = {
id: string;
messagesAdded?: Array<{ message: HistoryMessage }>;
messagesDeleted?: Array<{ message: HistoryMessage }>;
labelsAdded?: Array<{ message: HistoryMessage; labelIds: string[] }>;
labelsRemoved?: Array<{ message: HistoryMessage; labelIds: string[] }>;
};
type HistoryResponse = {
history?: HistoryRecord[];
nextPageToken?: string;
historyId: string;
};
type IncrementalResult =
| { stale: false; added: number; deleted: number; relabeled: number; maxHistoryId: string }
| { stale: true };
async function syncIncremental(
token: string,
db: Database,
lastHistoryId: string,
emailAccount: string,
): Promise<IncrementalResult> {
let pageToken: string | undefined;
let added = 0;
let deleted = 0;
let relabeled = 0;
let latestHistoryId = lastHistoryId;
do {
const url = new URL(`${GMAIL_BASE}/history`);
url.searchParams.set('startHistoryId', lastHistoryId);
for (const ht of ['messageAdded', 'messageDeleted', 'labelAdded', 'labelRemoved']) {
url.searchParams.append('historyTypes', ht);
}
if (pageToken) url.searchParams.set('pageToken', pageToken);
let data: HistoryResponse;
try {
const res = await fetch(url.toString(), { headers: { Authorization: `Bearer ${token}` } });
if (!res.ok) {
const error = await res.text().catch(() => '');
throw new Error(`Gmail API error (${res.status}): ${error}`);
}
data = (await res.json()) as HistoryResponse;
} catch (err) {
// historyId too old — Gmail returns 404
if (err instanceof Error && err.message.includes('404')) {
return { stale: true };
}
throw err;
}
latestHistoryId = data.historyId;
for (const record of data.history ?? []) {
// New messages
for (const { message } of record.messagesAdded ?? []) {
try {
const msg = (await gmailGet(token, `/messages/${message.id}`, { format: 'raw' })) as {
id: string;
internalDate?: string;
raw: string;
labelIds?: string[];
historyId?: string;
};
const rawEmail = Buffer.from(msg.raw, 'base64url').toString('utf-8');
upsertFromRawEml({ db, id: msg.id, raw: rawEmail, integration: 'gmail', emailAccount, labels: msg.labelIds });
added++;
} catch {
/* skip individual failures */
}
}
// Deleted messages
for (const { message } of record.messagesDeleted ?? []) {
db.run('UPDATE emails SET deleted = 1 WHERE id = ?', [message.id]);
deleted++;
}
// Labels added
for (const { message, labelIds } of record.labelsAdded ?? []) {
const row = db.query('SELECT labels FROM emails WHERE id = ?').get(message.id) as { labels: string | null } | null;
if (row) {
const existing = row.labels ? row.labels.split(',') : [];
const merged = [...new Set([...existing, ...labelIds.map((l) => l.toLowerCase())])];
updateEmailLabels(db, message.id, merged);
relabeled++;
}
}
// Labels removed
for (const { message, labelIds } of record.labelsRemoved ?? []) {
const row = db.query('SELECT labels FROM emails WHERE id = ?').get(message.id) as { labels: string | null } | null;
if (row) {
const removeSet = new Set(labelIds.map((l) => l.toLowerCase()));
const remaining = (row.labels ? row.labels.split(',') : []).filter((l) => !removeSet.has(l));
updateEmailLabels(db, message.id, remaining);
relabeled++;
}
}
}
pageToken = data.nextPageToken;
} while (pageToken);
return { stale: false, added, deleted, relabeled, maxHistoryId: latestHistoryId };
}
const MONTH_NAMES = ['January', 'February', 'March', 'April', 'May', 'June', 'July', 'August', 'September', 'October', 'November', 'December'];
@@ -341,55 +344,117 @@ const gmailSyncHandler: JobHandler = {
run: async (ctx) => {
const creds = await loadCredentials(ctx.job.userId);
const token = await getValidAccessToken(creds);
// Store token in shared meta for the next step
ctx.meta.accessToken = token;
ctx.meta.outputDir = join(DATA_PATH, ctx.job.userId, 'Gmail', 'emails');
},
},
{
name: 'Sync emails',
run: async (ctx) => {
const token = ctx.meta.accessToken as string;
const outputDir = ctx.meta.outputDir as string;
const year = ctx.meta.year as number | undefined;
let totalSaved = 0;
let totalSkipped = 0;
let totalErrors = 0;
if (year) {
// Year-scoped sync: month by month with progress
const months = buildMonthRanges(year);
for (let i = 0; i < months.length; i++) {
const month = months[i]!;
await ctx.updateProgress({ current: i, total: months.length, label: month.label });
const query = `after:${month.after} before:${month.before}`;
const { saved, skipped, errors } = await syncInbox(token, outputDir, query, (p) => {
const label = `${month.label} — ${p.saved} saved`;
ctx.updateProgress({ current: i, total: months.length, label });
});
totalSaved += saved;
totalSkipped += skipped;
totalErrors += errors;
const db = openEmailDb(ctx.job.userId);
try {
// Try incremental sync first (only for non-year-scoped syncs)
if (!year) {
const lastHistoryId = getSyncMeta(db, 'last_history_id');
if (lastHistoryId) {
await ctx.updateProgress({ current: 0, total: 0, label: 'Incremental sync' });
console.log(`[gmail-sync] Attempting incremental sync from historyId ${lastHistoryId}`);
const result = await syncIncremental(token, db, lastHistoryId, ctx.job.userId);
if (!result.stale) {
setSyncMeta(db, 'last_history_id', result.maxHistoryId);
setSyncMeta(db, 'last_sync_at', new Date().toISOString());
console.log(`[gmail-sync] Incremental: +${result.added} added, -${result.deleted} deleted, ~${result.relabeled} relabeled`);
await ctx.updateProgress({ current: 1, total: 1, label: 'Done (incremental)' });
return;
}
console.log('[gmail-sync] historyId stale, falling back to full sync');
}
}
await ctx.updateProgress({ current: months.length, total: months.length, label: 'Done' });
} else {
// Full sync: all emails with per-page progress
await ctx.updateProgress({ current: 0, total: 0, label: 'Starting sync' });
const { saved, skipped, errors } = await syncInbox(token, outputDir, undefined, (p) => {
const label = `Saved ${p.saved}, skipped ${p.skipped} (page ${p.page})`;
ctx.updateProgress({ current: p.saved + p.skipped + p.errors, total: 0, label });
});
totalSaved = saved;
totalSkipped = skipped;
totalErrors = errors;
await ctx.updateProgress({ current: 1, total: 1, label: 'Done' });
// Full sync
let totalSaved = 0;
let totalSkipped = 0;
let totalErrors = 0;
let allLabelMaps: Map<string, string[]> = new Map();
let maxHistoryId: string | null = null;
if (year) {
// Year-scoped sync: count total emails first, then sync month by month
const yearQuery = `after:${year}/1/1 before:${year + 1}/1/1`;
await ctx.updateProgress({ current: 0, total: 0, label: 'Counting emails...' });
const totalEmails = await countMessages(token, yearQuery);
console.log(`[gmail-sync] Pre-flight: ${totalEmails} emails for ${year}`);
const months = buildMonthRanges(year);
for (let i = 0; i < months.length; i++) {
const month = months[i]!;
await ctx.updateProgress({ current: totalSaved + totalSkipped, total: totalEmails, label: month.label });
const query = `after:${month.after} before:${month.before}`;
const result = await syncInbox(token, db, ctx.job.userId, query, (p) => {
const current = totalSaved + p.saved + p.skipped + p.errors;
const label = `${month.label} — Saved ${(totalSaved + p.saved).toLocaleString()} of ${totalEmails.toLocaleString()}`;
ctx.updateProgress({ current, total: totalEmails, label });
});
totalSaved += result.saved;
totalSkipped += result.skipped;
totalErrors += result.errors;
for (const [id, labels] of result.labelMap) allLabelMaps.set(id, labels);
if (result.maxHistoryId) {
if (!maxHistoryId || BigInt(result.maxHistoryId) > BigInt(maxHistoryId)) {
maxHistoryId = result.maxHistoryId;
}
}
}
await ctx.updateProgress({ current: totalEmails, total: totalEmails, label: 'Done' });
} else {
// If we have a last_sync_date (e.g. from migration), scope the sync to only newer emails
const lastSyncDate = getSyncMeta(db, 'last_sync_date');
let syncQuery: string | undefined;
if (lastSyncDate) {
const d = new Date(lastSyncDate);
syncQuery = `after:${d.getFullYear()}/${d.getMonth() + 1}/${d.getDate()}`;
console.log(`[gmail-sync] Scoping full sync with query: ${syncQuery}`);
}
// Count total emails for accurate progress
await ctx.updateProgress({ current: 0, total: 0, label: 'Counting emails...' });
const totalEmails = await countMessages(token, syncQuery);
console.log(`[gmail-sync] Pre-flight: ${totalEmails} emails`);
await ctx.updateProgress({ current: 0, total: totalEmails, label: 'Starting sync' });
const result = await syncInbox(token, db, ctx.job.userId, syncQuery, (p) => {
const current = p.saved + p.skipped + p.errors;
const label = `Saved ${p.saved.toLocaleString()} of ${totalEmails.toLocaleString()}`;
ctx.updateProgress({ current, total: totalEmails, label });
});
totalSaved = result.saved;
totalSkipped = result.skipped;
totalErrors = result.errors;
allLabelMaps = result.labelMap;
maxHistoryId = result.maxHistoryId;
await ctx.updateProgress({ current: totalEmails, total: totalEmails, label: 'Done' });
}
console.log(`[gmail-sync] Full: saved ${totalSaved}, skipped ${totalSkipped}, errors ${totalErrors}`);
// Update labels for emails that were skipped but have new label data
let labelsUpdated = 0;
for (const [id, labels] of allLabelMaps) {
updateEmailLabels(db, id, labels);
labelsUpdated++;
}
if (labelsUpdated > 0) console.log(`[gmail-sync] Updated labels for ${labelsUpdated} emails`);
// Store sync state
if (maxHistoryId) {
setSyncMeta(db, 'last_history_id', maxHistoryId);
}
setSyncMeta(db, 'last_sync_at', new Date().toISOString());
} finally {
db.close();
}
console.log(`[gmail-sync] Saved ${totalSaved}, skipped ${totalSkipped}, errors ${totalErrors}`);
rebuildIndex(outputDir);
console.log('[gmail-sync] Index rebuilt');
},
},
],