Merge remote-tracking branch 'origin/email-imap'

src/servers/sidecar/email-cron.ts

@@ -0,0 +1,87 @@
+import { getAllSyncedAccounts, getUserById, getUserIntegration } from 'officerdb';
+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') {
+        const integration = await getUserIntegration(account.userId, 'google');
+        const config = integration?.config as Record<string, unknown> | undefined;
+        const accessToken = config?.accessToken as string | undefined;
+        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;
+  }
+}

src/servers/sidecar/index.ts

@@ -5,6 +5,7 @@ import { startAnthropicProxy, getProxySecret, ensureProxySecret } from './proxy'
 import * as claudeManager from './claude-manager';
 import * as piManager from './pi-manager';
 import * as queueRunner from './queue-runner';
+import { initEmailCron, stopEmailCron } from './email-cron';
 
 const PORT = Number(process.env.SIDECAR_PORT ?? '5100');
 const startedAt = Date.now();
@@ -31,6 +32,9 @@ queueRunner.initQueue().catch((err) => {
   console.error('[sidecar] failed to initialize queue:', err);
 });
 
+// Start email sync cron
+// initEmailCron(); // TODO: re-enable after initial sync testing
+
 // ── WebSocket connections ──
 
 const clients = new Set<ServerWebSocket<unknown>>();
@@ -251,6 +255,7 @@ console.log(`[sidecar] listening on 127.0.0.1:${PORT}`);
 
 async function shutdown(signal: string) {
   console.log(`[sidecar] ${signal} received, saving state...`);
+  stopEmailCron();
   await flushAndSave();
   releaseLock();
   process.exit(0);

src/servers/sidecar/queue-runner.ts

@@ -5,6 +5,17 @@ import { getHandler } from '../queue/handler-registry';
 // Import handlers to register them
 import '../queue/handlers';
 
+function formatDuration(ms: number): string {
+  const s = Math.floor(ms / 1000);
+  if (s < 60) return `${s}s`;
+  const m = Math.floor(s / 60);
+  const rem = s % 60;
+  if (m < 60) return rem > 0 ? `${m}m${rem}s` : `${m}m`;
+  const h = Math.floor(m / 60);
+  const remM = m % 60;
+  return remM > 0 ? `${h}h${remM}m` : `${h}h`;
+}
+
 const activeLanes = new Map<string, boolean>();
 const PROGRESS_THROTTLE_MS = 1000;
 
@@ -73,6 +84,12 @@ async function resumeInterruptedJobs() {
       console.log(`[sidecar:queue] reset interrupted job ${job.id} back to queued`);
       lanesToKick.add(job.lane);
     } else if (job.status === 'queued') {
+      // Clear retry delay on restart — no reason to wait after a sidecar restart
+      if (job.retryAt) {
+        job.retryAt = undefined;
+        await writeJob(job);
+        console.log(`[sidecar:queue] cleared retry delay for job ${job.id}`);
+      }
       lanesToKick.add(job.lane);
     }
   }
@@ -136,8 +153,9 @@ async function runJob(job: Job) {
   job.retryAt = undefined;
   await writeJob(job);
   const isRetry = (job.retries ?? 0) > 0;
+  const startTime = Date.now();
   console.log(
-    `[sidecar:queue] ${isRetry ? 'resuming' : 'running'} job ${job.id} (${job.type})${isRetry ? ` retry ${job.retries}` : ''}`,
+    `[sidecar:queue] ▶ ${isRetry ? 'resuming' : 'running'} job ${job.id} (${job.type})${isRetry ? ` retry ${job.retries}` : ''}`,
   );
 
   const sharedMeta: Record<string, unknown> = { ...(job.meta ?? {}) };
@@ -187,6 +205,7 @@ async function runJob(job: Job) {
       await writeJob(fresh);
     } catch (err) {
       const errorMessage = err instanceof Error ? err.message : String(err);
+      console.error(`[sidecar:queue] step "${step.name}" failed: ${errorMessage}`);
       step.status = 'failed';
       step.error = errorMessage;
       step.completedAt = Date.now();
@@ -218,7 +237,7 @@ async function runJob(job: Job) {
       fresh.completedAt = Date.now();
       fresh.meta = { ...fresh.meta, ...sharedMeta };
       await writeJob(fresh);
-      console.error(`[sidecar:queue] job ${fresh.id} failed at step "${step.name}":`, errorMessage);
+      console.error(`[sidecar:queue] ✗ job ${fresh.id} failed at step "${step.name}" in ${formatDuration(Date.now() - startTime)}:`, errorMessage);
       await notifyFailure(fresh);
       return;
     }
@@ -230,7 +249,7 @@ async function runJob(job: Job) {
     final.completedAt = Date.now();
     final.meta = { ...final.meta, ...sharedMeta };
     await writeJob(final);
-    console.log(`[sidecar:queue] job ${final.id} completed`);
+    console.log(`[sidecar:queue] ✓ job ${final.id} completed in ${formatDuration(Date.now() - startTime)}`);
     await notifyCompletion(final);
   }
 }