email: move the mail store and every route into the sidecar
Email was the one sidecar built inside out. The platform held ~1,800 lines — the per-account
SQLite store, all 14 HTTP routes, account CRUD, resync, IMAP validation — while the 314-line
sidecar was a scheduler that reached BACK into the platform to do anything
(`import { performResync } from '../../api/email/resync'`).
The sidecar now serves its own HTTP listener and announces `email:server`, and
/api/email/* on the platform is createSidecarProxy like every other one: 1,801 lines down
to 22, with no mail knowledge left in it — not a message, not a folder, not a credential.
The routes moved verbatim, Hono and all. http.ts only reconstructs what the platform's
middleware used to provide: `user` on the context, from the X-Officer-User header the proxy
injects (trusted because this server binds loopback), and an error handler that turns
custom-errors into status codes.
The /email/events SSE stream went with them, which removes a whole round trip: the IDLE
watcher used to send `email:new` over the registration socket so the platform could push to
its SSE clients. Those clients are here now, so it calls broadcastEmailNew in-process and
`email:new` is gone from the wire protocol.
DELIBERATELY NOT DONE YET, and left backwards on purpose rather than half-moved:
- The two sync handlers (email-sync 381 lines, gmail-sync 712) still run in the platform's
queue and now import the store from its new home — a platform → sidecar import, which is
the wrong direction and is temporary. Moving them is option (A) from the plan: the sidecar
schedules its own syncs, independent of the platform Jobs list.
- accounts.ts still imports queue/init to enqueue a sync and to report sync status, and
index.ts still carries the queue-over-WS shim that inversion needs.
- The three channel handlers still open the mail store directly rather than asking over HTTP.
Two things worth knowing while testing: a from-scratch sync holds a proxied request open
well past the 60s idle default, hence timeoutSeconds on the proxy; and `gmail-sync` is
hardcoded in all three channel handlers even though the only account is provider=gmail with
auth_type=password, which routes to IMAP — so "sync emails" from a chat channel is
almost certainly already broken, and folds into the next stage.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5
@@ -0,0 +1,259 @@
|
||||
import { createRouter } from '../../create-router';
|
||||
import { BAD_REQUEST, NOT_FOUND } from '../../custom-errors';
|
||||
import {
|
||||
getEmailAccounts,
|
||||
getEmailAccount,
|
||||
createEmailAccount,
|
||||
deleteEmailAccount,
|
||||
updateEmailAccountStatus,
|
||||
} from 'officerdb';
|
||||
import { getValidGoogleAccessToken } from '../../api/integrations/google-auth';
|
||||
import { validateImapConnection } from './imap-validate';
|
||||
import { enqueueJob, listAllJobs } from '../../queue/init';
|
||||
import { openEmailDb, getSyncMeta } from './store';
|
||||
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) {
|
||||
try {
|
||||
const jobs = await listAllJobs();
|
||||
activeJobAccountIds = new Set(
|
||||
jobs
|
||||
.filter((j) => (j.type === 'email-sync' || j.type === 'gmail-sync') && (j.status === 'queued' || j.status === 'running'))
|
||||
.map((j) => (j.meta as Record<string, unknown> | undefined)?.emailAccountId as number)
|
||||
.filter(Boolean),
|
||||
);
|
||||
} catch {
|
||||
// Sidecar unavailable — all syncing/queued accounts are stale
|
||||
}
|
||||
|
||||
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
|
||||
let isFirstSync = true;
|
||||
if (account.provider === 'gmail' && account.authType === 'oauth') {
|
||||
const db = openEmailDb(user.email, account.email);
|
||||
try {
|
||||
isFirstSync = !getSyncMeta(db, 'last_sync_at');
|
||||
} finally {
|
||||
db.close();
|
||||
}
|
||||
} else {
|
||||
const syncMeta = (account.syncMeta ?? {}) as Record<string, unknown>;
|
||||
isFirstSync = !syncMeta.last_sync_at;
|
||||
}
|
||||
|
||||
// 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 jobType = account.provider === 'gmail' && account.authType === 'oauth' ? 'gmail-sync' : 'email-sync';
|
||||
|
||||
const job = await enqueueJob({
|
||||
lane: 'email',
|
||||
type: jobType,
|
||||
userId: user.email,
|
||||
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, jobId: job.id }, 201);
|
||||
});
|
||||
|
||||
accountsRouter.post('/validate', async (ctx) => {
|
||||
const user = ctx.get('user');
|
||||
const body = ctx.get('body') as ValidateBody;
|
||||
|
||||
if (!body.imapHost || !body.imapPort || !body.authType || !body.email) {
|
||||
throw BAD_REQUEST('Missing required fields');
|
||||
}
|
||||
|
||||
const authResult = await resolveAuth(user.id, body.authType, body.email, body.credentials);
|
||||
if (!authResult.ok) return ctx.json({ ok: false, error: authResult.error });
|
||||
|
||||
const result = await validateImapConnection({
|
||||
host: body.imapHost,
|
||||
port: body.imapPort,
|
||||
secure: body.imapSecure,
|
||||
user: body.email,
|
||||
...authResult.auth,
|
||||
});
|
||||
|
||||
return ctx.json(result);
|
||||
});
|
||||
|
||||
type AuthResult = { ok: true; auth: { pass?: string; accessToken?: string } } | { ok: false; error: string };
|
||||
|
||||
async function resolveAuth(
|
||||
userId: number,
|
||||
authType: string,
|
||||
email: string,
|
||||
credentials: Record<string, unknown>,
|
||||
): Promise<AuthResult> {
|
||||
if (authType === 'oauth') {
|
||||
const integrationId = credentials.userIntegrationId as number | undefined;
|
||||
if (!integrationId) return { ok: false, error: 'Missing userIntegrationId for OAuth' };
|
||||
|
||||
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}` };
|
||||
}
|
||||
@@ -1,5 +1,5 @@
|
||||
import { getAllSyncedAccounts, getUserById } from 'officerdb';
|
||||
import { performResync } from '../../api/email/resync';
|
||||
import { performResync } from './resync';
|
||||
|
||||
const INTERVAL_MS = 10 * 60 * 1000; // 10 minutes
|
||||
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import type { ImapFlow } from 'imapflow';
|
||||
import { getAllSyncedAccounts, getUserById } from 'officerdb';
|
||||
import { performResync } from '../../api/email/resync';
|
||||
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
|
||||
|
||||
@@ -0,0 +1,48 @@
|
||||
import { Hono } from 'hono';
|
||||
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 two
|
||||
// things the platform's middleware used to provide: the authenticated user 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 `user` from the JWT. Here it comes from the header the proxy injects —
|
||||
// trusted because this server binds loopback only and nothing else can reach it.
|
||||
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);
|
||||
ctx.set('user', { id } as never);
|
||||
await next();
|
||||
});
|
||||
|
||||
app.route('/', emailRouter as never);
|
||||
|
||||
// custom-errors carry a `status`; anything else is a 500 with no detail leaked to the caller.
|
||||
app.onError((err, ctx) => {
|
||||
const status = (err as { status?: number }).status;
|
||||
if (typeof status === 'number' && status >= 400 && status < 600) {
|
||||
return ctx.json({ error: err.message }, status as 400);
|
||||
}
|
||||
console.error('[email] unhandled error', err);
|
||||
return ctx.json({ error: 'internal 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;
|
||||
}
|
||||
@@ -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 };
|
||||
}
|
||||
}
|
||||
@@ -2,6 +2,8 @@ import type { SidecarEvent } from '../protocol';
|
||||
import type { Job, EnqueueParams } from '../../queue/types';
|
||||
import { initEmailCron, stopEmailCron } from './email-cron';
|
||||
import { initEmailIdle, stopEmailIdle } from './email-idle';
|
||||
import { broadcastEmailNew } from './routes';
|
||||
import { startEmailServer } from './http';
|
||||
import { createSidecarConnector } from '../connect';
|
||||
|
||||
const API_URL = process.env.API_URL ?? `ws://127.0.0.1:${process.env.PORT ?? '5000'}`;
|
||||
@@ -73,6 +75,12 @@ function handleCommand(cmd: Record<string, unknown>, reply: ReplyFn) {
|
||||
}
|
||||
}
|
||||
|
||||
// ── 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({
|
||||
@@ -83,11 +91,16 @@ const connection = createSidecarConnector({
|
||||
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.
|
||||
initEmailIdle((userEmail) => connection.send({ type: 'email:new', userEmail }));
|
||||
// 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));
|
||||
},
|
||||
});
|
||||
|
||||
|
||||
@@ -0,0 +1,309 @@
|
||||
import { createHash } from 'node:crypto';
|
||||
import {
|
||||
getUserIntegration,
|
||||
upsertUserIntegration,
|
||||
getServerIntegration,
|
||||
getEmailAccount,
|
||||
updateEmailAccountSyncMeta,
|
||||
} from 'officerdb';
|
||||
import { refreshGoogleAccessToken } from '@@/api/integrations/google-auth';
|
||||
import { openEmailDb, getSyncMeta, setSyncMeta, upsertFromRawEml } from './store';
|
||||
import {
|
||||
type GmailCredentials,
|
||||
loadGmailCredentials,
|
||||
gmailApiSync,
|
||||
} from '../../queue/handlers/gmail-sync';
|
||||
|
||||
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 syncMeta = (freshAccount.syncMeta ?? {}) as Record<string, unknown>;
|
||||
|
||||
const db = openEmailDb(userEmail, account.email);
|
||||
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();
|
||||
await updateEmailAccountSyncMeta(account.id, syncMeta as Record<string, unknown>);
|
||||
|
||||
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,
|
||||
);
|
||||
}
|
||||
@@ -0,0 +1,433 @@
|
||||
import { mkdir } from 'node:fs/promises';
|
||||
import { join } from 'node:path';
|
||||
import nodemailer from 'nodemailer';
|
||||
import type { EmailMessage } from 'types';
|
||||
import { createRouter } from '../../create-router';
|
||||
import * as errors from '@@/custom-errors';
|
||||
import { getEmailAttachmentCacheDir } from '@@/data-path';
|
||||
import { getEmailAccounts } from 'officerdb';
|
||||
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();
|
||||
}
|
||||
});
|
||||
@@ -0,0 +1,746 @@
|
||||
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 'officerdb';
|
||||
import type { EmailSummary } from '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;
|
||||
}
|
||||
Reference in New Issue
Block a user