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

Merge branch 'sidecars-email' into sidecars

Browse Source
This commit is contained in:
pastilhas
2026-07-30 05:43:01 +00:00
parent b27dd7512b e3e67748c9
commit a5cb7b4246
2 changed files with 38 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/servers/api/email/email.ts
+38 -19
View File
@@ -14,7 +14,9 @@ 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 }> {
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');
@@ -35,9 +37,14 @@ emailRouter.post('/send', async (ctx) => {
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 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 })),
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(
@@ -164,14 +171,24 @@ emailRouter.get('/search', async (ctx) => {
}
});
// `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');
const page = Number(ctx.req.query('page') ?? '1');
const limit = Number(ctx.req.query('limit') ?? '50');
// `|| 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 folderWhere = folder === 'all' ? 'deleted = 0' : `deleted = 0 AND labels LIKE '%${folder}%'`;
const { where: folderWhere, params: folderParams } = folderFilter(folder);
const db = await openUserEmailDb(user.email, user.id);
if (!db) return ctx.json({ messages: [], total: 0 });
@@ -190,10 +207,10 @@ emailRouter.get('/messages', async (ctx) => {
) WHERE rn = 1
ORDER BY date DESC LIMIT ? OFFSET ?`,
)
.all(limit, offset) as Record<string, unknown>[];
.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() as { total: number };
.get(...folderParams) as { total: number };
const messages = rows.map(rowToSummary);
return ctx.json({ messages, total: countRow.total });
} finally {
@@ -202,9 +219,9 @@ emailRouter.get('/messages', async (ctx) => {
});
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 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,
@@ -250,9 +267,10 @@ emailRouter.get('/thread/:id', async (ctx) => {
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;
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;
@@ -363,23 +381,24 @@ emailRouter.get('/stats', async (ctx) => {
const user = ctx.get('user');
const folder = ctx.req.query('folder') ?? 'inbox';
const folderWhere = folder === 'all' ? 'deleted = 0' : `deleted = 0 AND labels LIKE '%${folder}%'`;
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() as { count: number })
.count;
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() as Array<{ from_domain: string; count: number }>;
.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() as Array<{ from_address: string; from_name: string; count: number }>;
.all(...folderParams) as Array<{ from_address: string; from_name: string; count: number }>;
return ctx.json({ total, byDomain, bySender });
} finally {
src/servers/sidecar/email-cron.ts
-92
View File
@@ -1,92 +0,0 @@
import { getAllSyncedAccounts, getUserById } from 'officerdb';
import { getValidGoogleAccessToken } from '../api/integrations/google-auth';
import * as queueRunner from './queue-runner';
const INTERVAL_MS = 10 * 60 * 1000; // 10 minutes
let timer: ReturnType<typeof setInterval> | null = null;
async function tick() {
try {
const accounts = await getAllSyncedAccounts();
if (accounts.length === 0) return;
const allJobs = await queueRunner.listAllJobs();
const activeEmailSyncIds = new Set(
allJobs
.filter((j) => j.type === 'email-sync' && (j.status === 'queued' || j.status === 'running'))
.map((j) => (j.meta as Record<string, unknown> | undefined)?.emailAccountId),
);
for (const account of accounts) {
if (activeEmailSyncIds.has(account.id)) continue;
const user = await getUserById(account.userId);
if (!user) continue;
// Resolve IMAP auth
const imapAuth: Record<string, unknown> = { user: account.email };
if (account.authType === 'oauth') {
let accessToken: string | null = null;
try {
accessToken = await getValidGoogleAccessToken(account.userId);
} catch (err) {
console.log(`[email-cron] Skipping ${account.email}: token refresh failed —`, err instanceof Error ? err.message : err);
continue;
}
if (!accessToken) {
console.log(`[email-cron] Skipping ${account.email}: no OAuth access token`);
continue;
}
imapAuth.accessToken = accessToken;
} else {
const creds = account.credentials as Record<string, unknown>;
imapAuth.pass = creds.password;
}
try {
await queueRunner.enqueue({
lane: 'email',
type: 'email-sync',
userId: user.email,
meta: {
emailAccountId: account.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,
},
});
console.log(`[email-cron] Enqueued incremental sync for ${account.email}`);
} catch (err) {
console.error(`[email-cron] Failed to enqueue sync for ${account.email}:`, err instanceof Error ? err.message : err);
}
}
} catch (err) {
console.error('[email-cron] Error:', err instanceof Error ? err.message : err);
}
}
export function initEmailCron() {
if (timer) return;
console.log(`[email-cron] Starting email sync cron (every ${INTERVAL_MS / 60_000} min)`);
timer = setInterval(tick, INTERVAL_MS);
// Run first tick after a short delay to let the queue initialize
setTimeout(tick, 30_000);
}
export function stopEmailCron() {
if (timer) {
clearInterval(timer);
timer = null;
}
}