From e9e144962d873d2a7e1771689eb514b6854720c8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Padez?= Date: Fri, 31 Jul 2026 13:33:46 +0000 Subject: [PATCH] email: the sidecar schedules its own syncs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Stage 2, and the end of the inversion. The two sync handlers (1,093 lines) ran in the platform's queue, which meant the sidecar reached back over its registration socket to ask the platform to enqueue work, and the credentials travelled through Postgres job metadata to get there. Option (A) from the plan: they run here now, and the Jobs screen is left to the things it actually describes. The handlers moved almost unedited. Their bodies were already a list of steps taking a context, so sync-runner.ts synthesizes that context and runs them; what went away is the JobHandler wrapper and the registration. `job.userId` is the OWNER'S EMAIL rather than a numeric id — the queue's naming — and it resolves the mail store path, so it is called out in the type. That is the same field whose absence made the mailbox read as empty two commits ago; it is set from user.email and checked this time. Deliberately not a queue: one run per account, no persistence, no retry. A failure is picked up by the ten-minute cron like any other, and a sync interrupted by a restart resumes from the stored cursor rather than the beginning. PermanentError survives as a local class — it signalled "do not retry" to the queue and now just carries its message to the sync state. accounts.ts asks the runner whether an account is syncing instead of scanning job rows, and the queue-over-WS shim in index.ts is gone: enqueueViaWs, listJobsViaWs, the pending-response map and the queue branch in the command handler. Nothing but a port crosses that socket now. The three chat channels stop opening the mail store directly. They each carried their own copy of count-rows / enqueue / poll / count-again, coupling three chat bridges to the mail schema — and they enqueued `gmail-sync` unconditionally, the OAuth path, for an app-password account that syncs over IMAP, so the command was already broken. One shared helper calls a new POST /sync-now on the sidecar, which syncs and reports what arrived. queue/handlers/ is now empty; both handlers there were email. The queue is untouched and still serves the Jobs screen. Not moved, and fine where they are: scripts/migrate-emails-to-sqlite.ts and scripts/seed-imap-uids.ts are one-off maintenance scripts that open the store directly. Co-Authored-By: Claude Opus 5 (1M context) --- src/servers/channels/discord/handler.ts | 79 +--------------- src/servers/channels/email-sync-command.ts | 45 +++++++++ src/servers/channels/telegram/handler.ts | 76 +-------------- src/servers/channels/whatsapp/handler.ts | 73 +------------- src/servers/queue/handlers/index.ts | 10 +- src/servers/sidecar/email/accounts.ts | 34 ++++--- .../email/gmail-api.ts} | 21 +++-- .../email/imap-sync.ts} | 20 ++-- src/servers/sidecar/email/index.ts | 51 +--------- src/servers/sidecar/email/resync.ts | 2 +- src/servers/sidecar/email/routes.ts | 29 ++++++ src/servers/sidecar/email/sync-runner.ts | 94 +++++++++++++++++++ 12 files changed, 236 insertions(+), 298 deletions(-) create mode 100644 src/servers/channels/email-sync-command.ts rename src/servers/{queue/handlers/gmail-sync.ts => sidecar/email/gmail-api.ts} (97%) rename src/servers/{queue/handlers/email-sync.ts => sidecar/email/imap-sync.ts} (96%) create mode 100644 src/servers/sidecar/email/sync-runner.ts diff --git a/src/servers/channels/discord/handler.ts b/src/servers/channels/discord/handler.ts index 5a574d58..d348e09a 100644 --- a/src/servers/channels/discord/handler.ts +++ b/src/servers/channels/discord/handler.ts @@ -6,7 +6,6 @@ import { chunkMessage } from './chunker'; import { listChatModels } from '@@/api/chat/list-models'; import { enqueueJob } from '../../queue/init'; import { readJob } from '@@/queue/storage'; -import { openUserEmailDb } from '@@/sidecar/email/store'; import type { ModelInfo } from '@@/api/chat/types'; import { toShellUsername } from '@@/data-path'; @@ -36,6 +35,8 @@ async function getVisibleModels(): Promise { }); } +import { runEmailSyncCommand } from '../email-sync-command'; + type CommandContext = { content: string; channel: SendableChannel; @@ -45,80 +46,10 @@ type CommandContext = { }; async function handleEmailSync(ctx: CommandContext): Promise { - const { channel, email, userId } = ctx; - - // Count emails before sync - let countBefore = 0; - try { - const db = await openUserEmailDb(email, userId); - if (db) { - const row = db.query('SELECT COUNT(*) as count FROM emails WHERE deleted = 0').get() as { count: number }; - countBefore = row.count; - db.close(); - } - } catch { - // DB might not exist yet - } - + const { channel, userId } = ctx; await channel.send('Syncing emails...'); - - const job = await enqueueJob({ lane: 'google-api', type: 'gmail-sync', userId: email }); - - // Poll until done - const poll = async (): Promise<'completed' | 'failed' | 'cancelled'> => { - for (let i = 0; i < 120; i++) { - await new Promise((r) => setTimeout(r, 3000)); - const current = await readJob(job.id); - if (!current) return 'failed'; - if (current.status === 'completed' || current.status === 'failed' || current.status === 'cancelled') { - return current.status; - } - } - return 'failed'; - }; - - const status = await poll(); - - if (status !== 'completed') { - await channel.send('Email sync failed. Check the queue dashboard for details.'); - return; - } - - // Count emails after sync and get newest ones - try { - const db = await openUserEmailDb(email, userId); - if (!db) return; - const row = db.query('SELECT COUNT(*) as count FROM emails WHERE deleted = 0').get() as { count: number }; - const countAfter = row.count; - const newCount = countAfter - countBefore; - - if (newCount <= 0) { - db.close(); - await channel.send('Sync complete — no new emails.'); - return; - } - - const newest = db - .query('SELECT from_name, from_address, subject FROM emails WHERE deleted = 0 ORDER BY date DESC LIMIT ?') - .all(Math.min(newCount, 20)) as Array<{ from_name: string | null; from_address: string; subject: string }>; - db.close(); - - const lines = newest.map((e) => { - const sender = e.from_name || e.from_address; - return `- **${sender}**: ${e.subject}`; - }); - - let text = `Sync complete — **${newCount}** new email${newCount !== 1 ? 's' : ''}`; - if (newCount > 20) text += ` (showing latest 20)`; - text += `:\n\n${lines.join('\n')}`; - - const chunks = chunkMessage(text); - for (const chunk of chunks) { - await channel.send(chunk); - } - } catch (err) { - await channel.send(`Sync complete but failed to read results: ${err instanceof Error ? err.message : String(err)}`); - } + const text = await runEmailSyncCommand(userId); + for (const chunk of chunkMessage(text)) await channel.send(chunk); } async function handleCommand(ctx: CommandContext): Promise { diff --git a/src/servers/channels/email-sync-command.ts b/src/servers/channels/email-sync-command.ts new file mode 100644 index 00000000..c8ef820b --- /dev/null +++ b/src/servers/channels/email-sync-command.ts @@ -0,0 +1,45 @@ +import { getEmailServerUrl } from '../api/email/router'; + +// The "sync my email" chat command, once, for all three channels. +// +// Telegram, Discord and WhatsApp each carried their own copy: open the mail store directly, count rows, +// enqueue a `gmail-sync` job, poll it, count again, diff. That coupled three chat bridges to the mail +// schema, and the job type was hardcoded to the OAuth path even though the account syncs over IMAP — so +// the command was already broken before sync moved into the sidecar. Now it is one HTTP call to the +// sidecar, which does the sync and reports what arrived. + +type SyncNowResponse = { + saved: number; + skipped?: number; + errors?: number; + newest: Array<{ from_name: string | null; from_address: string; subject: string }>; + error?: string; +}; + +/** Runs the sync and returns the message to send back to the user. */ +export async function runEmailSyncCommand(userId: number): Promise { + const base = getEmailServerUrl(); + if (!base) return 'Email is not available right now — the mail service is starting up.'; + + let res: Response; + try { + res = await fetch(`${base}/sync-now`, { + method: 'POST', + // Loopback-only, same trust as the platform's own proxy. + headers: { 'X-Officer-User': String(userId) }, + }); + } catch { + return 'Email sync failed — the mail service is unreachable.'; + } + + if (!res.ok) return `Email sync failed (${res.status}).`; + + const body = (await res.json()) as SyncNowResponse; + if (body.error) return body.error; + if (body.saved <= 0) return 'Sync complete — no new emails.'; + + const lines = body.newest.map((e) => `- *${e.from_name || e.from_address}*: ${e.subject}`); + let text = `Sync complete — *${body.saved}* new email${body.saved !== 1 ? 's' : ''}`; + if (body.saved > 20) text += ' (showing latest 20)'; + return `${text}:\n\n${lines.join('\n')}`; +} diff --git a/src/servers/channels/telegram/handler.ts b/src/servers/channels/telegram/handler.ts index 95e72ee3..6ca1dd54 100644 --- a/src/servers/channels/telegram/handler.ts +++ b/src/servers/channels/telegram/handler.ts @@ -7,7 +7,6 @@ import { getTelegramBot } from './bot'; import { listChatModels } from '@@/api/chat/list-models'; import { enqueueJob } from '../../queue/init'; import { readJob } from '@@/queue/storage'; -import { openUserEmailDb } from '@@/sidecar/email/store'; import type { ModelInfo } from '@@/api/chat/types'; import { toShellUsername } from '@@/data-path'; @@ -37,6 +36,8 @@ async function getVisibleModels(): Promise { }); } +import { runEmailSyncCommand } from '../email-sync-command'; + type CommandContext = { content: string; send: SendFn; @@ -46,77 +47,10 @@ type CommandContext = { }; async function handleEmailSync(ctx: CommandContext): Promise { - const { send, email, userId } = ctx; - - let countBefore = 0; - try { - const db = await openUserEmailDb(email, userId); - if (db) { - const row = db.query('SELECT COUNT(*) as count FROM emails WHERE deleted = 0').get() as { count: number }; - countBefore = row.count; - db.close(); - } - } catch { - // DB might not exist yet - } - + const { send, userId } = ctx; await send('Syncing emails...'); - - const job = await enqueueJob({ lane: 'google-api', type: 'gmail-sync', userId: email }); - - const poll = async (): Promise<'completed' | 'failed' | 'cancelled'> => { - for (let i = 0; i < 120; i++) { - await new Promise((r) => setTimeout(r, 3000)); - const current = await readJob(job.id); - if (!current) return 'failed'; - if (current.status === 'completed' || current.status === 'failed' || current.status === 'cancelled') { - return current.status; - } - } - return 'failed'; - }; - - const status = await poll(); - - if (status !== 'completed') { - await send('Email sync failed. Check the queue dashboard for details.'); - return; - } - - try { - const db = await openUserEmailDb(email, userId); - if (!db) return; - const row = db.query('SELECT COUNT(*) as count FROM emails WHERE deleted = 0').get() as { count: number }; - const countAfter = row.count; - const newCount = countAfter - countBefore; - - if (newCount <= 0) { - db.close(); - await send('Sync complete — no new emails.'); - return; - } - - const newest = db - .query('SELECT from_name, from_address, subject FROM emails WHERE deleted = 0 ORDER BY date DESC LIMIT ?') - .all(Math.min(newCount, 20)) as Array<{ from_name: string | null; from_address: string; subject: string }>; - db.close(); - - const lines = newest.map((e) => { - const sender = e.from_name || e.from_address; - return `- *${sender}*: ${e.subject}`; - }); - - let text = `Sync complete — *${newCount}* new email${newCount !== 1 ? 's' : ''}`; - if (newCount > 20) text += ` (showing latest 20)`; - text += `:\n\n${lines.join('\n')}`; - - const chunks = chunkMessage(text); - for (const chunk of chunks) { - await send(chunk); - } - } catch (err) { - await send(`Sync complete but failed to read results: ${err instanceof Error ? err.message : String(err)}`); - } + const text = await runEmailSyncCommand(userId); + for (const chunk of chunkMessage(text)) await send(chunk); } async function handleCommand(ctx: CommandContext): Promise { diff --git a/src/servers/channels/whatsapp/handler.ts b/src/servers/channels/whatsapp/handler.ts index 57e1dd09..1ff1d773 100644 --- a/src/servers/channels/whatsapp/handler.ts +++ b/src/servers/channels/whatsapp/handler.ts @@ -6,7 +6,6 @@ import { getWhatsAppClient } from './bot'; import { listChatModels } from '@@/api/chat/list-models'; import { enqueueJob } from '../../queue/init'; import { readJob } from '@@/queue/storage'; -import { openUserEmailDb } from '@@/sidecar/email/store'; import type { ModelInfo } from '@@/api/chat/types'; import { toShellUsername } from '@@/data-path'; @@ -41,6 +40,8 @@ async function getVisibleModels(): Promise { }); } +import { runEmailSyncCommand } from '../email-sync-command'; + type CommandContext = { content: string; send: SendFn; @@ -50,74 +51,10 @@ type CommandContext = { }; async function handleEmailSync(ctx: CommandContext): Promise { - const { send, email, userId } = ctx; - - let countBefore = 0; - try { - const db = await openUserEmailDb(email, userId); - if (db) { - const row = db.query('SELECT COUNT(*) as count FROM emails WHERE deleted = 0').get() as { count: number }; - countBefore = row.count; - db.close(); - } - } catch { - // DB might not exist yet - } - + const { send, userId } = ctx; await send('Syncing emails...'); - - const job = await enqueueJob({ lane: 'google-api', type: 'gmail-sync', userId: email }); - - const poll = async (): Promise<'completed' | 'failed' | 'cancelled'> => { - for (let i = 0; i < 120; i++) { - await new Promise((r) => setTimeout(r, 3000)); - const current = await readJob(job.id); - if (!current) return 'failed'; - if (current.status === 'completed' || current.status === 'failed' || current.status === 'cancelled') { - return current.status; - } - } - return 'failed'; - }; - - const status = await poll(); - - if (status !== 'completed') { - await send('Email sync failed. Check the queue dashboard for details.'); - return; - } - - try { - const db = await openUserEmailDb(email, userId); - if (!db) return; - const row = db.query('SELECT COUNT(*) as count FROM emails WHERE deleted = 0').get() as { count: number }; - const countAfter = row.count; - const newCount = countAfter - countBefore; - - if (newCount <= 0) { - db.close(); - await send('Sync complete — no new emails.'); - return; - } - - const newest = db - .query('SELECT from_name, from_address, subject FROM emails WHERE deleted = 0 ORDER BY date DESC LIMIT ?') - .all(Math.min(newCount, 20)) as Array<{ from_name: string | null; from_address: string; subject: string }>; - db.close(); - - const lines = newest.map((e) => { - const sender = e.from_name || e.from_address; - return `- *${sender}*: ${e.subject}`; - }); - - let text = `Sync complete — *${newCount}* new email${newCount !== 1 ? 's' : ''}`; - if (newCount > 20) text += ` (showing latest 20)`; - text += `:\n\n${lines.join('\n')}`; - - await send(text); - } catch (err) { - await send(`Sync complete but failed to read results: ${err instanceof Error ? err.message : String(err)}`); - } + const text = await runEmailSyncCommand(userId); + await send(text); // WhatsApp's 65k limit means no chunking is needed } async function handleCommand(ctx: CommandContext): Promise { diff --git a/src/servers/queue/handlers/index.ts b/src/servers/queue/handlers/index.ts index fae81d29..c2c30cbd 100644 --- a/src/servers/queue/handlers/index.ts +++ b/src/servers/queue/handlers/index.ts @@ -1,2 +1,8 @@ -import './gmail-sync'; -import './email-sync'; +// Empty on purpose. +// +// Both handlers that lived here were email syncs, and they moved into the officer-email sidecar, which +// schedules its own work now (sidecar/email/sync-runner.ts). The queue itself is untouched and still +// serves the Jobs screen; it simply no longer has anything to do with mail. +// +// New handlers register here by side-effect import, as before. +export {}; diff --git a/src/servers/sidecar/email/accounts.ts b/src/servers/sidecar/email/accounts.ts index 47696f1b..8d77689c 100644 --- a/src/servers/sidecar/email/accounts.ts +++ b/src/servers/sidecar/email/accounts.ts @@ -9,7 +9,9 @@ import { } from 'officerdb'; import { getValidGoogleAccessToken } from '../../api/integrations/google-auth'; import { validateImapConnection } from './imap-validate'; -import { enqueueJob, listAllJobs } from '../../queue/init'; +import { startSync, isSyncing, getSyncStates } from './sync-runner'; +import { imapSyncSteps } from './imap-sync'; +import { gmailSyncSteps } from './gmail-api'; import { openEmailDb, getSyncMeta } from './store'; import { performResync } from './resync'; @@ -45,17 +47,9 @@ accountsRouter.get('/', async (ctx) => { let activeJobAccountIds = new Set(); 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 | undefined)?.emailAccountId as number) - .filter(Boolean), - ); - } catch { - // Sidecar unavailable — all syncing/queued accounts are stale - } + // 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)) { @@ -177,12 +171,16 @@ accountsRouter.post('/:id/sync', async (ctx) => { 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 useGmailApi = account.provider === 'gmail' && account.authType === 'oauth'; - const job = await enqueueJob({ - lane: 'email', - type: jobType, - userId: user.email, + 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, @@ -201,7 +199,7 @@ accountsRouter.post('/:id/sync', async (ctx) => { }, }); - return ctx.json({ ok: true, jobId: job.id }, 201); + return ctx.json({ ok: true }, 201); }); accountsRouter.post('/validate', async (ctx) => { diff --git a/src/servers/queue/handlers/gmail-sync.ts b/src/servers/sidecar/email/gmail-api.ts similarity index 97% rename from src/servers/queue/handlers/gmail-sync.ts rename to src/servers/sidecar/email/gmail-api.ts index 028cc582..b8f6d55f 100644 --- a/src/servers/queue/handlers/gmail-sync.ts +++ b/src/servers/sidecar/email/gmail-api.ts @@ -1,8 +1,6 @@ import type { Database } from 'bun:sqlite'; import { createHash } from 'node:crypto'; -import { type JobHandler, PermanentError } from '../types'; -import { registerHandler } from '../handler-registry'; -import { openEmailDb, upsertFromRawEml, setSyncMeta, getSyncMeta } from '../../sidecar/email/store'; +import { openEmailDb, upsertFromRawEml, setSyncMeta, getSyncMeta } from './store'; import { getUserByEmail, getUserIntegration, @@ -15,6 +13,12 @@ import { } from 'officerdb'; 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 = { @@ -547,13 +551,13 @@ async function imapFullSync({ creds, db, onProgress }: ImapSyncParams): Promise< // ── Handler ── -const gmailSyncHandler: JobHandler = { +const gmailSyncHandler = { type: 'gmail-sync', retry: { delayMs: 15 * 60 * 1000, maxRetries: 10 }, steps: [ { name: 'Verify credentials', - run: async (ctx) => { + run: async (ctx: SyncCtx) => { const creds = await loadGmailCredentials(ctx.job.userId); const emailAccountId = (ctx.meta as Record).emailAccountId as number | undefined; const syncAccount = emailAccountId ? await getEmailAccount(emailAccountId) : undefined; @@ -620,7 +624,7 @@ const gmailSyncHandler: JobHandler = { }, { name: 'Sync emails', - run: async (ctx) => { + run: async (ctx: SyncCtx) => { const creds = ctx.meta.creds as GmailCredentials; const emailAccountId = (ctx.meta as Record).emailAccountId as number | undefined; @@ -680,7 +684,7 @@ const gmailSyncHandler: JobHandler = { }, { name: 'Finalize', - run: async (ctx) => { + run: async (ctx: SyncCtx) => { const result = ctx.meta.syncResult as SyncResult | undefined; const emailAccountId = ctx.meta.emailAccountId as number | undefined; @@ -709,4 +713,5 @@ const gmailSyncHandler: JobHandler = { ], }; -registerHandler(gmailSyncHandler); +/** The handler's steps, for the sidecar's own runner. */ +export const gmailSyncSteps: SyncStep[] = gmailSyncHandler.steps as SyncStep[]; diff --git a/src/servers/queue/handlers/email-sync.ts b/src/servers/sidecar/email/imap-sync.ts similarity index 96% rename from src/servers/queue/handlers/email-sync.ts rename to src/servers/sidecar/email/imap-sync.ts index cdbef18b..e9bf56ad 100644 --- a/src/servers/queue/handlers/email-sync.ts +++ b/src/servers/sidecar/email/imap-sync.ts @@ -1,8 +1,5 @@ import { createHash } from 'node:crypto'; -import type { JobHandler } from '../types'; -import { PermanentError } from '../types'; -import { registerHandler } from '../handler-registry'; -import { openEmailDb, upsertFromRawEml } from '../../sidecar/email/store'; +import { openEmailDb, upsertFromRawEml } from './store'; import { refreshGoogleAccessToken } from '../../api/integrations/google-auth'; import { getEmailAccount, @@ -15,6 +12,12 @@ import { setDockPaths, } from 'officerdb'; +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 = { @@ -108,13 +111,13 @@ async function resolveImapAuth(meta: EmailSyncMeta): Promise<{ user: string; pas // ── Handler ── -const emailSyncHandler: JobHandler = { +const emailSyncHandler = { type: 'email-sync', retry: { delayMs: 15 * 60 * 1000, maxRetries: 5 }, steps: [ { name: 'Sync emails', - run: async (ctx) => { + run: async (ctx: SyncCtx) => { const { ImapFlow } = await import('imapflow'); const meta = ctx.meta as unknown as EmailSyncMeta; const { account, userEmail } = meta; @@ -352,7 +355,7 @@ const emailSyncHandler: JobHandler = { }, { name: 'Finalize', - run: async (ctx) => { + run: async (ctx: SyncCtx) => { const meta = ctx.meta as unknown as EmailSyncMeta; const saved = meta.saved ?? 0; @@ -378,4 +381,5 @@ const emailSyncHandler: JobHandler = { ], }; -registerHandler(emailSyncHandler); +/** The handler's steps, for the sidecar's own runner. */ +export const imapSyncSteps: SyncStep[] = emailSyncHandler.steps as SyncStep[]; diff --git a/src/servers/sidecar/email/index.ts b/src/servers/sidecar/email/index.ts index f83a9b82..d3b529e4 100644 --- a/src/servers/sidecar/email/index.ts +++ b/src/servers/sidecar/email/index.ts @@ -1,5 +1,4 @@ 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'; @@ -8,39 +7,9 @@ import { createSidecarConnector } from '../connect'; const API_URL = process.env.API_URL ?? `ws://127.0.0.1:${process.env.PORT ?? '5000'}`; -// ── Queue access via WS ── - -let reqCounter = 0; -const pendingQueue = new Map void; reject: (e: Error) => void; timer: Timer }>(); - -function nextQueueId(): string { - return `eq_${Date.now()}_${++reqCounter}`; -} - -function sendQueueCommand(cmd: Record): Promise { - return new Promise((resolve, reject) => { - const id = cmd.id as string; - const timer = setTimeout(() => { - pendingQueue.delete(id); - reject(new Error(`Queue command ${cmd.type} timed out`)); - }, 30_000); - pendingQueue.set(id, { resolve, reject, timer }); - connection.send(cmd as SidecarEvent); - }); -} - -async function enqueueViaWs(params: EnqueueParams): Promise { - const res = (await sendQueueCommand({ type: 'queue:enqueue', id: nextQueueId(), params })) as Record; - if (res.type === 'queue:enqueued') return res.job as Job; - if (res.type === 'queue:error') throw new Error(res.error as string); - throw new Error('Unexpected response'); -} - -async function listJobsViaWs(): Promise { - const res = (await sendQueueCommand({ type: 'queue:list', id: nextQueueId() })) as Record; - if (res.type === 'queue:list') return res.jobs as Job[]; - throw new Error('Unexpected response'); -} +// 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 ── @@ -53,20 +22,6 @@ function handleCommand(cmd: Record, reply: ReplyFn) { break; default: - // Check if this is a queue response (from API server responding to our queue commands) - if ( - typeof cmd.type === 'string' && - cmd.type.startsWith('queue:') && - cmd.id && - pendingQueue.has(cmd.id as string) - ) { - const pending = pendingQueue.get(cmd.id as string)!; - pendingQueue.delete(cmd.id as string); - clearTimeout(pending.timer); - pending.resolve(cmd); - return; - } - reply({ type: 'error', id: cmd.id as string, diff --git a/src/servers/sidecar/email/resync.ts b/src/servers/sidecar/email/resync.ts index 45eb7a2a..8673afff 100644 --- a/src/servers/sidecar/email/resync.ts +++ b/src/servers/sidecar/email/resync.ts @@ -12,7 +12,7 @@ import { type GmailCredentials, loadGmailCredentials, gmailApiSync, -} from '../../queue/handlers/gmail-sync'; +} from './gmail-api'; export type ResyncResult = { saved: number; skipped: number; errors: number }; diff --git a/src/servers/sidecar/email/routes.ts b/src/servers/sidecar/email/routes.ts index 667fb077..8ad411de 100644 --- a/src/servers/sidecar/email/routes.ts +++ b/src/servers/sidecar/email/routes.ts @@ -8,6 +8,7 @@ import { getEmailAttachmentCacheDir } from '@@/data-path'; import { getEmailAccounts } from 'officerdb'; import { openEmailDb, openUserEmailDb, rowToSummary, getSyncMeta, searchEmails } from './store'; import { accountsRouter } from './accounts'; +import { performResync } from './resync'; export const emailRouter = createRouter(); @@ -364,6 +365,34 @@ emailRouter.delete('/messages/:id', async (ctx) => { } }); +// POST /sync-now — run a resync and report what arrived. +// +// For the chat channels ("sync my email" from Telegram/Discord/WhatsApp). They used to open the mail +// store directly and enqueue a `gmail-sync` job, which stopped existing when sync moved in here; and the +// job type was hardcoded to the OAuth path even though an app-password account syncs over IMAP, so that +// command had been failing regardless. One call now: sync, then say what is new. +emailRouter.post('/sync-now', async (ctx) => { + const user = ctx.get('user'); + const accounts = await getEmailAccounts(user.id); + const account = accounts.find((a) => a.enabled) ?? accounts[0]; + if (!account) return ctx.json({ saved: 0, newest: [], error: 'No email account configured' }); + + const result = await performResync({ accountId: account.id, userEmail: user.email, userId: user.id }); + + const db = openEmailDb(user.email, account.email); + try { + const newest = + result.saved > 0 + ? (db + .query('SELECT from_name, from_address, subject FROM emails WHERE deleted = 0 ORDER BY date DESC LIMIT ?') + .all(Math.min(result.saved, 20)) as Array<{ from_name: string | null; from_address: string; subject: string }>) + : []; + return ctx.json({ saved: result.saved, skipped: result.skipped, errors: result.errors, newest }); + } finally { + db.close(); + } +}); + emailRouter.get('/sync-status', async (ctx) => { const user = ctx.get('user'); diff --git a/src/servers/sidecar/email/sync-runner.ts b/src/servers/sidecar/email/sync-runner.ts new file mode 100644 index 00000000..a8a23646 --- /dev/null +++ b/src/servers/sidecar/email/sync-runner.ts @@ -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(); + +/** 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 }; + +/** + * 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//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; + job: { userId: string }; + updateProgress: (p: SyncProgress) => Promise | 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; + steps: SyncStep[]; + onDone?: (ok: boolean) => Promise | 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; +}