email, extracted from the officer platform

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.
This commit is contained in:
Claude Opus 5
2026-08-18 00:29:50 +00:00
commit 980a9d3c38
30 changed files with 5616 additions and 0 deletions
+258
View File
@@ -0,0 +1,258 @@
import { createRouter } from '@@/create-router';
import { BAD_REQUEST, NOT_FOUND } from '@@/custom-errors';
import { getEmailAccounts, getEmailAccount, createEmailAccount, deleteEmailAccount, updateEmailAccountStatus } from '../db/queries';
import { getValidGoogleAccessToken } from '@@/api/integrations/google-auth';
import { validateImapConnection } from './imap-validate';
import { startSync, isSyncing, getSyncStates } from './sync-runner';
import { imapSyncSteps } from './imap-sync';
import { gmailSyncSteps } from './gmail-api';
import { openEmailDb } from './store';
import { hasSyncedBefore } from './sync-meta';
import { performResync } from './resync';
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);
// Check for stale syncing/queued accounts with no active job
const staleIds: number[] = [];
const hasActiveAccounts = accounts.some((a) => a.status === 'syncing' || a.status === 'queued');
let activeJobAccountIds = new Set<number>();
if (hasActiveAccounts) {
// In-process now: a running sync is one this process started, not a queued job row. An account marked
// syncing with nothing running is stale — the usual cause is a restart mid-sync.
activeJobAccountIds = new Set(
getSyncStates()
.filter((s) => s.status === 'running')
.map((s) => s.accountId),
);
for (const a of accounts) {
if ((a.status === 'syncing' || a.status === 'queued') && !activeJobAccountIds.has(a.id)) {
staleIds.push(a.id);
}
}
// Reset stale accounts in background
if (staleIds.length > 0) {
for (const id of staleIds) {
updateEmailAccountStatus(id, 'connected').catch(() => {});
}
}
}
return ctx.json(
accounts.map((a) => ({
id: a.id,
provider: a.provider,
email: a.email,
displayName: a.displayName,
enabled: a.enabled,
status: staleIds.includes(a.id) ? 'connected' : 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');
// Determine if this is a first sync or resync
// One answer for both providers now. This used to fork: Gmail read SQLite, IMAP read the Postgres
// column, which is how the two drifted apart in the first place.
let isFirstSync = true;
{
const db = openEmailDb(user.email, account.email);
try {
isFirstSync = !hasSyncedBefore(db, account.syncMeta as Record<string, unknown>);
} finally {
db.close();
}
}
// Resync: call directly, no job queue
if (!isFirstSync) {
await updateEmailAccountStatus(id, 'syncing');
try {
const result = await performResync({ accountId: id, userEmail: user.email, userId: user.id });
await updateEmailAccountStatus(id, 'synced');
return ctx.json({ ok: true, saved: result.saved });
} catch (err) {
await updateEmailAccountStatus(id, 'synced').catch(() => {});
throw err;
}
}
// First sync: enqueue a job
const authResult = await resolveAuth(
user.id,
account.authType,
account.email,
account.credentials as Record<string, unknown>,
);
if (!authResult.ok) throw BAD_REQUEST(authResult.error);
await updateEmailAccountStatus(id, 'queued');
// Gmail API sync needs OAuth; a gmail account with an app password syncs over IMAP instead.
const useGmailApi = account.provider === 'gmail' && account.authType === 'oauth';
if (isSyncing(id)) return ctx.json({ ok: true, alreadyRunning: true });
startSync({
accountId: id,
// The OWNER's email, not the account's — the store path is keyed by it.
ownerEmail: user.email,
steps: useGmailApi ? gmailSyncSteps : imapSyncSteps,
onDone: (ok) => updateEmailAccountStatus(id, ok ? 'synced' : 'error').then(() => undefined),
meta: {
emailAccountId: id,
userEmail: user.email,
account: {
id: account.id,
userId: account.userId,
email: account.email,
imapHost: account.imapHost,
imapPort: account.imapPort,
imapSecure: account.imapSecure,
provider: account.provider,
authType: account.authType,
credentials: account.credentials,
},
imapAuth: { user: account.email, ...authResult.auth },
},
});
return ctx.json({ ok: true }, 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' };
let accessToken: string | null;
try {
accessToken = await getValidGoogleAccessToken(userId);
} catch (err) {
return {
ok: false,
error: `Token refresh failed — reconnect Google account: ${err instanceof Error ? err.message : err}`,
};
}
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}` };
}
+45
View File
@@ -0,0 +1,45 @@
import { getUserById } from 'officerdb';
import { getAllSyncedAccounts } from '../db/queries';
import { performResync } from './resync';
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;
for (const account of accounts) {
const user = await getUserById(account.userId);
if (!user) continue;
try {
console.log(`[email-cron] Resyncing ${account.email}`);
const result = await performResync({ accountId: account.id, userEmail: user.email, userId: user.id });
if (result.saved > 0) {
console.log(`[email-cron] ${account.email}: ${result.saved} new emails`);
}
} catch (err) {
console.error(`[email-cron] Failed to resync ${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);
setTimeout(tick, 30_000);
}
export function stopEmailCron() {
if (timer) {
clearInterval(timer);
timer = null;
}
}
+171
View File
@@ -0,0 +1,171 @@
import type { ImapFlow } from 'imapflow';
import { getUserById } from 'officerdb';
import { getAllSyncedAccounts } from '../db/queries';
import { performResync } from './resync';
import { getValidGoogleAccessToken } from '@@/api/integrations/google-auth';
// Real-time email via IMAP IDLE: one persistent connection per account. imapflow auto-enters IDLE
// while a mailbox is open and pushes an `exists` event the instant new mail lands — we then run the
// same incremental resync the cron uses. The cron stays on as a slow backstop for missed events.
type Account = Awaited<ReturnType<typeof getAllSyncedAccounts>>[number];
type Watcher = {
account: Account;
client: ImapFlow | null;
closing: boolean;
syncing: boolean;
pending: boolean; // an exists event arrived mid-sync → run once more
backoff: number;
reconnectTimer: ReturnType<typeof setTimeout> | null;
};
const RECONNECT_BASE = 5_000;
const RECONNECT_MAX = 5 * 60_000;
const RECONCILE_MS = 5 * 60_000;
const watchers = new Map<number, Watcher>();
let reconcileTimer: ReturnType<typeof setInterval> | null = null;
let notifyNewMail: ((userEmail: string) => void) | null = null;
async function resolveAuth(account: Account): Promise<{ user: string; pass?: string; accessToken?: string } | null> {
if (account.authType === 'oauth') {
const token = await getValidGoogleAccessToken(account.userId).catch(() => null);
return token ? { user: account.email, accessToken: token } : null;
}
const pass = (account.credentials as Record<string, unknown> | null)?.password as string | undefined;
return pass ? { user: account.email, pass } : null;
}
// Coalesced incremental sync — a new event during a sync queues exactly one follow-up.
async function runSync(w: Watcher): Promise<void> {
if (w.syncing) {
w.pending = true;
return;
}
w.syncing = true;
try {
const user = await getUserById(w.account.userId);
if (!user) return;
const result = await performResync({ accountId: w.account.id, userEmail: user.email, userId: user.id });
if (result.saved > 0) {
console.log(`[email-idle] ${w.account.email}: ${result.saved} new`);
notifyNewMail?.(user.email);
}
} catch (err) {
console.error(`[email-idle] sync failed for ${w.account.email}:`, err instanceof Error ? err.message : err);
} finally {
w.syncing = false;
if (w.pending) {
w.pending = false;
void runSync(w);
}
}
}
function scheduleReconnect(w: Watcher): void {
if (w.closing || w.reconnectTimer) return;
const delay = w.backoff;
w.backoff = Math.min(w.backoff * 2, RECONNECT_MAX);
w.reconnectTimer = setTimeout(() => {
w.reconnectTimer = null;
void connect(w);
}, delay);
}
async function connect(w: Watcher): Promise<void> {
if (w.closing) return;
const auth = await resolveAuth(w.account);
if (!auth) {
scheduleReconnect(w);
return;
}
const { ImapFlow } = await import('imapflow');
const client = new ImapFlow({
host: w.account.imapHost,
port: w.account.imapPort,
secure: w.account.imapSecure,
auth,
logger: false,
emitLogs: false,
});
w.client = client;
client.on('exists', () => void runSync(w));
client.on('error', () => {
/* a 'close' event follows and handles the reconnect */
});
client.on('close', () => {
w.client = null;
scheduleReconnect(w);
});
try {
await client.connect();
await client.mailboxOpen('INBOX'); // imapflow now auto-IDLEs and emits `exists` on new mail
w.backoff = RECONNECT_BASE; // reset after a clean connect
console.log(`[email-idle] watching ${w.account.email}`);
void runSync(w); // catch up on anything that arrived while disconnected
} catch (err) {
console.error(`[email-idle] connect failed for ${w.account.email}:`, err instanceof Error ? err.message : err);
w.client = null;
scheduleReconnect(w);
}
}
function startWatcher(account: Account): void {
if (watchers.has(account.id)) return;
const w: Watcher = {
account,
client: null,
closing: false,
syncing: false,
pending: false,
backoff: RECONNECT_BASE,
reconnectTimer: null,
};
watchers.set(account.id, w);
void connect(w);
}
function stopWatcher(id: number): void {
const w = watchers.get(id);
if (!w) return;
w.closing = true;
if (w.reconnectTimer) clearTimeout(w.reconnectTimer);
w.client?.logout().catch(() => {});
watchers.delete(id);
}
// Keep watchers in sync with the account list (added/removed/credential changes).
async function reconcile(): Promise<void> {
try {
const accounts = await getAllSyncedAccounts();
const ids = new Set(accounts.map((a) => a.id));
for (const a of accounts) {
const w = watchers.get(a.id);
if (!w) startWatcher(a);
else w.account = a; // refresh host/creds for the next reconnect
}
for (const id of [...watchers.keys()]) if (!ids.has(id)) stopWatcher(id);
} catch (err) {
console.error('[email-idle] reconcile error:', err instanceof Error ? err.message : err);
}
}
export function initEmailIdle(onNewMail?: (userEmail: string) => void): void {
if (reconcileTimer) return;
notifyNewMail = onNewMail ?? null;
console.log('[email-idle] starting IMAP IDLE watchers');
setTimeout(() => void reconcile(), 5_000);
reconcileTimer = setInterval(() => void reconcile(), RECONCILE_MS);
}
export function stopEmailIdle(): void {
if (reconcileTimer) {
clearInterval(reconcileTimer);
reconcileTimer = null;
}
for (const id of [...watchers.keys()]) stopWatcher(id);
}
+707
View File
@@ -0,0 +1,707 @@
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[];
+78
View File
@@ -0,0 +1,78 @@
import { Hono } from 'hono';
import { getUserById } from 'officerdb';
import { CustomError } from '@@/custom-errors';
import { bodyParser } from '@@/_middlewares/body-parser';
import { emailRouter } from './routes';
// The email sidecar's own listener. `/api/email/*` on the platform is a proxy onto this — the platform
// authenticates the owner, injects X-Officer-User, and forwards without reading the body.
//
// The routes moved here verbatim, Hono and all, so this file's only real job is to reconstruct the three
// things the platform's middleware used to provide: the authenticated user on the context, the parsed
// body on the context, and an error handler that turns thrown custom-errors into status codes.
type HonoVariables = { user: { id: number }; body: Record<string, unknown>; origin: string };
const app = new Hono<{ Variables: HonoVariables }>();
// The platform's userMiddleware set the whole user row on the context from the JWT. The proxy injects only
// the id (X-Officer-User), so the rest is loaded here — and it has to be the WHOLE row, not just the id:
// the mail store is keyed by the owner's email (DATA_PATH/<owner>/email_accounts/<account>/emails.db), so
// a user object without `email` silently resolves to the wrong path and the mailbox reads as empty.
//
// Cached because this runs on every request and the owner does not change; single-user is a hard invariant.
let cachedUser: { id: number; email: string } | null = null;
app.use('*', async (ctx, next) => {
const id = Number(ctx.req.header('X-Officer-User'));
if (!Number.isFinite(id) || id <= 0) return ctx.json({ error: 'missing X-Officer-User' }, 401);
if (!cachedUser || cachedUser.id !== id) {
const row = await getUserById(id);
if (!row?.email) return ctx.json({ error: `unknown user ${id}` }, 401);
cachedUser = row as { id: number; email: string };
}
ctx.set('user', cachedUser as never);
await next();
});
// Handlers read the body off the context (`ctx.get('body')`), which only the platform's middleware used to
// provide. Without this every write route here destructured undefined — `POST /accounts` threw on
// `body.provider` and surfaced as a bare "failed to add account" toast. `isProxiedPath` is inert in this
// process (the prefix set is populated by `createSidecarProxy`, which runs in the platform), so nothing is
// skipped; multipart on `/send` is parsed once and served from Hono's cache when the route asks again.
app.use('*', bodyParser());
app.route('/', emailRouter as never);
// Mirrors the platform's own handler (`hono.ts` onError) rather than inventing a second error shape. Two
// bugs lived in the version that didn't: custom-errors carry `statusCode`, NOT `status`, so every
// deliberate 4xx fell through to the 500 branch with its message replaced by "internal error"; and the
// body was JSON where the whole API answers errors as plain text, which `useClient` reads with
// `res.text()` — so even a correct message reached the client wrapped in `{"error":…}`.
app.onError((err, ctx) => {
if (err instanceof CustomError) {
if (err.returnValue) {
return typeof err.returnValue === 'string'
? ctx.text(err.returnValue, err.statusCode)
: ctx.json(err.returnValue, err.statusCode);
}
return ctx.text(err.message, err.statusCode);
}
console.error('[email] unhandled error', err);
return ctx.text('Internal Server Error', 500);
});
export function startEmailServer(): number {
const server = Bun.serve({
port: 0,
hostname: '127.0.0.1',
// A from-scratch mailbox sync answers slowly; the platform proxy extends its own side to match.
idleTimeout: 255,
fetch: app.fetch,
});
const port = server.port;
if (port == null) throw new Error('[email] failed to acquire a port');
console.log(`[email] http server listening on http://127.0.0.1:${port}`);
return port;
}
+427
View File
@@ -0,0 +1,427 @@
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[];
+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 };
}
}
+72
View File
@@ -0,0 +1,72 @@
import type { SidecarEvent } from '@@/sidecar/protocol';
import { initEmailCron, stopEmailCron } from './email-cron';
import { initEmailIdle, stopEmailIdle } from './email-idle';
import { broadcastEmailNew } from './routes';
import { startEmailServer } from './http';
import { createSidecarConnector } from '@@/sidecar/connect';
import { API_URL } from '@@/officer-url.mjs';
// The sidecar used to reach BACK into the platform's queue over this socket to get a sync run —
// enqueueViaWs / listJobsViaWs and a pending-response map. Syncs run in this process now
// (sync-runner.ts), so the shim is gone and nothing but a port crosses the socket at startup.
// ── Command handlers ──
type ReplyFn = (msg: SidecarEvent) => void;
function handleCommand(cmd: Record<string, unknown>, reply: ReplyFn) {
switch (cmd.type) {
case 'ping':
reply({ type: 'pong', id: cmd.id as string });
break;
default:
reply({
type: 'error',
id: cmd.id as string,
error: `Unknown command type: ${cmd.type}`,
});
}
}
// ── HTTP server ──
//
// Started before the registration socket so the port is known by the time we announce it. `/api/email/*`
// on the platform is a proxy onto this.
const serverPort = startEmailServer();
// ── Connect to API server ──
const connection = createSidecarConnector({
apiUrl: `${API_URL}/api/sidecar/register`,
name: 'email',
handles: ['email'],
onCommand(cmd, reply) {
handleCommand(cmd as Record<string, unknown>, reply as ReplyFn);
},
onConnected() {
// The platform forgets the port when the socket drops, and this listener outlives an officer restart,
// so re-announce on every reconnect.
connection.send({ type: 'email:server', port: serverPort });
// Start email cron once connected (so queue commands can reach API server)
initEmailCron();
// Real-time push via IMAP IDLE; the cron above is the slow backstop. On new mail, tell the API
// server so it can push an SSE event to that user's open /email page.
// Straight to this process's own SSE clients — the /email/events stream lives here now, so the
// round trip out to officer and back (the `email:new` wire event) is gone.
initEmailIdle((userEmail) => broadcastEmailNew(userEmail));
},
});
// ── Graceful shutdown ──
function shutdown(signal: string) {
console.log(`[email] ${signal} received, shutting down...`);
stopEmailIdle();
stopEmailCron();
connection.destroy();
process.exit(0);
}
process.on('SIGTERM', () => shutdown('SIGTERM'));
process.on('SIGINT', () => shutdown('SIGINT'));
+327
View File
@@ -0,0 +1,327 @@
import { createHash } from 'node:crypto';
import { getUserIntegration, upsertUserIntegration, getServerIntegration } from 'officerdb';
import { getEmailAccount } from '../db/queries';
import { refreshGoogleAccessToken } from '@@/api/integrations/google-auth';
import { openEmailDb, getSyncMeta, setSyncMeta, upsertFromRawEml } from './store';
import { readSyncMeta, writeSyncMeta } from './sync-meta';
import { type GmailCredentials, loadGmailCredentials, gmailApiSync } from './gmail-api';
export type ResyncResult = { saved: number; skipped: number; errors: number };
// ── Gmail resync (REST API, history-based) ──
async function refreshCredentials(creds: GmailCredentials): Promise<GmailCredentials> {
if (!creds.accessToken || !creds.refreshToken) return creds;
const tokenExpired = !creds.expiresAt || creds.expiresAt < Date.now() + 60_000;
if (!tokenExpired) return creds;
console.log('[resync] Refreshing OAuth token');
const refreshed = await refreshGoogleAccessToken(creds.refreshToken);
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 },
});
return { ...creds, accessToken: refreshed.accessToken, expiresAt: refreshed.expiresAt };
}
async function gmailResync(userEmail: string, accountEmail: string): Promise<ResyncResult> {
let creds = await loadGmailCredentials(userEmail);
if (!creds.accessToken) {
throw new Error('OAuth not configured — connect Google in Settings → Integrations for resyncs');
}
creds = await refreshCredentials(creds);
const db = openEmailDb(userEmail, accountEmail);
try {
const result = await gmailApiSync({ creds, db });
setSyncMeta(db, 'last_sync_date', new Date().toISOString().split('T')[0]!);
setSyncMeta(db, 'last_sync_at', new Date().toISOString());
console.log(`[resync] Gmail done: saved ${result.saved}, skipped ${result.skipped}, errors ${result.errors}`);
return result;
} finally {
db.close();
}
}
// ── Generic IMAP resync ──
type ImapAccountInfo = {
id: number;
userId: number;
email: string;
imapHost: string;
imapPort: number;
imapSecure: boolean;
provider: string;
authType: string;
credentials: Record<string, unknown>;
};
type FolderInfo = {
specialUse?: string;
path: string;
flags: Set<string>;
status?: { uidNext?: number; uidValidity?: number };
};
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']);
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();
}
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);
}
async function resolveImapAuth(
account: ImapAccountInfo,
): Promise<{ user: string; pass?: string; accessToken?: string }> {
if (account.authType !== 'oauth') {
return { user: account.email, pass: account.credentials.password as string };
}
const userGoogle = await getUserIntegration(account.userId, 'google');
const config = userGoogle?.config as Record<string, unknown> | undefined;
if (!config?.refreshToken) throw new Error('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: account.email, accessToken: config.accessToken as string };
}
const refreshed = await refreshGoogleAccessToken(config.refreshToken as string);
const serverGoogle = await getServerIntegration('google');
await upsertUserIntegration({
userId: account.userId,
provider: 'google',
serverIntegrationId: serverGoogle?.id,
config: { ...config, accessToken: refreshed.accessToken, expiresAt: refreshed.expiresAt },
});
return { user: account.email, accessToken: refreshed.accessToken };
}
async function imapResync(account: ImapAccountInfo, userEmail: string): Promise<ResyncResult> {
const { ImapFlow } = await import('imapflow');
const imapAuth = await resolveImapAuth(account);
const freshAccount = await getEmailAccount(account.id);
if (!freshAccount) throw new Error(`Email account ${account.id} not found`);
const db = openEmailDb(userEmail, account.email);
// The position lives in this file now; the Postgres column is only a one-time backfill source for
// accounts that predate the move. See sync-meta.ts.
const syncMeta: Record<string, unknown> = readSyncMeta(db, freshAccount.syncMeta as Record<string, unknown>);
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);
let saved = 0;
let skipped = 0;
let errors = 0;
const client = new ImapFlow({
host: account.imapHost,
port: account.imapPort,
secure: account.imapSecure,
auth: imapAuth,
logger: false,
socketTimeout: 5 * 60 * 1000,
});
try {
await client.connect();
console.log('[resync] IMAP connected');
const folders = (await client.list({ statusQuery: { uidNext: true, uidValidity: true } })) as FolderInfo[];
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;
const lastUid =
storedUidValidity && uidValidity === storedUidValidity && storedLastUid ? parseInt(storedLastUid, 10) : 0;
if (lastUid > 0 && uidNext <= lastUid + 1) continue;
let lock;
try {
lock = await client.getMailboxLock(folder.path);
} catch {
continue;
}
try {
const mailbox = client.mailbox;
if (!mailbox) continue;
const mbUidValidity = String(mailbox.uidValidity);
const effectiveLastUid = storedUidValidity === mbUidValidity ? 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;
}
try {
upsertFromRawEml({
db,
id,
raw,
integration: account.provider,
emailAccount: account.email,
labels: [label],
});
existingIds.add(id);
saved++;
} catch {
errors++;
}
}
} catch (fetchErr) {
const errMsg = fetchErr instanceof Error ? fetchErr.message : String(fetchErr);
if (!errMsg.includes('Nothing to fetch')) {
console.log(`[resync] Fetch error in ${folder.path}: ${errMsg}`);
errors++;
}
}
syncMeta[uidValidityKey] = mbUidValidity;
if (maxUid > effectiveLastUid) {
syncMeta[lastUidKey] = String(maxUid);
}
} finally {
lock.release();
}
}
await client.logout().catch(() => {});
} catch (err) {
await client.logout().catch(() => {});
throw err;
} finally {
db.close();
}
syncMeta.last_sync_at = new Date().toISOString();
// Reopened deliberately: the sync loop above closes the db in its `finally`, and the position must be
// written after the messages it describes are safely stored.
const metaDb = openEmailDb(userEmail, account.email);
try {
writeSyncMeta(metaDb, syncMeta);
} finally {
metaDb.close();
}
console.log(`[resync] IMAP done: saved ${saved}, skipped ${skipped}, errors ${errors}`);
return { saved, skipped, errors };
}
// ── Public API ──
type ResyncParams = {
accountId: number;
userEmail: string;
userId: number;
};
// Coalesce concurrent resyncs of the same account within this process (IDLE + cron + manual can all
// fire) — a caller arriving mid-resync just awaits the one already running.
const resyncInFlight = new Map<number, Promise<ResyncResult>>();
export function performResync(params: ResyncParams): Promise<ResyncResult> {
const running = resyncInFlight.get(params.accountId);
if (running) return running;
const p = doResync(params).finally(() => resyncInFlight.delete(params.accountId));
resyncInFlight.set(params.accountId, p);
return p;
}
async function doResync({ accountId, userEmail, userId }: ResyncParams): Promise<ResyncResult> {
const account = await getEmailAccount(accountId);
if (!account || account.userId !== userId) throw new Error('Account not found');
// Gmail API resync needs OAuth; a gmail account authed with an app password resyncs over IMAP.
if (account.provider === 'gmail' && account.authType === 'oauth') {
return gmailResync(userEmail, account.email);
}
return imapResync(
{
id: account.id,
userId: account.userId,
email: account.email,
imapHost: account.imapHost,
imapPort: account.imapPort,
imapSecure: account.imapSecure,
provider: account.provider,
authType: account.authType,
credentials: account.credentials as Record<string, unknown>,
},
userEmail,
);
}
+433
View File
@@ -0,0 +1,433 @@
import { mkdir } from 'node:fs/promises';
import { join } from 'node:path';
import nodemailer from 'nodemailer';
import type { EmailMessage } from '../db/types';
import { createRouter } from '@@/create-router';
import * as errors from '@@/custom-errors';
import { getEmailAttachmentCacheDir } from '@@/data-path';
import { getEmailAccounts } from '../db/queries';
import { openEmailDb, openUserEmailDb, rowToSummary, getSyncMeta, searchEmails } from './store';
import { accountsRouter } from './accounts';
export const emailRouter = createRouter();
emailRouter.route('/accounts', accountsRouter);
// ── Sending (SMTP) — sends as the connected account using its app password ──
async function getSmtpTransport(
userId: number,
): Promise<{ transport: ReturnType<typeof nodemailer.createTransport>; from: string }> {
const accounts = await getEmailAccounts(userId);
const acct = accounts.find((a) => a.enabled) ?? accounts[0];
if (!acct) throw errors.BAD_REQUEST('No email account configured');
const pass = (acct.credentials as Record<string, unknown> | null)?.password as string | undefined;
if (!pass) throw errors.BAD_REQUEST('This account has no SMTP password — sending needs an app-password account');
// Gmail: smtp.gmail.com:465 (SSL). Derive from the IMAP host for other providers.
const host = acct.provider === 'gmail' ? 'smtp.gmail.com' : acct.imapHost.replace(/^imap\./, 'smtp.');
const transport = nodemailer.createTransport({ host, port: 465, secure: true, auth: { user: acct.email, pass } });
return { transport, from: acct.email };
}
// POST /send — compose/reply (multipart; `files` are attachments). Gmail auto-files the sent copy in
// "Sent", so IMAP sync picks it up.
emailRouter.post('/send', async (ctx) => {
const user = ctx.get('user');
const form = await ctx.req.parseBody({ all: true });
const str = (v: unknown) => (typeof v === 'string' ? v : '');
const to = str(form.to).trim();
if (!to) throw errors.BAD_REQUEST('At least one recipient is required');
const toFiles = (raw: unknown) =>
(Array.isArray(raw) ? raw : raw ? [raw] : []).filter((f): f is File => f instanceof File);
const attachments = await Promise.all(
toFiles(form.files).map(async (f) => ({
filename: f.name,
content: Buffer.from(await f.arrayBuffer()),
contentType: f.type || undefined,
})),
);
// Inline images: cid `inline-<i>` matches the `<img src="cid:inline-i">` the composer put in the html.
const inline = await Promise.all(
toFiles(form.inline).map(async (f, i) => ({
filename: f.name,
content: Buffer.from(await f.arrayBuffer()),
contentType: f.type || undefined,
cid: `inline-${i}`,
contentDisposition: 'inline' as const,
})),
);
const allAttachments = [...attachments, ...inline];
const html = str(form.html).trim();
const inReplyTo = str(form.inReplyTo).trim();
const { transport, from } = await getSmtpTransport(user.id);
await transport.sendMail({
from,
to,
cc: str(form.cc).trim() || undefined,
bcc: str(form.bcc).trim() || undefined,
subject: str(form.subject).trim() || '(no subject)',
text: str(form.body),
...(html ? { html } : {}),
...(allAttachments.length ? { attachments: allAttachments } : {}),
...(inReplyTo ? { headers: { 'In-Reply-To': inReplyTo, References: inReplyTo } } : {}),
});
return ctx.json({ ok: true });
});
// GET /contacts?q= — address autocomplete from people you've received mail from, ranked by frequency.
emailRouter.get('/contacts', async (ctx) => {
const user = ctx.get('user');
const like = `%${(ctx.req.query('q') ?? '').trim().toLowerCase()}%`;
const db = await openUserEmailDb(user.email, user.id);
if (!db) return ctx.json([]);
try {
const rows = db
.query(
`SELECT lower(from_address) AS address, from_name AS name, count(*) AS c
FROM emails
WHERE from_address IS NOT NULL AND from_address != '' AND deleted = 0
AND (lower(from_address) LIKE ? OR lower(from_name) LIKE ?)
GROUP BY lower(from_address)
ORDER BY c DESC LIMIT 10`,
)
.all(like, like) as Array<{ address: string; name: string | null }>;
return ctx.json(rows.map((r) => ({ address: r.address, name: r.name || '' })));
} finally {
db.close();
}
});
// ── Real-time: per-user SSE stream of email events (fed by the IMAP IDLE watcher) ──
const emailSseClients = new Map<string, Set<ReadableStreamDefaultController<Uint8Array>>>();
// Called from the sidecar-message handler when the IDLE watcher saves new mail for a user.
export function broadcastEmailNew(userEmail: string): void {
const set = emailSseClients.get(userEmail);
if (!set || set.size === 0) return;
const payload = new TextEncoder().encode(`data: ${JSON.stringify({ type: 'new-mail' })}\n\n`);
for (const ctrl of set) {
try {
ctrl.enqueue(payload);
} catch {
/* dead controller — cleaned up on cancel */
}
}
}
// EventSource endpoint (auth via ?token= handled by userMiddleware). The frontend opens this only
// while on /email, so events stop the moment you navigate away.
emailRouter.get('/events', (ctx) => {
const email = ctx.get('user').email;
const enc = new TextEncoder();
let controllerRef: ReadableStreamDefaultController<Uint8Array>;
let ping: ReturnType<typeof setInterval>;
const stream = new ReadableStream<Uint8Array>({
start(controller) {
controllerRef = controller;
let set = emailSseClients.get(email);
if (!set) {
set = new Set();
emailSseClients.set(email, set);
}
set.add(controller);
controller.enqueue(enc.encode('retry: 3000\n\n'));
ping = setInterval(() => {
try {
controller.enqueue(enc.encode(': ping\n\n'));
} catch {
/* closed */
}
}, 25_000);
},
cancel() {
clearInterval(ping);
const set = emailSseClients.get(email);
set?.delete(controllerRef);
if (set && set.size === 0) emailSseClients.delete(email);
},
});
return new Response(stream, {
headers: { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache', Connection: 'keep-alive' },
});
});
// Full-text search across all mail (subject / sender / recipients / snippet / body), newest first.
emailRouter.get('/search', async (ctx) => {
const user = ctx.get('user');
const q = (ctx.req.query('q') ?? '').trim();
const page = Math.max(1, Number(ctx.req.query('page') ?? '1') || 1);
const limit = Math.min(100, Math.max(1, Number(ctx.req.query('limit') ?? '50') || 50));
const offset = (page - 1) * limit;
if (!q) return ctx.json({ messages: [], total: 0 });
const db = await openUserEmailDb(user.email, user.id);
if (!db) return ctx.json({ messages: [], total: 0 });
try {
const { rows, total } = searchEmails(db, q, limit, offset);
return ctx.json({ messages: rows.map(rowToSummary), total });
} finally {
db.close();
}
});
// `folder` is request input and was being interpolated straight into the SQL. It is bound now — the
// fragment and its parameters travel together because each call site builds several statements from the
// same fragment and has to spread the params in the right order.
type FolderFilter = { where: string; params: string[] };
const folderFilter = (folder: string): FolderFilter =>
folder === 'all'
? { where: 'deleted = 0', params: [] }
: { where: 'deleted = 0 AND labels LIKE ?', params: [`%${folder}%`] };
emailRouter.get('/messages', async (ctx) => {
const user = ctx.get('user');
// `|| n` also catches NaN from a non-numeric query param, which used to reach the bindings as NaN.
const page = Math.max(Number(ctx.req.query('page') ?? '1') || 1, 1);
const limit = Math.max(Number(ctx.req.query('limit') ?? '50') || 50, 1);
const folder = ctx.req.query('folder') ?? 'inbox';
const offset = (page - 1) * limit;
const { where: folderWhere, params: folderParams } = folderFilter(folder);
const db = await openUserEmailDb(user.email, user.id);
if (!db) return ctx.json({ messages: [], total: 0 });
try {
// One row per conversation: the latest message in each thread within this folder, plus the
// thread's message count and how many are unread. COALESCE guards any un-backfilled rows.
const rows = db
.query(
`SELECT * FROM (
SELECT e.*,
COUNT(*) OVER (PARTITION BY COALESCE(thread_id, id)) AS thread_count,
SUM(CASE WHEN read = 0 THEN 1 ELSE 0 END) OVER (PARTITION BY COALESCE(thread_id, id)) AS thread_unread,
ROW_NUMBER() OVER (PARTITION BY COALESCE(thread_id, id) ORDER BY date DESC, id DESC) AS rn
FROM emails e
WHERE ${folderWhere}
) WHERE rn = 1
ORDER BY date DESC LIMIT ? OFFSET ?`,
)
.all(...folderParams, limit, offset) as Record<string, unknown>[];
const countRow = db
.query(`SELECT COUNT(DISTINCT COALESCE(thread_id, id)) as total FROM emails WHERE ${folderWhere}`)
.get(...folderParams) as { total: number };
const messages = rows.map(rowToSummary);
return ctx.json({ messages, total: countRow.total });
} finally {
db.close();
}
});
function buildMessage(db: ReturnType<typeof openEmailDb>, row: Record<string, unknown>): EmailMessage {
const attachmentRows = db
.query('SELECT * FROM attachments WHERE email_id = ? ORDER BY idx')
.all(row.id as string) as Array<Record<string, unknown>>;
const from = row.from_name ? `${row.from_name} <${row.from_address}>` : (row.from_address as string);
return {
id: row.id as string,
from,
to: row.to_address as string,
cc: (row.cc as string) ?? undefined,
subject: row.subject as string,
date: row.date as string,
snippet: row.snippet as string,
html: (row.html as string) ?? undefined,
text: (row.text_body as string) ?? undefined,
attachments: attachmentRows.map((a) => ({
filename: a.filename as string,
size: a.size as number,
contentType: a.content_type as string,
})),
...(row.attachment_count ? { attachmentCount: row.attachment_count as number } : {}),
...(row.read ? { read: true } : {}),
...(row.labels ? { labels: (row.labels as string).split(',') } : {}),
};
}
emailRouter.get('/messages/:id', async (ctx) => {
const user = ctx.get('user');
const id = ctx.req.param('id');
const db = await openUserEmailDb(user.email, user.id);
if (!db) return ctx.text('Not found', 404);
try {
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);
return ctx.json(buildMessage(db, row));
} finally {
db.close();
}
});
// GET /thread/:id — the full conversation containing message :id, oldest message first.
emailRouter.get('/thread/:id', async (ctx) => {
const user = ctx.get('user');
const id = ctx.req.param('id');
const db = await openUserEmailDb(user.email, user.id);
if (!db) return ctx.text('Not found', 404);
try {
const head = db.query('SELECT thread_id, subject FROM emails WHERE id = ? AND deleted = 0').get(id) as {
thread_id: string | null;
subject: string;
} | null;
if (!head) return ctx.text('Not found', 404);
const threadKey = head.thread_id ?? id;
const rows = db
.query('SELECT * FROM emails WHERE COALESCE(thread_id, id) = ? AND deleted = 0 ORDER BY date ASC')
.all(threadKey) as Record<string, unknown>[];
const messages = rows.map((row) => buildMessage(db, row));
return ctx.json({ id, subject: head.subject, messages });
} finally {
db.close();
}
});
// PATCH /thread/:id/read — mark every message in the conversation as read.
emailRouter.patch('/thread/:id/read', async (ctx) => {
const user = ctx.get('user');
const id = ctx.req.param('id');
const db = await openUserEmailDb(user.email, user.id);
if (!db) return ctx.text('Not found', 404);
try {
const head = db.query('SELECT thread_id FROM emails WHERE id = ?').get(id) as { thread_id: string | null } | null;
if (!head) return ctx.text('Not found', 404);
const threadKey = head.thread_id ?? id;
db.run('UPDATE emails SET read = 1 WHERE COALESCE(thread_id, id) = ? AND read = 0', [threadKey]);
return ctx.json({ ok: true });
} finally {
db.close();
}
});
emailRouter.post('/messages/:id/attachments/:index/extract', async (ctx) => {
const user = ctx.get('user');
const id = ctx.req.param('id');
const index = Number(ctx.req.param('index'));
const db = await openUserEmailDb(user.email, user.id);
if (!db) return ctx.text('Attachment not found', 404);
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;
if (!row || !row.content) return ctx.text('Attachment not found', 404);
const fileName = row.filename ?? 'unknown';
const attachDir = getEmailAttachmentCacheDir(user.email);
const destPath = join(attachDir, fileName);
const destFile = Bun.file(destPath);
if (!(await destFile.exists())) {
await mkdir(attachDir, { recursive: true });
const binary = Buffer.from(row.content, 'base64');
await Bun.write(destPath, binary);
}
return ctx.json({ filePath: `email_accounts/attachment_cache/${fileName}`, fileName, root: 'user-data' });
} catch {
return ctx.text('Failed to extract attachment', 500);
} finally {
db.close();
}
});
emailRouter.patch('/messages/:id/read', async (ctx) => {
const user = ctx.get('user');
const id = ctx.req.param('id');
const db = await openUserEmailDb(user.email, user.id);
if (!db) return ctx.json({ ok: true });
try {
db.run('UPDATE emails SET read = 1 WHERE id = ? AND read = 0', [id]);
return ctx.json({ ok: true });
} finally {
db.close();
}
});
emailRouter.delete('/messages/:id', async (ctx) => {
const user = ctx.get('user');
const id = ctx.req.param('id');
const db = await openUserEmailDb(user.email, user.id);
if (!db) return ctx.text('Not found', 404);
try {
const result = db.run('UPDATE emails SET deleted = 1 WHERE id = ? AND deleted = 0', [id]);
if (result.changes === 0) return ctx.text('Not found', 404);
return ctx.json({ ok: true });
} finally {
db.close();
}
});
emailRouter.get('/sync-status', async (ctx) => {
const user = ctx.get('user');
const db = await openUserEmailDb(user.email, user.id);
if (!db) return ctx.json({ lastSyncAt: null });
try {
const lastSyncAt = getSyncMeta(db, 'last_sync_at');
return ctx.json({ lastSyncAt });
} finally {
db.close();
}
});
emailRouter.get('/stats', async (ctx) => {
const user = ctx.get('user');
const folder = ctx.req.query('folder') ?? 'inbox';
const { where: folderWhere, params: folderParams } = folderFilter(folder);
const db = await openUserEmailDb(user.email, user.id);
if (!db) return ctx.json({ total: 0, byDomain: [], bySender: [] });
try {
const total = (
db.query(`SELECT COUNT(*) as count FROM emails WHERE ${folderWhere}`).get(...folderParams) 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(...folderParams) 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(...folderParams) as Array<{ from_address: string; from_name: string; count: number }>;
return ctx.json({ total, byDomain, bySender });
} finally {
db.close();
}
});
emailRouter.get('/labels', async (ctx) => {
const user = ctx.get('user');
const db = await openUserEmailDb(user.email, user.id);
if (!db) return ctx.json({ labels: [] });
try {
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) {
for (const label of row.labels.split(',')) {
const trimmed = label.trim().toLowerCase();
if (trimmed) counts.set(trimmed, (counts.get(trimmed) ?? 0) + 1);
}
}
const labels = [...counts.entries()].map(([label, count]) => ({ label, count })).sort((a, b) => b.count - a.count);
return ctx.json({ labels });
} finally {
db.close();
}
});
+795
View File
@@ -0,0 +1,795 @@
import { Database } from 'bun:sqlite';
import { createHash } from 'node:crypto';
import { join, dirname } from 'node:path';
import { chmodSync, mkdirSync } from 'node:fs';
import { getEmailDbPath } from '@@/data-path';
import { getEmailAccounts } from '../db/queries';
import type { EmailSummary } from '../db/types';
const SCHEMA_TABLES = `
CREATE TABLE IF NOT EXISTS emails (
id TEXT PRIMARY KEY,
integration TEXT NOT NULL DEFAULT 'gmail',
email_account TEXT NOT NULL DEFAULT '',
from_name TEXT,
from_address TEXT,
from_domain TEXT,
to_address TEXT,
cc TEXT,
subject TEXT,
date TEXT,
snippet TEXT,
html TEXT,
text_body TEXT,
attachment_count INTEGER DEFAULT 0,
read INTEGER DEFAULT 0,
deleted INTEGER DEFAULT 0,
labels TEXT,
thread_id TEXT
);
CREATE TABLE IF NOT EXISTS attachments (
id INTEGER PRIMARY KEY AUTOINCREMENT,
email_id TEXT REFERENCES emails(id) ON DELETE CASCADE,
idx INTEGER,
filename TEXT,
size INTEGER,
content_type TEXT,
content TEXT
);
CREATE TABLE IF NOT EXISTS sync_meta (
key TEXT PRIMARY KEY,
value TEXT
);
`;
const SCHEMA_INDEXES = `
CREATE INDEX IF NOT EXISTS idx_emails_date ON emails(date);
CREATE INDEX IF NOT EXISTS idx_emails_from_domain ON emails(from_domain);
CREATE INDEX IF NOT EXISTS idx_emails_from_address ON emails(from_address);
CREATE INDEX IF NOT EXISTS idx_emails_integration ON emails(integration);
CREATE INDEX IF NOT EXISTS idx_emails_email_account ON emails(email_account);
CREATE INDEX IF NOT EXISTS idx_emails_labels ON emails(labels);
CREATE INDEX IF NOT EXISTS idx_emails_thread ON emails(thread_id);
`;
/** Convert label IDs to lowercase comma-separated string for storage */
function labelsToString(labels?: string[]): string | null {
if (!labels || labels.length === 0) return null;
return labels.map((l) => l.toLowerCase()).join(',');
}
/** Convert stored comma-separated labels back to array */
function labelsFromString(value: unknown): string[] | undefined {
if (typeof value !== 'string' || !value) return undefined;
return value.split(',');
}
function extractAddress(headerValue: string): { name: string; address: string } {
const match = headerValue.match(/^"?(.+?)"?\s*<(.+?)>$/);
if (match) return { name: match[1]!.trim(), address: match[2]!.toLowerCase() };
const bare = headerValue.trim().toLowerCase();
return { name: '', address: bare };
}
function extractDomain(address: string): string {
const at = address.lastIndexOf('@');
return at >= 0 ? address.slice(at + 1) : '';
}
// ── Conversation threading ──
// `id` is sha1(Message-Id) (see resync.messageIdToStableId), so hashing a referenced Message-Id the
// same way yields the *id of that referenced email*. That makes header-based threading trivial:
// a reply's thread_id is the hash of its root Message-Id, which equals the root email's own id.
const hashMsgId = (msgId: string): string => createHash('sha1').update(msgId).digest('hex').slice(0, 16);
/** Ordered Message-Ids this email references (References first, root→leaf; else In-Reply-To). */
function extractReferenceIds(raw: string): string[] {
const refs = extractFullHeader(raw, 'References') || extractFullHeader(raw, 'In-Reply-To');
return Array.from(refs.matchAll(/<([^>]+)>/g), (m) => m[1]!.trim()).filter(Boolean);
}
/** Compute a header-based thread_id for a freshly ingested email (falls back to its own id = new thread). */
function computeThreadId(db: Database, id: string, raw: string): string {
const refIds = extractReferenceIds(raw);
if (refIds.length === 0) return id; // no ancestors → this email is a thread root
const hashed = refIds.map(hashMsgId);
// Adopt an ancestor's thread if we already have one stored (robust to In-Reply-To-only clients).
const placeholders = hashed.map(() => '?').join(',');
const found = db
.query(`SELECT thread_id FROM emails WHERE id IN (${placeholders}) AND thread_id IS NOT NULL LIMIT 1`)
.get(...hashed) as { thread_id: string } | null;
return found?.thread_id ?? hashMsgId(refIds[0]!);
}
const RE_PREFIX = /^\s*((re|fwd?|aw|wg|sv|vs|res|antw)\s*(\[\d+\])?\s*:\s*)+/i;
/** Normalize a subject for fallback grouping: strip reply/forward prefixes, fold whitespace, lowercase. */
function normalizeSubject(subject: string): string {
return subject.replace(RE_PREFIX, '').replace(/\s+/g, ' ').trim().toLowerCase();
}
const firstAddress = (value: unknown): string => {
if (typeof value !== 'string') return '';
const first = value.split(',')[0] ?? '';
return (first.match(/<([^>]+)>/)?.[1] ?? first).trim().toLowerCase();
};
/**
* Subject-based thread key for mail synced before header capture (no References available).
* Groups by normalized subject + the counterpart address, so recurring 1:1 conversations collapse
* while unrelated same-subject mail from different people stays apart. Trivial subjects stay ungrouped.
*/
function fallbackThreadId(row: {
id: string;
subject: unknown;
from_address: unknown;
to_address: unknown;
email_account: unknown;
}): string {
const norm = normalizeSubject(typeof row.subject === 'string' ? row.subject : '');
if (!norm) return row.id;
const me = typeof row.email_account === 'string' ? row.email_account.toLowerCase() : '';
const from = typeof row.from_address === 'string' ? row.from_address.toLowerCase() : '';
const counterpart = from && from !== me ? from : firstAddress(row.to_address) || from;
return `s:${norm}|${counterpart}`;
}
export function openEmailDb(ownerEmail: string, accountEmail: string): Database {
const dbPath = getEmailDbPath(ownerEmail, accountEmail);
mkdirSync(dirname(dbPath), { recursive: true });
const db = new Database(dbPath, { create: true });
// The API and the email sidecar both open this file; wait out a concurrent writer (e.g. a resync
// or the one-time thread_id backfill) instead of failing immediately with "database is locked".
db.exec('PRAGMA busy_timeout = 5000');
db.exec('PRAGMA journal_mode = DELETE');
db.exec('PRAGMA foreign_keys = ON');
db.exec(SCHEMA_TABLES);
migrate(db);
db.exec(SCHEMA_INDEXES);
ensureFts(db);
// Try to chmod, but don't crash if permission denied (e.g., file owned by different user)
try {
chmodSync(dbPath, 0o666);
} catch (err) {
// File exists with correct permissions, or owned by another user - that's fine
}
return db;
}
/**
* Open the email DB for a user's configured account. Emails are stored per account
* (email_accounts/<account>/emails.db); for now we use the user's first account. Returns null if the
* user has no email account configured yet.
*/
export async function openUserEmailDb(ownerEmail: string, userId: number): Promise<Database | null> {
const accounts = await getEmailAccounts(userId);
const account = accounts.find((a) => a.enabled) ?? accounts[0];
return account ? openEmailDb(ownerEmail, account.email) : null;
}
function migrate(db: Database): void {
const cols = db.query('PRAGMA table_info(emails)').all() as Array<{ name: string }>;
const colNames = new Set(cols.map((c) => c.name));
if (!colNames.has('deleted')) {
db.exec('ALTER TABLE emails ADD COLUMN deleted INTEGER DEFAULT 0');
}
if (!colNames.has('integration')) {
db.exec("ALTER TABLE emails ADD COLUMN integration TEXT NOT NULL DEFAULT 'gmail'");
}
if (!colNames.has('email_account')) {
db.exec("ALTER TABLE emails ADD COLUMN email_account TEXT NOT NULL DEFAULT ''");
}
if (!colNames.has('labels')) {
db.exec('ALTER TABLE emails ADD COLUMN labels TEXT');
}
if (!colNames.has('thread_id')) {
db.exec('ALTER TABLE emails ADD COLUMN thread_id TEXT');
}
// Backfill thread_id for any rows missing it. Header data isn't kept for already-synced mail, so
// these use the subject-based fallback. New mail gets an exact header-based thread_id at insert.
backfillThreadIds(db);
// Ensure sync_meta table exists (for DBs created before it was added to SCHEMA_TABLES)
db.exec('CREATE TABLE IF NOT EXISTS sync_meta (key TEXT PRIMARY KEY, value TEXT)');
// Add content column to attachments if missing
const attCols = db.query('PRAGMA table_info(attachments)').all() as Array<{ name: string }>;
const attColNames = new Set(attCols.map((c) => c.name));
if (!attColNames.has('content')) {
db.exec('ALTER TABLE attachments ADD COLUMN content TEXT');
}
}
function backfillThreadIds(db: Database): void {
const rows = db
.query('SELECT id, subject, from_address, to_address, email_account FROM emails WHERE thread_id IS NULL')
.all() as Array<{
id: string;
subject: unknown;
from_address: unknown;
to_address: unknown;
email_account: unknown;
}>;
if (rows.length === 0) return;
const update = db.prepare('UPDATE emails SET thread_id = ? WHERE id = ?');
db.exec('BEGIN');
try {
for (const row of rows) update.run(fallbackThreadId(row), row.id);
db.exec('COMMIT');
} catch (err) {
db.exec('ROLLBACK');
throw err;
}
}
// ── Full-text search (FTS5) ──
// A standalone FTS5 index kept in sync with `emails` on every upsert. unicode61 + diacritic folding
// gives accent-insensitive matching; per-term prefix queries make it feel incremental.
function ensureFts(db: Database): void {
db.exec(
`CREATE VIRTUAL TABLE IF NOT EXISTS emails_fts USING fts5(
id UNINDEXED, subject, sender, recipients, snippet, body,
tokenize = 'unicode61 remove_diacritics 2'
)`,
);
const fts = (db.query('SELECT count(*) AS c FROM emails_fts').get() as { c: number }).c;
const total = (db.query('SELECT count(*) AS c FROM emails').get() as { c: number }).c;
if (fts === 0 && total > 0) {
// Backfill existing mail (first run after this feature ships).
db.exec(
`INSERT INTO emails_fts (id, subject, sender, recipients, snippet, body)
SELECT id, COALESCE(subject, ''),
TRIM(COALESCE(from_name, '') || ' ' || COALESCE(from_address, '')),
TRIM(COALESCE(to_address, '') || ' ' || COALESCE(cc, '')),
COALESCE(snippet, ''), COALESCE(text_body, '')
FROM emails`,
);
}
}
const ftsDeleteStmt = 'DELETE FROM emails_fts WHERE id = ?';
const ftsInsertStmt =
'INSERT INTO emails_fts (id, subject, sender, recipients, snippet, body) VALUES (?, ?, ?, ?, ?, ?)';
function syncFtsRow(
db: Database,
id: string,
subject: string,
sender: string,
recipients: string,
snippet: string,
body: string,
): void {
db.run(ftsDeleteStmt, [id]);
db.run(ftsInsertStmt, [id, subject, sender, recipients, snippet, body]);
}
// Gmail-style query parsing. Free text → full-text (prefix-AND). Operators:
// from:/to:/subject:/body: → FTS5 column filters
// has:attachment, is:unread/read, label:X, before:/after:YYYY-MM-DD → SQL filters on `emails`
// Unknown operators fall back to free text. Quoted values (from:"a b") match as an exact phrase.
const FTS_COLUMNS: Record<string, string> = { from: 'sender', to: 'recipients', subject: 'subject', body: 'body' };
const ftsTerm = (value: string, column: string | null, prefix: boolean): string => {
const esc = value.replace(/"/g, '""');
return `${column ? column + ':' : ''}"${esc}"${prefix ? '*' : ''}`;
};
const parseSearchDate = (v: string): string | null => {
const iso = /^\d{4}-\d{2}-\d{2}$/.test(v) ? `${v}T00:00:00.000Z` : v;
const d = new Date(iso);
return Number.isNaN(d.getTime()) ? null : d.toISOString();
};
// One AND-group of terms (a single OR branch). `fts` is the FTS5 MATCH expression for this branch;
// `where`/`params` are its structured SQL filters.
type Branch = { fts: string; where: string[]; params: string[] };
function parseBranch(q: string): Branch {
const fts: string[] = [];
const where: string[] = [];
const params: string[] = [];
const re = /(\w+):("[^"]*"|\S+)|"([^"]*)"|(\S+)/g;
let m: RegExpExecArray | null;
while ((m = re.exec(q)) !== null) {
if (m[1]) {
const op = m[1].toLowerCase();
const raw = m[2]!;
const quoted = raw.startsWith('"') && raw.endsWith('"');
const val = quoted ? raw.slice(1, -1) : raw;
if (!val) continue;
const lower = val.toLowerCase();
if (FTS_COLUMNS[op]) {
fts.push(ftsTerm(val, FTS_COLUMNS[op], !quoted));
} else if (op === 'label') {
where.push('e.labels LIKE ?');
params.push(`%${lower}%`);
} else if (op === 'has' && (lower === 'attachment' || lower === 'attachments')) {
where.push('e.attachment_count > 0');
} else if (op === 'is' && (lower === 'unread' || lower === 'read')) {
where.push(lower === 'unread' ? 'e.read = 0' : 'e.read = 1');
} else if (op === 'before' || op === 'older') {
const d = parseSearchDate(val);
if (d) {
where.push('e.date < ?');
params.push(d);
}
} else if (op === 'after' || op === 'newer') {
const d = parseSearchDate(val);
if (d) {
where.push('e.date >= ?');
params.push(d);
}
} else {
// Unknown operator — treat the whole "op:val" token as free text.
fts.push(ftsTerm(`${op}:${val}`, null, !quoted));
}
} else if (m[3] !== undefined) {
if (m[3].trim()) fts.push(ftsTerm(m[3], null, false)); // quoted phrase → exact
} else if (m[4]) {
fts.push(ftsTerm(m[4], null, true)); // bare word → prefix
}
}
return { fts: fts.join(' '), where, params };
}
export function searchEmails(
db: Database,
q: string,
limit: number,
offset: number,
): { rows: Record<string, unknown>[]; total: number } {
// Split on top-level uppercase OR into branches (Gmail-style; lowercase "or" stays a search word).
const branches = q
.split(/\s+OR\s+/)
.map(parseBranch)
.filter((b) => b.fts || b.where.length > 0);
if (branches.length === 0) return { rows: [], total: 0 };
// Each branch becomes one self-contained condition: its FTS terms via an `id IN (FTS subquery)` so
// full-text and structured filters share a WHERE and branches can be OR'd. All ANDed within a branch.
const conds: string[] = [];
const params: string[] = [];
for (const b of branches) {
const parts: string[] = [];
if (b.fts) {
parts.push('e.id IN (SELECT emails_fts.id FROM emails_fts WHERE emails_fts MATCH ?)');
params.push(b.fts);
}
parts.push(...b.where);
params.push(...b.params);
conds.push(`(${parts.join(' AND ')})`);
}
const whereSql = `e.deleted = 0 AND (${conds.join(' OR ')})`;
const rows = db
.query(`SELECT e.* FROM emails e WHERE ${whereSql} ORDER BY e.date DESC LIMIT ? OFFSET ?`)
.all(...params, limit, offset) as Record<string, unknown>[];
const total = (db.query(`SELECT count(*) AS c FROM emails e WHERE ${whereSql}`).get(...params) as { c: number }).c;
return { rows, total };
}
type ParsedEmail = {
id: string;
integration: string;
emailAccount: string;
fromName: string;
fromAddress: string;
to: string;
cc?: string;
subject: string;
date: string;
snippet: string;
html?: string;
text?: string;
attachments: Array<{ filename: string; size: number; contentType: string; content: string }>;
labels?: string[];
threadId?: string;
};
const upsertEmailStmt = `
INSERT OR REPLACE INTO emails (id, integration, email_account, from_name, from_address, from_domain, to_address, cc, subject, date, snippet, html, text_body, attachment_count, labels, thread_id)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`;
const deleteAttachmentsStmt = 'DELETE FROM attachments WHERE email_id = ?';
const insertAttachmentStmt =
'INSERT INTO attachments (email_id, idx, filename, size, content_type, content) VALUES (?, ?, ?, ?, ?, ?)';
export function upsertEmail(db: Database, email: ParsedEmail): void {
const domain = extractDomain(email.fromAddress);
db.exec('BEGIN');
try {
db.run(upsertEmailStmt, [
email.id,
email.integration,
email.emailAccount,
email.fromName,
email.fromAddress,
domain,
email.to,
email.cc ?? null,
email.subject,
email.date,
email.snippet,
email.html ?? null,
email.text ?? null,
email.attachments.length,
labelsToString(email.labels),
email.threadId ?? email.id,
]);
db.run(deleteAttachmentsStmt, [email.id]);
for (let i = 0; i < email.attachments.length; i++) {
const att = email.attachments[i]!;
db.run(insertAttachmentStmt, [email.id, i, att.filename, att.size, att.contentType, att.content]);
}
syncFtsRow(
db,
email.id,
email.subject ?? '',
`${email.fromName ?? ''} ${email.fromAddress ?? ''}`.trim(),
`${email.to ?? ''} ${email.cc ?? ''}`.trim(),
email.snippet ?? '',
email.text ?? '',
);
db.exec('COMMIT');
} catch (err) {
db.exec('ROLLBACK');
throw err;
}
}
/** Upsert a single email from its raw RFC822 text using fast header parsing. */
type UpsertFromRawEmlParams = {
db: Database;
id: string;
raw: string;
integration: string;
emailAccount: string;
labels?: string[];
};
export function upsertFromRawEml({ db, id, raw, integration, emailAccount, labels }: UpsertFromRawEmlParams): void {
const from = extractHeader(raw, 'From');
const { name, address } = extractAddress(from);
const to = extractHeader(raw, 'To');
const cc = extractHeader(raw, 'Cc') || null;
const subject = extractHeader(raw, 'Subject') || '(no subject)';
const dateStr = extractHeader(raw, 'Date');
const date = dateStr ? new Date(dateStr).toISOString() : new Date(0).toISOString();
const snippet = extractSnippet(raw);
const attachments = parseAttachments(raw);
const domain = extractDomain(address);
const { html, text } = extractBody(raw);
const threadId = computeThreadId(db, id, raw);
db.run(upsertEmailStmt, [
id,
integration,
emailAccount,
name,
address,
domain,
to,
cc,
subject,
date,
snippet,
html,
text,
attachments.length,
labelsToString(labels),
threadId,
]);
if (attachments.length > 0) {
db.run(deleteAttachmentsStmt, [id]);
for (let i = 0; i < attachments.length; i++) {
const att = attachments[i]!;
db.run(insertAttachmentStmt, [id, i, att.filename, att.size, att.contentType, att.content]);
}
}
syncFtsRow(db, id, subject, `${name} ${address}`.trim(), `${to} ${cc ?? ''}`.trim(), snippet, text ?? '');
}
/** Convert a db row to an EmailSummary for the API */
export function rowToSummary(row: Record<string, unknown>): EmailSummary {
const from = row.from_name ? `${row.from_name} <${row.from_address}>` : (row.from_address as string);
const labels = labelsFromString(row.labels);
return {
id: row.id as string,
from,
to: row.to_address as string,
subject: row.subject as string,
date: row.date as string,
snippet: row.snippet as string,
...(row.attachment_count ? { attachmentCount: row.attachment_count as number } : {}),
...(row.read ? { read: true } : {}),
...(labels ? { labels } : {}),
...(row.thread_count && (row.thread_count as number) > 1 ? { threadCount: row.thread_count as number } : {}),
...(row.thread_unread ? { threadUnread: row.thread_unread as number } : {}),
};
}
// ── Sync meta helpers ──
export function getSyncMeta(db: Database, key: string): string | null {
const row = db.query('SELECT value FROM sync_meta WHERE key = ?').get(key) as { value: string } | null;
return row?.value ?? null;
}
export function setSyncMeta(db: Database, key: string, value: string): void {
db.run('INSERT OR REPLACE INTO sync_meta (key, value) VALUES (?, ?)', [key, value]);
}
export function updateEmailLabels(db: Database, id: string, labels: string[]): void {
db.run('UPDATE emails SET labels = ? WHERE id = ?', [labelsToString(labels), id]);
}
// ── Header parsing helpers (same logic as gmail-sync) ──
function decodeMimeWords(text: string): string {
return text.replace(/=\?([^?]+)\?(B|Q)\?([^?]+)\?=/gi, (_, charset, encoding, encoded) => {
try {
const normalizedCs = normalizeCharset(charset.toLowerCase());
if (encoding.toUpperCase() === 'B') {
const buf = Buffer.from(encoded, 'base64');
return new TextDecoder(normalizedCs, { fatal: false }).decode(buf);
}
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 new TextDecoder(normalizedCs, { fatal: false }).decode(Buffer.from(bytes));
} 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()) : '';
}
/** Extract a header value including folded continuation lines (lines starting with whitespace) */
function extractFullHeader(raw: string, name: string): string {
const headerEnd = findHeaderEnd(raw);
const headerBlock = headerEnd !== -1 ? raw.slice(0, headerEnd) : raw.slice(0, 4096);
const lines = headerBlock.split(/\r?\n/);
let result = '';
let capturing = false;
for (const line of lines) {
if (new RegExp(`^${name}:\\s*`, 'i').test(line)) {
result = line.replace(new RegExp(`^${name}:\\s*`, 'i'), '');
capturing = true;
} else if (capturing && /^[\t ]/.test(line)) {
result += ' ' + line.trim();
} else if (capturing) {
break;
}
}
return result.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 (body.trimStart().startsWith('--')) {
const afterBoundary = body.slice(body.indexOf('\n') + 1);
const partBodyStart = findHeaderEnd(afterBoundary);
if (partBodyStart !== -1) body = afterBoundary.slice(partBodyStart);
}
const nextBoundary = body.indexOf('\n--');
if (nextBoundary !== -1) body = body.slice(0, nextBoundary);
return body.replace(/\s+/g, ' ').trim().slice(0, 120);
}
function decodeQuotedPrintableBytes(text: string): Buffer {
const cleaned = text.replace(/=\r?\n/g, '');
const bytes: number[] = [];
for (let i = 0; i < cleaned.length; i++) {
if (cleaned[i] === '=' && i + 2 < cleaned.length) {
const hex = cleaned.slice(i + 1, i + 3);
const val = parseInt(hex, 16);
if (!isNaN(val)) {
bytes.push(val);
i += 2;
continue;
}
}
bytes.push(cleaned.charCodeAt(i));
}
return Buffer.from(bytes);
}
function extractCharset(contentType: string): string {
const match = contentType.match(/charset=["']?([^"';\s]+)/i);
return match?.[1]?.toLowerCase() ?? 'utf-8';
}
function normalizeCharset(charset: string): string {
const map: Record<string, string> = {
'iso-8859-1': 'latin1',
'iso_8859-1': 'latin1',
'iso-8859-15': 'latin1',
'iso_8859-15': 'latin1',
'windows-1250': 'latin1',
'windows-1251': 'latin1',
'windows-1252': 'latin1',
'windows-1254': 'latin1',
'us-ascii': 'ascii',
ascii: 'ascii',
};
return map[charset] ?? charset;
}
function decodePartBody(body: string, encoding: string, charset = 'utf-8'): string {
const enc = encoding.toLowerCase();
const normalizedCharset = normalizeCharset(charset);
if (enc === 'base64') {
const buf = Buffer.from(body.replace(/\s/g, ''), 'base64');
return new TextDecoder(normalizedCharset, { fatal: false }).decode(buf);
}
if (enc === 'quoted-printable') {
const buf = decodeQuotedPrintableBytes(body);
return new TextDecoder(normalizedCharset, { fatal: false }).decode(buf);
}
return body;
}
function extractBody(raw: string): { html: string | null; text: string | null } {
const headerEnd = findHeaderEnd(raw);
if (headerEnd === -1) return { html: null, text: null };
const topCtRaw = extractFullHeader(raw, 'Content-Type');
const topCt = topCtRaw.toLowerCase();
const topEncoding = extractFullHeader(raw, 'Content-Transfer-Encoding');
// Non-multipart: single body
if (!topCt.includes('multipart')) {
const body = raw.slice(headerEnd);
const charset = extractCharset(topCtRaw);
const decoded = decodePartBody(body, topEncoding, charset);
if (topCt.includes('text/html')) return { html: decoded, text: null };
return { html: null, text: decoded };
}
// Multipart: extract boundary from the raw (case-sensitive) header
const boundaryMatch = topCtRaw.match(/boundary=["']?([^"';\s]+)/i);
if (!boundaryMatch) return { html: null, text: null };
const boundary = boundaryMatch[1]!;
let html: string | null = null;
let text: string | null = null;
const parts = raw.slice(headerEnd).split(`--${boundary}`);
for (const part of parts) {
if (part.startsWith('--') || !part.trim()) continue;
const partHeaderEnd = findHeaderEnd(part);
if (partHeaderEnd === -1) continue;
const partCtRaw = extractFullHeader(part, 'Content-Type');
const partCt = partCtRaw.toLowerCase();
const partEnc = extractFullHeader(part, 'Content-Transfer-Encoding');
const partCharset = extractCharset(partCtRaw);
const partBody = part.slice(partHeaderEnd);
// Recurse into nested multipart (e.g. multipart/alternative inside multipart/mixed)
if (partCt.includes('multipart')) {
const nested = extractBody(part.trim());
if (nested.html && !html) html = nested.html;
if (nested.text && !text) text = nested.text;
continue;
}
if (partCt.includes('text/html') && !html) {
html = decodePartBody(partBody, partEnc, partCharset);
} else if (partCt.includes('text/plain') && !text) {
text = decodePartBody(partBody, partEnc, partCharset);
}
}
return { html, text };
}
type AttachmentMeta = { filename: string; size: number; contentType: string; content: string };
function parseAttachments(raw: string): AttachmentMeta[] {
const results: AttachmentMeta[] = [];
// Match both "attachment" and "inline" dispositions
const regex = /^Content-Disposition:\s*(?:attachment|inline)[^\n]*/gim;
let match: RegExpExecArray | null;
while ((match = regex.exec(raw)) !== null) {
const pos = match.index;
// Walk backwards to find the start of this MIME part's headers
const partStart = raw.lastIndexOf('\n--', pos);
const headerBlock =
partStart !== -1 ? raw.slice(partStart, pos + 500) : raw.slice(Math.max(0, pos - 500), pos + 500);
// Skip inline parts without a filename (e.g. inline text/plain body parts)
const hasFilename = /filename/i.test(headerBlock);
if (!hasFilename) continue;
// Extract filename from Content-Disposition or Content-Type
const fnMatch = headerBlock.match(/filename\*?=(?:"([^"]+)"|([^\s;]+))/i);
let rawFilename = fnMatch ? (fnMatch[1] ?? fnMatch[2] ?? 'unknown') : 'unknown';
// RFC 5987: filename*=charset''percent-encoded
const rfc5987Match = rawFilename.match(/^([^']*)'[^']*'(.+)/);
if (rfc5987Match) {
const cs = normalizeCharset(rfc5987Match[1]!.toLowerCase() || 'utf-8');
const encoded = rfc5987Match[2]!;
const bytes = encoded.replace(/%([0-9A-Fa-f]{2})/g, (_, h: string) => String.fromCharCode(parseInt(h, 16)));
rawFilename = new TextDecoder(cs, { fatal: false }).decode(Buffer.from(bytes, 'binary'));
}
const filename = decodeMimeWords(rawFilename);
// Extract content-type
const ctMatch = headerBlock.match(/^Content-Type:\s*([^\s;]+)/im);
const contentType = ctMatch?.[1] ?? 'application/octet-stream';
// Extract full body content as base64
const partHeaderEnd = findHeaderEnd(raw.slice(pos));
let content = '';
let size = 0;
if (partHeaderEnd !== -1) {
const bodyStart = pos + partHeaderEnd;
const boundaryEnd = raw.indexOf('\n--', bodyStart);
const bodyRaw = boundaryEnd !== -1 ? raw.slice(bodyStart, boundaryEnd) : raw.slice(bodyStart);
// Detect encoding from part headers
const encMatch = headerBlock.match(/^Content-Transfer-Encoding:\s*(\S+)/im);
const encoding = encMatch?.[1]?.toLowerCase() ?? 'base64';
if (encoding === 'base64') {
content = bodyRaw.replace(/\s/g, '');
} else {
// For quoted-printable or 7bit/8bit, re-encode to base64
const buf = encoding === 'quoted-printable' ? decodeQuotedPrintableBytes(bodyRaw) : Buffer.from(bodyRaw);
content = buf.toString('base64');
}
size = Math.floor((content.length * 3) / 4);
}
results.push({ filename, size, contentType, content });
}
return results;
}
+124
View File
@@ -0,0 +1,124 @@
import { describe, expect, it } from 'bun:test';
import { Database } from 'bun:sqlite';
import { readSyncMeta, writeSyncMeta, hasSyncedBefore } from './sync-meta';
// Moving an account's sync position out of Postgres and into its own emails.db. Both failure modes are
// expensive and neither is loud, so they are pinned here:
//
// migrate when we should not -> a restored older mailbox inherits a newer position and SKIPS mail
// fail to migrate -> an existing account looks brand new and re-downloads everything
const freshDb = () => {
const db = new Database(':memory:');
db.exec('CREATE TABLE IF NOT EXISTS sync_meta (key TEXT PRIMARY KEY, value TEXT)');
return db;
};
describe('readSyncMeta', () => {
it('reads what the file already knows', () => {
const db = freshDb();
writeSyncMeta(db, { last_sync_at: '2026-01-01T00:00:00Z', 'imap_lastuid:INBOX': '4200' });
expect(readSyncMeta(db)).toEqual({ last_sync_at: '2026-01-01T00:00:00Z', 'imap_lastuid:INBOX': '4200' });
});
it('migrates keys the file is MISSING, even when it already has others', () => {
// The case that broke a live 6.9 GB mailbox. The older Gmail backfill wrote last_sync_at and the
// uidvalidity set into SQLite and never the imap_lastuid ones, so the file was non-empty and
// half-migrated at once. All-or-nothing skipped it, leaving no lastuid for any folder — and a
// missing lastuid refetches that folder from UID 1.
const db = freshDb();
writeSyncMeta(db, { last_sync_at: '2026-07-23T00:00:00Z', 'imap_uidvalidity:INBOX': '2' });
const meta = readSyncMeta(db, {
last_sync_at: '2026-08-10T00:00:00Z', // newer, but the file's own answer stands
'imap_uidvalidity:INBOX': '2',
'imap_lastuid:INBOX': '4200', // absent from the file — this is what must arrive
});
expect(meta['imap_lastuid:INBOX']).toBe('4200');
expect(meta.last_sync_at).toBe('2026-07-23T00:00:00Z');
// Persisted, so the fallback is not needed next time.
expect(readSyncMeta(db)['imap_lastuid:INBOX']).toBe('4200');
});
it('migrates from Postgres exactly once, when the file has nothing', () => {
// The account existed before the move. Without this it looks like a first sync and re-downloads
// the entire mailbox.
const db = freshDb();
const pg = { last_sync_at: '2026-01-01T00:00:00Z', 'imap_uidvalidity:INBOX': '99' };
expect(readSyncMeta(db, pg)).toEqual({ last_sync_at: '2026-01-01T00:00:00Z', 'imap_uidvalidity:INBOX': '99' });
// Backfilled, so the fallback is never consulted again.
expect(readSyncMeta(db)).toEqual({ last_sync_at: '2026-01-01T00:00:00Z', 'imap_uidvalidity:INBOX': '99' });
});
it('NEVER lets Postgres override a position the file already has', () => {
// The dangerous direction. Restore an older emails.db onto a server whose Postgres row is newer,
// and topping up from Postgres would advance past mail this file does not contain — skipped
// permanently, because nothing looks below lastuid again. The file wins, always.
const db = freshDb();
writeSyncMeta(db, { 'imap_lastuid:INBOX': '100' });
const meta = readSyncMeta(db, { 'imap_lastuid:INBOX': '9999' });
// The key the file has is never overwritten — that is what stops a restored older mailbox being
// advanced past mail it does not contain.
expect(meta['imap_lastuid:INBOX']).toBe('100');
});
it('ignores non-string values rather than stringifying them', () => {
// The Postgres column is free-form jsonb. An object coerced to "[object Object]" would be a
// position that parses as a number nowhere and compares as garbage.
const db = freshDb();
expect(readSyncMeta(db, { good: 'yes', bad: { nested: true }, alsoBad: 42 } as never)).toEqual({ good: 'yes' });
});
it('is empty for a genuinely new account', () => {
expect(readSyncMeta(freshDb(), {})).toEqual({});
expect(readSyncMeta(freshDb(), null)).toEqual({});
});
});
describe('writeSyncMeta', () => {
it('upserts rather than duplicating, so a checkpoint mid-sync is safe to repeat', () => {
const db = freshDb();
writeSyncMeta(db, { 'imap_lastuid:INBOX': '10' });
writeSyncMeta(db, { 'imap_lastuid:INBOX': '20' });
expect(readSyncMeta(db)).toEqual({ 'imap_lastuid:INBOX': '20' });
});
it('skips null and undefined instead of writing them as text', () => {
// "null" as a stored lastuid would parse to NaN and compare false against every real UID.
const db = freshDb();
writeSyncMeta(db, { a: '1', b: null, c: undefined });
expect(readSyncMeta(db)).toEqual({ a: '1' });
});
it('stores numbers as their text form, since UIDs arrive both ways', () => {
const db = freshDb();
writeSyncMeta(db, { 'imap_lastuid:INBOX': 4200 });
expect(readSyncMeta(db)['imap_lastuid:INBOX']).toBe('4200');
});
});
describe('hasSyncedBefore', () => {
it('is false for a new account and true once the file records a sync', () => {
const db = freshDb();
expect(hasSyncedBefore(db)).toBe(false);
writeSyncMeta(db, { last_sync_at: '2026-01-01T00:00:00Z' });
expect(hasSyncedBefore(db)).toBe(true);
});
it('is true for an account that synced before the move but has not migrated yet', () => {
// The whole point: an unmigrated account must not be treated as new. This is the check that stands
// between an existing user and an unnecessary full re-sync of their mailbox.
expect(hasSyncedBefore(freshDb(), { last_sync_at: '2026-01-01T00:00:00Z' })).toBe(true);
});
it('is false when Postgres holds a position but no completed sync', () => {
// Folder UIDs without last_sync_at means a sync started and never finished.
expect(hasSyncedBefore(freshDb(), { 'imap_uidvalidity:INBOX': '99' })).toBe(false);
});
});
+94
View File
@@ -0,0 +1,94 @@
import type { Database } from 'bun:sqlite';
import { getSyncMeta, setSyncMeta } from './store';
// Where an account's sync position lives: IN ITS OWN emails.db, not in Postgres.
//
// ── Why this moved ──
//
// The messages are in `emails.db` and the position — `last_sync_at`, and per-folder `uidvalidity` +
// `lastuid` — used to be a jsonb column on `email_accounts` in Postgres. Two stores for one fact, and
// the split had a sharp edge that only appears when you try to move a mailbox to another machine.
//
// The expensive thing about an email account is the initial sync: hours of IMAP for a large mailbox. So
// "copy emails.db to the new server" is the obvious way to bring an account across. With the position in
// Postgres that silently does not work — the new server's `sync_meta` is empty, `!last_sync_at` says
// first sync, and it downloads the whole mailbox again on top of the one you just restored.
//
// The worse direction is quieter. Restore an OLDER emails.db while Postgres still holds a NEWER
// position, and the sidecar skips every message between the two — permanently, because nothing ever
// looks below `lastuid` again. Re-syncing is slow; skipping mail is data loss you do not notice.
//
// Keeping both in one file makes the file self-describing: it says what it holds and where it got to,
// and the two cannot disagree because they travel together.
//
// ── Why this is a generalisation, not a new idea ──
//
// The Gmail path already did exactly this (`gmail-api.ts`): read SQLite, fall back to the Postgres
// column, backfill SQLite so the fallback is only ever taken once. That was the right shape and only
// the IMAP path had not followed. This is that pattern, extracted so both use one copy of it.
//
// ── The Postgres column stays ──
//
// `email_accounts.sync_meta` is not dropped. It is the backfill source for every account that exists
// today, and it costs nothing to leave. It is no longer written to; see `readSyncMeta` for the one-way
// migration, which happens the first time an account syncs after this change.
/**
* The whole sync position for this account, from SQLite — migrating it out of Postgres once if this is
* the first read since the change.
*
* `pgFallback` is the Postgres jsonb as it stands. Pass it; the migration cannot happen without it, and
* an account whose position never migrates re-syncs its entire mailbox.
*/
export function readSyncMeta(db: Database, pgFallback?: Record<string, unknown> | null): Record<string, string> {
const rows = db.query('SELECT key, value FROM sync_meta').all() as Array<{ key: string; value: string }>;
const meta: Record<string, string> = {};
for (const row of rows) meta[row.key] = row.value;
// Merged PER KEY, and the file always wins a conflict.
//
// This started life as all-or-nothing — migrate only when SQLite is completely empty — on the
// assumption that a non-empty file is an authoritative one. A real account disproved it immediately.
// The older Gmail backfill wrote SOME keys into SQLite (`last_sync_at`, the `uidvalidity` set) and
// never the `imap_lastuid:*` ones, so the file was non-empty and half-migrated at the same time.
// All-or-nothing therefore skipped the migration, leaving no lastuid for any folder — and a missing
// lastuid means the next sync refetches that folder from UID 1. On the mailbox this was found on,
// that was 18,755 messages and 6.9 GB.
//
// Per-key with file-wins keeps the property that made all-or-nothing attractive: a restored older
// emails.db still overrides a newer Postgres row for every key it actually has, so it cannot be
// advanced past mail it does not contain. What it adds is the keys the file never had, which are
// exactly the ones whose absence is expensive.
if (pgFallback) {
for (const [key, value] of Object.entries(pgFallback)) {
if (typeof value !== 'string') continue;
if (key in meta) continue; // the file's own answer stands
meta[key] = value;
setSyncMeta(db, key, value);
}
}
return meta;
}
/**
* Persist the position. SQLite only — Postgres is no longer written.
*
* Called after the messages are already stored, which keeps the safe ordering the previous code had: 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.
*/
export function writeSyncMeta(db: Database, meta: Record<string, unknown>): void {
for (const [key, value] of Object.entries(meta)) {
if (value === undefined || value === null) continue;
setSyncMeta(db, key, String(value));
}
}
/** Has this account ever completed a sync? The question `isFirstSync` is really asking. */
export function hasSyncedBefore(db: Database, pgFallback?: Record<string, unknown> | null): boolean {
if (getSyncMeta(db, 'last_sync_at')) return true;
// Not yet migrated: an account that synced before this change has its answer in Postgres, and
// treating it as a first sync would re-download the whole mailbox.
const pgValue = pgFallback?.last_sync_at;
return typeof pgValue === 'string' && pgValue.length > 0;
}
+94
View File
@@ -0,0 +1,94 @@
// Sync scheduling, owned by the sidecar.
//
// These syncs used to be platform queue jobs, which gave them retries and a row in the Jobs list. That was
// the wrong home twice over: the credentials travelled through Postgres job metadata to get there, and a
// mailbox sync has nothing to do with what the Jobs screen is for. They run here now, tracked in memory,
// and the Jobs list is left to the things it actually describes.
//
// Deliberately NOT a queue: one run per account at a time, no persistence, no retry. A failed sync is
// retried by the ten-minute cron like any other, and a sync interrupted by a restart resumes from the
// stored sync cursor rather than from the beginning.
/** What a sync reports as it goes. `total: 0` means "unknown" — IMAP does not tell us up front. */
export type SyncProgress = { current: number; total: number; label: string };
export type SyncState = {
accountId: number;
status: 'running' | 'completed' | 'failed';
progress: SyncProgress | null;
startedAt: number;
finishedAt?: number;
error?: string;
};
const states = new Map<number, SyncState>();
/** The steps of a converted job handler: each is run in order with a synthesized context. */
export type SyncStep = { name: string; run: (ctx: SyncCtx) => Promise<void> | void };
/**
* What the moved handlers expect on their context. `job.userId` is the OWNER'S EMAIL, not a numeric id —
* the queue called it userId and the handlers use it to resolve the mail store path
* (DATA_PATH/<owner>/email_accounts/…). Getting this wrong reads as an empty mailbox rather than an error.
*/
export type SyncCtx = {
/** Same shape the queue's StepContext used, so the moved handlers read it unchanged. */
meta: Record<string, unknown>;
job: { userId: string };
updateProgress: (p: SyncProgress) => Promise<void> | void;
};
export function isSyncing(accountId: number): boolean {
return states.get(accountId)?.status === 'running';
}
export function getSyncStates(): SyncState[] {
return [...states.values()];
}
/**
* Run `steps` in the background for one account. Returns immediately — an initial mailbox sync takes
* minutes to hours, so nothing waits on it. Re-entrant calls for an account already syncing are ignored.
*/
export function startSync(params: {
accountId: number;
ownerEmail: string;
meta: Record<string, unknown>;
steps: SyncStep[];
onDone?: (ok: boolean) => Promise<void> | void;
}): SyncState {
const existing = states.get(params.accountId);
if (existing?.status === 'running') return existing;
const state: SyncState = {
accountId: params.accountId,
status: 'running',
progress: null,
startedAt: Date.now(),
};
states.set(params.accountId, state);
const ctx: SyncCtx = {
meta: params.meta,
job: { userId: params.ownerEmail },
updateProgress: (p) => {
state.progress = p;
},
};
void (async () => {
try {
for (const step of params.steps) await step.run(ctx);
state.status = 'completed';
} catch (err) {
state.status = 'failed';
state.error = err instanceof Error ? err.message : String(err);
console.error(`[email-sync] account ${params.accountId} failed:`, state.error);
} finally {
state.finishedAt = Date.now();
await params.onDone?.(state.status === 'completed');
}
})();
return state;
}