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

improve gmail sync: email input, permanent errors, auto-dock

Browse Source 
- add isync to setup.sh
- ask for gmail address alongside app password in integrations
- add PermanentError to job queue (skips retries for non-recoverable failures)
- use PermanentError for missing credentials, missing executable, auth failures
- auto-add /email to dock after successful gmail sync
- invalidate dock cache on sync completion for seamless UI update

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
pastilhas
2026-03-04 23:13:55 +00:00
co-authored by Claude Opus 4.6
parent 8737681464
commit aa0207436c
7 changed files with 146 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/servers/api/integrations/integrations.ts
+6 -3
View File
@@ -110,7 +110,7 @@ integrationsRouter.get('/google/verify', async (ctx) => {
const body = await res.json();
const valid = body.error === 'invalid_grant' || body.error === 'redirect_uri_mismatch';
return ctx.json({ valid, error: valid ? null : body.error_description ?? body.error });
return ctx.json({ valid, error: valid ? null : (body.error_description ?? body.error) });
});
// --- Personal: Google account connection status ---
@@ -132,17 +132,20 @@ integrationsRouter.get('/google/status', async (ctx) => {
integrationsRouter.put('/google/app-password', async (ctx) => {
const user = ctx.get('user');
const body = ctx.get('body') as { appPassword?: string };
const body = ctx.get('body') as { appPassword?: string; email?: string };
if (!body.appPassword) throw BAD_REQUEST('Missing appPassword');
const connection = await getUserIntegration(user.id, 'google');
const connConfig = (connection?.config as Record<string, unknown>) ?? {};
const updated: Record<string, unknown> = { ...connConfig, imapAppPassword: body.appPassword };
if (body.email) updated.email = body.email;
await upsertUserIntegration({
userId: user.id,
provider: 'google',
serverIntegrationId: connection?.serverIntegrationId ?? undefined,
config: { ...connConfig, imapAppPassword: body.appPassword },
config: updated,
});
return ctx.json({ ok: true });
src/servers/queue/engine.ts
+13 -6
View File
@@ -1,4 +1,4 @@
import type { Job, JobProgress, EnqueueParams, StepContext } from './types';
import { type Job, type JobProgress, type EnqueueParams, type StepContext, PermanentError } from './types';
import { readJob, writeJob, listAllJobs } from './storage';
import { getHandler } from './handler-registry';
import { sendMail } from 'emailer';
@@ -103,7 +103,9 @@ async function processNextInLane(lane: string) {
console.error(`[queue] Lane ${lane} processing error:`, err);
} finally {
const jobs = await listAllJobs();
const hasMore = jobs.some((j) => j.lane === lane && j.status === 'queued' && (!j.retryAt || j.retryAt <= Date.now()));
const hasMore = jobs.some(
(j) => j.lane === lane && j.status === 'queued' && (!j.retryAt || j.retryAt <= Date.now()),
);
if (hasMore) {
processNextInLane(lane);
} else {
@@ -127,7 +129,9 @@ async function runJob(job: Job) {
job.retryAt = undefined;
await writeJob(job);
const isRetry = (job.retries ?? 0) > 0;
console.log(`[queue] ${isRetry ? 'Resuming' : 'Running'} job ${job.id} (${job.type})${isRetry ? ` retry ${job.retries}` : ''}`);
console.log(
`[queue] ${isRetry ? 'Resuming' : 'Running'} job ${job.id} (${job.type})${isRetry ? ` retry ${job.retries}` : ''}`,
);
const sharedMeta: Record<string, unknown> = { ...(job.meta ?? {}) };
@@ -182,9 +186,10 @@ async function runJob(job: Job) {
step.error = errorMessage;
step.completedAt = Date.now();
// Check if handler supports retry
// Check if handler supports retry (PermanentError skips retries)
const isPermanent = err instanceof PermanentError;
const retries = (fresh.retries ?? 0) + 1;
if (handler.retry && retries <= handler.retry.maxRetries) {
if (!isPermanent && handler.retry && retries <= handler.retry.maxRetries) {
// Schedule retry: reset failed step to pending, re-queue
step.status = 'pending';
step.error = undefined;
@@ -198,7 +203,9 @@ async function runJob(job: Job) {
fresh.retries = retries;
fresh.retryAt = Date.now() + handler.retry.delayMs;
await writeJob(fresh);
console.log(`[queue] Job ${fresh.id} will retry (${retries}/${handler.retry.maxRetries}) in ${handler.retry.delayMs / 1000}s — failed at step "${step.name}": ${errorMessage}`);
console.log(
`[queue] Job ${fresh.id} will retry (${retries}/${handler.retry.maxRetries}) in ${handler.retry.delayMs / 1000}s — failed at step "${step.name}": ${errorMessage}`,
);
scheduleRetry(fresh.lane, handler.retry.delayMs);
return;
}
src/servers/queue/handlers/gmail-sync.ts
+38 -10
View File
@@ -2,10 +2,10 @@ import type { Database } from 'bun:sqlite';
import { createHash } from 'node:crypto';
import { join } from 'node:path';
import { readdir, readFile, writeFile, unlink, mkdir, stat } from 'node:fs/promises';
import type { JobHandler } from '../types';
import { type JobHandler, PermanentError } from '../types';
import { registerHandler } from '../handler-registry';
import { openEmailDb, upsertFromRawEml, setSyncMeta } from '../../api/email/email-db';
import { getUserByEmail, getUserIntegration } from 'officerdb';
import { getUserByEmail, getUserIntegration, getDockPaths, setDockPaths } from 'officerdb';
import { getMaildirPath } from '@@/data-path';
// ── Credentials ──
@@ -14,12 +14,12 @@ type ImapCredentials = { email: string; appPassword: string };
async function loadImapCredentials(userEmail: string): Promise<ImapCredentials> {
const dbUser = await getUserByEmail(userEmail);
if (!dbUser) throw new Error('User not found');
if (!dbUser) throw new PermanentError('User not found');
const userGoogle = await getUserIntegration(dbUser.id, 'google');
const config = userGoogle?.config as Record<string, unknown> | undefined;
if (!config?.imapAppPassword) {
throw new Error('Gmail App Password not configured — set it in Settings → Integrations');
throw new PermanentError('Gmail App Password not configured — set it in Settings → Integrations');
}
const gmailEmail = (config.email as string) ?? userEmail;
@@ -235,15 +235,21 @@ const gmailSyncHandler: JobHandler = {
await ctx.updateProgress({ current: 0, total: 0, label: 'Running mbsync...' });
try {
const proc = Bun.spawn(['mbsync', '-c', configPath, '-a', '-V'], {
stdout: 'pipe',
stderr: 'pipe',
});
let proc: ReturnType<typeof Bun.spawn>;
try {
proc = Bun.spawn(['mbsync', '-c', configPath, '-a', '-V'], {
stdout: 'pipe',
stderr: 'pipe',
});
} catch (spawnErr) {
const msg = spawnErr instanceof Error ? spawnErr.message : String(spawnErr);
throw new PermanentError(`Failed to start mbsync: ${msg}`);
}
// Stream stderr for live progress
let lastLine = '';
let stderrBuf = '';
const reader = proc.stderr.getReader();
const reader = (proc.stderr as ReadableStream<Uint8Array>).getReader();
const decoder = new TextDecoder();
const readLoop = (async () => {
while (true) {
@@ -285,6 +291,7 @@ const gmailSyncHandler: JobHandler = {
ctx.meta.gmailSyncRecoverable = true;
ctx.meta.gmailSyncEmailCount = emailCount;
ctx.meta.gmailSyncIsAuthFail = true;
throw new PermanentError(`Authentication failed — check your App Password`);
}
throw new Error(`mbsync exited with code ${exitCode}: ${stderrBuf.slice(-500)}`);
}
@@ -315,11 +322,32 @@ const gmailSyncHandler: JobHandler = {
});
});
console.log(`[gmail-sync] Import done: saved ${result.saved}, skipped ${result.skipped}, errors ${result.errors}`);
console.log(
`[gmail-sync] Import done: saved ${result.saved}, skipped ${result.skipped}, errors ${result.errors}`,
);
setSyncMeta(db, 'last_sync_date', new Date().toISOString().split('T')[0]!);
setSyncMeta(db, 'last_sync_at', new Date().toISOString());
// Auto-add /email to dock if not already present
if (result.saved > 0) {
try {
const dbUser = await getUserByEmail(ctx.job.userId);
if (dbUser) {
const paths = await getDockPaths(dbUser.id);
if (!paths) {
// User hasn't customized dock — initialize with defaults + /email
const defaults = ['/', '/files', '/automation', '/projects', '/dashboards', '/chat'];
await setDockPaths(dbUser.id, [...defaults, '/email']);
} else if (!paths.includes('/email')) {
await setDockPaths(dbUser.id, [...paths, '/email']);
}
}
} catch {
// Non-fatal — dock update is best-effort
}
}
await ctx.updateProgress({ current: 1, total: 1, label: `Done — ${result.saved} new emails` });
} finally {
db.close();
src/servers/queue/index.ts
+11 -1
View File
@@ -5,7 +5,17 @@ import './handlers';
export { enqueue, cancelJob } from './engine';
export { readJob, listAllJobs } from './storage';
export { registerHandler } from './handler-registry';
export type { Job, JobStep, JobStatus, JobProgress, StepContext, JobHandler, JobHandlerStep, EnqueueParams } from './types';
export { PermanentError } from './types';
export type {
Job,
JobStep,
JobStatus,
JobProgress,
StepContext,
JobHandler,
JobHandlerStep,
EnqueueParams,
} from './types';
export async function initQueue() {
await ensureQueueDir();
src/servers/queue/types.ts
+8
View File
@@ -65,3 +65,11 @@ export type EnqueueParams = {
meta?: Record<string, unknown>;
notify?: boolean;
};
/** Throw this from a step to fail the job immediately without retrying. */
export class PermanentError extends Error {
constructor(message: string) {
super(message);
this.name = 'PermanentError';
}
}