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; +}