diff --git a/scripts/reimport-gmail.ts b/scripts/reimport-gmail.ts index b10a3539..ade78d0f 100644 --- a/scripts/reimport-gmail.ts +++ b/scripts/reimport-gmail.ts @@ -25,9 +25,9 @@ try { const db = openEmailDb(email); console.log(`Importing from ${maildirPath}...`); -const result = await importMaildir(maildirPath, email, db, (saved, skipped) => { +const result = await importMaildir({ maildirPath, emailAccount: email, db, onProgress: (saved, skipped) => { process.stdout.write(`\r saved ${saved}, skipped ${skipped}`); -}); +}}); console.log(`\nDone: saved ${result.saved}, skipped ${result.skipped}, errors ${result.errors}`); db.close(); diff --git a/scripts/seed-imap-uids.ts b/scripts/seed-imap-uids.ts new file mode 100644 index 00000000..cef52ccb --- /dev/null +++ b/scripts/seed-imap-uids.ts @@ -0,0 +1,88 @@ +import { ImapFlow } from 'imapflow'; +import { getUserByEmail, getUserIntegration, getServerIntegration } from 'officerdb'; +import { openEmailDb, setSyncMeta } from '../src/servers/api/email/email-db'; + +const userEmail = process.argv[2]; +if (!userEmail) { + console.error('Usage: bun run scripts/seed-imap-uids.ts '); + process.exit(1); +} + +// ── Load credentials ── + +const dbUser = await getUserByEmail(userEmail); +if (!dbUser) throw new Error('User not found'); + +const userGoogle = await getUserIntegration(dbUser.id, 'google'); +const config = userGoogle?.config as Record | undefined; +if (!config?.accessToken) throw new Error('No OAuth tokens found'); + +// Refresh token if needed +let accessToken = config.accessToken as string; +const expiresAt = config.expiresAt as number | undefined; +if (!expiresAt || expiresAt < Date.now() + 60_000) { + console.log('Refreshing expired token...'); + const serverGoogle = await getServerIntegration('google'); + const serverConfig = serverGoogle?.config as Record; + const res = await fetch('https://oauth2.googleapis.com/token', { + method: 'POST', + headers: { 'Content-Type': 'application/x-www-form-urlencoded' }, + body: new URLSearchParams({ + client_id: serverConfig.clientId as string, + client_secret: serverConfig.clientSecret as string, + refresh_token: config.refreshToken as string, + grant_type: 'refresh_token', + }), + }); + if (!res.ok) throw new Error(`Token refresh failed: ${await res.text()}`); + const data = (await res.json()) as { access_token: string }; + accessToken = data.access_token; +} + +// ── Connect IMAP ── + +const client = new ImapFlow({ + host: 'imap.gmail.com', + port: 993, + secure: true, + auth: { user: config.email as string, accessToken }, + logger: false, +}); + +await client.connect(); +console.log('Connected to IMAP'); + +const GMAIL_PREFIX_RE = /^\[(?:Gmail|Google Mail)\]\//; +const SKIP_SUFFIXES = new Set(['All Mail', 'Trash', 'Spam', 'Bin']); + +const folders = await client.list(); +const db = openEmailDb(userEmail); + +let seeded = 0; + +for (const folder of folders) { + const suffix = folder.path.replace(GMAIL_PREFIX_RE, ''); + const isGmailFolder = suffix !== folder.path; + if (isGmailFolder && SKIP_SUFFIXES.has(suffix)) continue; + if (folder.specialUse && ['\\Trash', '\\Junk', '\\All'].includes(folder.specialUse)) continue; + + try { + const status = await client.status(folder.path, { uidNext: true, uidValidity: true }); + const lastUid = (status.uidNext ?? 1) - 1; + const uidValidity = String(status.uidValidity); + + setSyncMeta(db, `imap_lastuid:${folder.path}`, String(lastUid)); + setSyncMeta(db, `imap_uidvalidity:${folder.path}`, uidValidity); + + console.log(` ${folder.path}: lastUid=${lastUid}, uidValidity=${uidValidity}`); + seeded++; + } catch (err) { + console.log(` ${folder.path}: skipped (${err instanceof Error ? err.message : err})`); + } +} + +db.close(); +await client.logout(); + +console.log(`\nSeeded ${seeded} folders. Next sync will only fetch new messages.`); +process.exit(0); diff --git a/src/apps/officer-web/Screens/Dashboard/Email/EmailList.tsx b/src/apps/officer-web/Screens/Dashboard/Email/EmailList.tsx index e72bef88..99e7fa47 100644 --- a/src/apps/officer-web/Screens/Dashboard/Email/EmailList.tsx +++ b/src/apps/officer-web/Screens/Dashboard/Email/EmailList.tsx @@ -6,14 +6,13 @@ import { toast } from 'sonner'; import { Button } from '@/components/ui/button'; import { useClient } from 'hooks/useClient'; import { useGlobal } from 'hooks/useGlobal'; -import { useJobs } from 'hooks/useJobs'; import type { EmailSummary } from 'types'; -type GoogleStatus = { - configured: boolean; - connected: boolean; - email: string | null; - picture: string | null; +type EmailAccountRow = { + id: number; + provider: string; + email: string; + status: string; }; const LIMIT = 50; @@ -42,13 +41,13 @@ export const EmailList = () => { const [selectedId, setSelectedId] = useGlobal('EMAIL_SELECTED', null); const [folder, setFolder] = useGlobal('EMAIL_FOLDER', 'inbox'); const [page, setPage] = useState(1); - const { jobs, createJob } = useJobs({ type: 'gmail-sync' }); - const isSyncing = jobs.some((j) => j.status === 'queued' || j.status === 'running'); - - const { data: googleStatus } = useQuery({ - queryKey: ['google-status'], - queryFn: () => client.get('/integrations/google/status'), + const { data: emailAccounts = [], refetch: refetchAccounts } = useQuery({ + queryKey: ['email-accounts'], + queryFn: () => client.get('/email/accounts'), }); + const syncableAccount = emailAccounts.find((a) => a.status === 'connected' || a.status === 'synced'); + const isSyncing = emailAccounts.some((a) => a.status === 'syncing' || a.status === 'queued'); + const hasAccounts = emailAccounts.length > 0; const { data, isLoading } = useQuery({ queryKey: ['email-messages', page, folder], @@ -75,17 +74,23 @@ export const EmailList = () => { }; const handleSync = async () => { + if (!syncableAccount) return; try { - await createJob({ lane: 'google-api', type: 'gmail-sync', notify: false }); - toast.success('Gmail sync started'); - } catch { - toast.error('Failed to start sync'); + await client.post(`/email/accounts/${syncableAccount.id}/sync`, {}); + toast.success('Sync started'); + refetchAccounts(); + } catch (err) { + toast.error(err instanceof Error ? err.message : 'Failed to start sync'); } }; - // Refresh email list when a sync job completes + // Poll accounts while syncing, refresh email list when done const prevSyncing = useRef(false); useEffect(() => { + if (isSyncing) { + const interval = setInterval(refetchAccounts, 5000); + return () => clearInterval(interval); + } if (prevSyncing.current && !isSyncing) { queryClient.invalidateQueries({ queryKey: ['email-messages'] }); } @@ -134,16 +139,14 @@ export const EmailList = () => { return (
- {googleStatus && !googleStatus.configured ? ( - Google integration not configured. Contact your administrator. - ) : googleStatus && !googleStatus.connected ? ( + {!hasAccounts ? (
- Connect your Google account to sync emails + Add an email account to get started
- ) : ( + ) : syncableAccount ? (
No emails synced yet
+ ) : isSyncing ? ( +
+ + Syncing emails... +
+ ) : ( + No emails synced yet )}
); @@ -179,18 +189,17 @@ export const EmailList = () => { })} {total} - + + ) : null} {totalPages > 1 && (
+ )} + +
+ {/* Sync progress */} + {account.status === 'syncing' && progress && ( +
+ +

+ {progress.label ?? 'Syncing...'} +

+
+ )} + {account.status === 'syncing' && !progress && ( +
+ +

Syncing...

+
+ )} + + ); + })} + + )} + + {/* Provider picker */} + {showPicker && !activeForm && ( +
+ + + + +
+ )} + + {/* Gmail OAuth form */} + {activeForm === 'gmail-oauth' && ( +
+

Add Gmail (OAuth)

+ setForm({ ...form, displayName: ev.target.value })} + placeholder="Display name (optional)" + className="h-9 rounded-md border border-duck-dark/10 dark:border-foreground/10 bg-transparent px-3 text-sm" + /> +
+ + +
+
+ )} + + {/* Gmail App Password form */} + {activeForm === 'gmail-password' && ( +
+

Add Gmail (App Password)

+
+

To create an App Password:

+
    +
  1. + Go to{' '} + + Google App Passwords + +
    2. +
  2. You may need to enable 2-Step Verification first
    3. +
  3. Enter a name (e.g. "Officer") and click Create
    4. +
  4. Copy the 16-character password and paste it below
    5. +
+
+ setForm({ ...form, email: ev.target.value })} + placeholder="your@gmail.com" + className="h-9 rounded-md border border-duck-dark/10 dark:border-foreground/10 bg-transparent px-3 text-sm" + /> + setForm({ ...form, displayName: ev.target.value })} + placeholder="Display name (optional)" + className="h-9 rounded-md border border-duck-dark/10 dark:border-foreground/10 bg-transparent px-3 text-sm" + /> + setForm({ ...form, password: ev.target.value })} + placeholder="xxxx xxxx xxxx xxxx" + className="h-9 rounded-md border border-duck-dark/10 dark:border-foreground/10 bg-transparent px-3 text-sm" + /> +
+ + +
+
+ )} + + {/* Generic IMAP form */} + {activeForm === 'imap' && ( +
+

Add IMAP Account

+ setForm({ ...form, email: ev.target.value })} + placeholder="you@example.com" + className="h-9 rounded-md border border-duck-dark/10 dark:border-foreground/10 bg-transparent px-3 text-sm" + /> + setForm({ ...form, displayName: ev.target.value })} + placeholder="Display name (optional)" + className="h-9 rounded-md border border-duck-dark/10 dark:border-foreground/10 bg-transparent px-3 text-sm" + /> +
+ setForm({ ...form, imapHost: ev.target.value })} + placeholder="imap.example.com" + className="h-9 rounded-md border border-duck-dark/10 dark:border-foreground/10 bg-transparent px-3 text-sm" + /> + setForm({ ...form, imapPort: ev.target.value })} + placeholder="993" + className="h-9 w-20 rounded-md border border-duck-dark/10 dark:border-foreground/10 bg-transparent px-3 text-sm" + /> + +
+ setForm({ ...form, password: ev.target.value })} + placeholder="Password" + className="h-9 rounded-md border border-duck-dark/10 dark:border-foreground/10 bg-transparent px-3 text-sm" + /> +
+ + +
+
+ )} + + {/* Add account button */} + {!showPicker && !activeForm && ( + + )} + + ); +}; + +const StatusBadge = ({ status }: { status: string }) => { + switch (status) { + case 'connected': + return ( + + Connected + + ); + case 'queued': + return ( + + Queued + + ); + case 'syncing': + return ( + + Syncing + + ); + case 'synced': + return ( + + Synced + + ); + default: + return null; + } +}; + +const ProviderIcon = ({ provider }: { provider: string }) => { + switch (provider) { + case 'gmail': + return ; + case 'outlook': + return ; + default: + return ; + } +}; diff --git a/src/apps/officer-web/Screens/Dashboard/Settings/IntegrationsSettings/GoogleAccount.tsx b/src/apps/officer-web/Screens/Dashboard/Settings/IntegrationsSettings/GoogleAccount.tsx index 992b5a15..5dc8b5cf 100644 --- a/src/apps/officer-web/Screens/Dashboard/Settings/IntegrationsSettings/GoogleAccount.tsx +++ b/src/apps/officer-web/Screens/Dashboard/Settings/IntegrationsSettings/GoogleAccount.tsx @@ -1,64 +1,32 @@ import { useState, useEffect } from 'react'; import { toast } from 'sonner'; -import { RefreshCw, Loader2, CheckCircle2, XCircle } from 'lucide-react'; import { Button } from '@/components/ui/button'; import { useClient } from 'hooks/useClient'; -import { useAuth } from 'hooks/useAuth'; -import { useQueryClient } from '@tanstack/react-query'; -import { useJobs } from 'hooks/useJobs'; type GoogleStatus = { connected: boolean; email: string | null; picture: string | null; configured: boolean; - hasAppPassword: boolean; -}; - -const formatTime = (ts: number | string) => { - const d = new Date(ts); - return d.toLocaleString(undefined, { dateStyle: 'medium', timeStyle: 'short' }); }; export const GoogleAccount = () => { const client = useClient(); - const { user } = useAuth(); - const queryClient = useQueryClient(); const [isLoading, setIsLoading] = useState(true); const [status, setStatus] = useState({ connected: false, email: null, picture: null, configured: false, - hasAppPassword: false, }); - const [gmailEmail, setGmailEmail] = useState(''); - const [appPassword, setAppPassword] = useState(''); - const [showPasswordInput, setShowPasswordInput] = useState(false); - const [savingPassword, setSavingPassword] = useState(false); - const [lastSyncAt, setLastSyncAt] = useState(null); - const [dismissedError, setDismissedError] = useState(false); - const { jobs, createJob, refetch } = useJobs({ type: 'gmail-sync' }); - const activeJob = jobs.find((j) => j.status === 'queued' || j.status === 'running'); - const lastJob = jobs[0]; - - const fetchStatus = () => { - client - .get('/integrations/google/status') - .then((s) => { - setStatus(s); - setGmailEmail(s.email ?? user?.email ?? ''); - }) - .catch(() => {}) - .finally(() => setIsLoading(false)); - client - .get<{ lastSyncAt: string | null }>('/email/sync-status') - .then((res) => setLastSyncAt(res.lastSyncAt)) - .catch(() => {}); - }; useEffect(() => { - fetchStatus(); + client + .get('/integrations/google/status') + .then(setStatus) + .catch(() => {}) + .finally(() => setIsLoading(false)); + const params = new URLSearchParams(window.location.search); const result = params.get('google'); if (result === 'success') { @@ -71,27 +39,6 @@ export const GoogleAccount = () => { } }, []); - // Refresh sync status and dock when a job finishes - useEffect(() => { - if (!activeJob && lastJob?.status === 'completed') { - client - .get<{ lastSyncAt: string | null }>('/email/sync-status') - .then((res) => setLastSyncAt(res.lastSyncAt)) - .catch(() => {}); - queryClient.invalidateQueries({ queryKey: ['DOCK'] }); - } - }, [activeJob, lastJob?.status]); - - const handleSync = async (year?: number) => { - try { - await createJob({ lane: 'google-api', type: 'gmail-sync', meta: year ? { year } : undefined }); - await refetch(); - toast.success("Gmail sync started — you'll receive an email when it's done"); - } catch { - toast.error('Failed to start Gmail sync'); - } - }; - const handleConnect = () => { const params = new URLSearchParams({ token: client.token ?? '', @@ -110,201 +57,41 @@ export const GoogleAccount = () => { } }; - const handleSaveAppPassword = async () => { - if (!appPassword.trim() || !gmailEmail.trim()) return; - setSavingPassword(true); - try { - await client.put('/integrations/google/app-password', { - appPassword: appPassword.trim(), - email: gmailEmail.trim(), - }); - setStatus({ ...status, hasAppPassword: true, email: gmailEmail.trim() }); - setAppPassword(''); - setShowPasswordInput(false); - toast.success('App password saved'); - } catch { - toast.error('Failed to save app password'); - } finally { - setSavingPassword(false); - } - }; - if (isLoading) return null; - - const currentStep = activeJob?.steps[activeJob.currentStep]; - const progress = currentStep?.progress; + if (!status.configured) return null; return ( -
- {/* Gmail Sync — independent of OAuth */} -
-
-

Gmail Sync

-

- Import and sync your Gmail emails with your Officer inbox. This uses a Google App Password for a direct IMAP - connection — the only thing it can do is download your emails. It cannot send, delete, or modify anything in - your account. -

-
-
-

App Password

- {status.hasAppPassword && !showPasswordInput ? ( -
- Configured - {status.email && ( - ({status.email}) - )} - -
- ) : ( - <> -
-

To create an App Password:

-
    -
  1. - Go to{' '} - - Google App Passwords - -
    2. -
  2. You may need to enable 2-Step Verification first
    3. -
  3. Enter a name (e.g. "Officer") and click Create
    4. -
  4. Copy the 16-character password and paste it below
    5. -
-
-
- setGmailEmail(ev.target.value)} - placeholder="your@gmail.com" - className="h-9 rounded-md border border-duck-dark/10 dark:border-foreground/10 bg-transparent px-3 text-sm" - /> -
- setAppPassword(ev.target.value)} - placeholder="xxxx xxxx xxxx xxxx" - className="flex-1 h-9 rounded-md border border-duck-dark/10 dark:border-foreground/10 bg-transparent px-3 text-sm" - /> - -
-
- - )} -
- {activeJob && ( -
- +
+
+

Google Account

+

+ Connect your Google account for Calendar, Contacts, and other Google services. +

+
+ {status.connected ? ( + <> +
+
-

- {activeJob.status === 'queued' ? 'Queued' : 'Syncing'} - {progress?.label ? ` — ${progress.label}` : ''} -

- {progress && progress.total > 0 && ( -
-
-
- )} - {progress && progress.total === 0 && progress.current > 0 && ( -

- {progress.current.toLocaleString()} emails processed -

- )} +

Connected

+

{status.email}

+ {status.picture && ( + + )}
- )} - {!activeJob && lastJob?.status === 'failed' && !dismissedError && ( - - )} - {!activeJob && lastSyncAt && ( -
- - Last sync completed {formatTime(lastSyncAt)} -
- )} + + + ) : ( -
- - {/* OAuth — for Calendar and other Google services */} - {status.configured && ( -
-
-

Google Account

-

- Connect your Google account for Calendar, Contacts, and other Google services. This is separate from Gmail - sync above. -

-
- {status.connected ? ( - <> -
-
-
-

Connected

-

{status.email}

-
- {status.picture && ( - - )} -
- - - ) : ( - - )} -
)}
); diff --git a/src/apps/officer-web/Screens/Dashboard/Settings/IntegrationsSettings/index.tsx b/src/apps/officer-web/Screens/Dashboard/Settings/IntegrationsSettings/index.tsx index 3a756ca0..f4690197 100644 --- a/src/apps/officer-web/Screens/Dashboard/Settings/IntegrationsSettings/index.tsx +++ b/src/apps/officer-web/Screens/Dashboard/Settings/IntegrationsSettings/index.tsx @@ -1,5 +1,5 @@ import { useMemo } from 'react'; -import { Puzzle, KeyRound, UserCircle, MessageCircle, Globe, Wrench } from 'lucide-react'; +import { Puzzle, KeyRound, UserCircle, MessageCircle, Globe, Wrench, Mail } from 'lucide-react'; import type { LayoutNode, PanelComponents } from 'officerdev'; import { WorkspaceLayout } from 'officerdev'; import { useAuth } from 'hooks/useAuth'; @@ -17,6 +17,7 @@ import { WhatsAppBotConfig } from './WhatsAppBotConfig'; import { WhatsAppAccount } from './WhatsAppAccount'; import { BrowserRelay } from './BrowserRelay'; import { ApifyConfig } from './ApifyConfig'; +import { EmailAccounts } from './EmailAccounts'; const GLOBAL_KEY = 'INTEGRATIONS_SETTINGS_SELECTED'; const TAB_KEY = 'INTEGRATIONS_SETTINGS_TAB'; @@ -60,6 +61,13 @@ const enterpriseSections: SettingsSection[] = [ ]; const personalSections: SettingsSection[] = [ + { + key: 'email-accounts', + icon: Mail, + title: 'Email', + description: 'Connect email accounts for inbox sync', + content: , + }, { key: 'google-account', icon: UserCircle, diff --git a/src/databases/officer_db/src/index.ts b/src/databases/officer_db/src/index.ts index ae1f895f..865cef1f 100644 --- a/src/databases/officer_db/src/index.ts +++ b/src/databases/officer_db/src/index.ts @@ -20,7 +20,14 @@ export { export { readServerSettings, writeServerSettings, readConfigValue, writeConfigValue } from './queries/server-config'; -export { getUserSettings, setUserSettings, getUserState, patchUserState, getDockPaths, setDockPaths } from './queries/user-data'; +export { + getUserSettings, + setUserSettings, + getUserState, + patchUserState, + getDockPaths, + setDockPaths, +} from './queries/user-data'; export { getServerIntegrations, @@ -35,5 +42,15 @@ export { findUserByIntegrationConfig, } from './queries/integrations'; +export { + getEmailAccounts, + getEmailAccount, + createEmailAccount, + deleteEmailAccount, + updateEmailAccountStatus, + updateEmailAccountSyncMeta, + getAllSyncedAccounts, +} from './queries/email-accounts'; + export { db } from './db'; export * as schema from './schema'; diff --git a/src/databases/officer_db/src/queries/email-accounts.ts b/src/databases/officer_db/src/queries/email-accounts.ts new file mode 100644 index 00000000..a7d1803c --- /dev/null +++ b/src/databases/officer_db/src/queries/email-accounts.ts @@ -0,0 +1,41 @@ +import { eq, and } from 'drizzle-orm'; +import { db } from '../db'; +import { emailAccounts } from '../schema'; +import type { EmailAccountInsert, EmailAccountSelect } from '../types'; + +export async function getEmailAccounts(userId: number): Promise { + return db.select().from(emailAccounts).where(eq(emailAccounts.userId, userId)); +} + +export async function getEmailAccount(id: number): Promise { + const [row] = await db.select().from(emailAccounts).where(eq(emailAccounts.id, id)); + return row; +} + +export async function createEmailAccount(params: EmailAccountInsert): Promise { + const [row] = await db.insert(emailAccounts).values(params).returning(); + return row!; +} + +export async function deleteEmailAccount(id: number, userId: number): Promise { + const [row] = await db + .delete(emailAccounts) + .where(and(eq(emailAccounts.id, id), eq(emailAccounts.userId, userId))) + .returning({ id: emailAccounts.id }); + return !!row; +} + +export async function updateEmailAccountStatus(id: number, status: string): Promise { + await db.update(emailAccounts).set({ status }).where(eq(emailAccounts.id, id)); +} + +export async function updateEmailAccountSyncMeta(id: number, syncMeta: Record): Promise { + await db.update(emailAccounts).set({ syncMeta }).where(eq(emailAccounts.id, id)); +} + +export async function getAllSyncedAccounts(): Promise { + return db + .select() + .from(emailAccounts) + .where(and(eq(emailAccounts.status, 'synced'), eq(emailAccounts.enabled, true))); +} diff --git a/src/databases/officer_db/src/schema/email.ts b/src/databases/officer_db/src/schema/email.ts new file mode 100644 index 00000000..8c7faf84 --- /dev/null +++ b/src/databases/officer_db/src/schema/email.ts @@ -0,0 +1,26 @@ +import { pgTable, serial, integer, text, boolean, timestamp, jsonb, unique } from 'drizzle-orm/pg-core'; +import { users } from './auth'; + +export const emailAccounts = pgTable( + 'email_accounts', + { + id: serial('id').primaryKey(), + userId: integer('user_id') + .notNull() + .references(() => users.id, { onDelete: 'cascade' }), + provider: text('provider').notNull(), + email: text('email').notNull(), + displayName: text('display_name'), + imapHost: text('imap_host').notNull(), + imapPort: integer('imap_port').notNull(), + imapSecure: boolean('imap_secure').notNull().default(true), + authType: text('auth_type').notNull(), + credentials: jsonb('credentials').notNull().default({}), + enabled: boolean('enabled').notNull().default(true), + createdAt: timestamp('created_at', { withTimezone: true }).notNull().defaultNow(), + status: text('status').notNull().default('connected'), + syncMeta: jsonb('sync_meta').notNull().default({}), + updatedAt: timestamp('updated_at', { withTimezone: true }).notNull().defaultNow(), + }, + (table) => [unique('uq_email_accounts_user_email').on(table.userId, table.email)], +); diff --git a/src/databases/officer_db/src/schema/index.ts b/src/databases/officer_db/src/schema/index.ts index 09031760..35fd13c3 100644 --- a/src/databases/officer_db/src/schema/index.ts +++ b/src/databases/officer_db/src/schema/index.ts @@ -5,3 +5,4 @@ export * from './dashboards'; export * from './agent-items'; export * from './operations'; export * from './server'; +export * from './email'; diff --git a/src/databases/officer_db/src/types.ts b/src/databases/officer_db/src/types.ts index aac768ca..58676751 100644 --- a/src/databases/officer_db/src/types.ts +++ b/src/databases/officer_db/src/types.ts @@ -100,6 +100,11 @@ export type QueueJobInsert = typeof Schema.queueJobs.$inferInsert; export type TerminalContainerSelect = typeof Schema.terminalContainers.$inferSelect; export type TerminalContainerInsert = typeof Schema.terminalContainers.$inferInsert; +// ── Email ── + +export type EmailAccountSelect = typeof Schema.emailAccounts.$inferSelect; +export type EmailAccountInsert = typeof Schema.emailAccounts.$inferInsert; + // ── Server ── export type ServerConfigSelect = typeof Schema.serverConfig.$inferSelect; diff --git a/src/servers/api/email/accounts.ts b/src/servers/api/email/accounts.ts new file mode 100644 index 00000000..56b119f6 --- /dev/null +++ b/src/servers/api/email/accounts.ts @@ -0,0 +1,222 @@ +import { createRouter } from '../../create-router'; +import { BAD_REQUEST, NOT_FOUND } from '../../custom-errors'; +import { + getEmailAccounts, + getEmailAccount, + createEmailAccount, + deleteEmailAccount, + getUserIntegration, + updateEmailAccountStatus, +} from 'officerdb'; +import { validateImapConnection } from './imap-validate'; +import * as sidecar from '../../sidecar-client'; + +type CreateAccountBody = { + provider: string; + email: string; + displayName?: string; + imapHost: string; + imapPort: number; + imapSecure: boolean; + authType: string; + credentials: Record; +}; + +type ValidateBody = { + imapHost: string; + imapPort: number; + imapSecure: boolean; + authType: string; + email: string; + credentials: Record; +}; + +export const accountsRouter = createRouter(); + +accountsRouter.get('/', async (ctx) => { + const user = ctx.get('user'); + const accounts = await getEmailAccounts(user.id); + + // Check for stale syncing/queued accounts with no active job + const staleIds: number[] = []; + const hasActiveAccounts = accounts.some((a) => a.status === 'syncing' || a.status === 'queued'); + let activeJobAccountIds = new Set(); + + if (hasActiveAccounts) { + try { + const jobs = await sidecar.listJobs(); + activeJobAccountIds = new Set( + jobs + .filter((j) => j.type === 'email-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 + } + + for (const a of accounts) { + if ((a.status === 'syncing' || a.status === 'queued') && !activeJobAccountIds.has(a.id)) { + staleIds.push(a.id); + } + } + + // Reset stale accounts in background + if (staleIds.length > 0) { + for (const id of staleIds) { + updateEmailAccountStatus(id, 'connected').catch(() => {}); + } + } + } + + return ctx.json( + accounts.map((a) => ({ + id: a.id, + provider: a.provider, + email: a.email, + displayName: a.displayName, + enabled: a.enabled, + status: staleIds.includes(a.id) ? 'connected' : a.status, + createdAt: a.createdAt, + })), + ); +}); + +accountsRouter.post('/', async (ctx) => { + const user = ctx.get('user'); + const body = ctx.get('body') as CreateAccountBody; + + if (!body.provider || !body.email || !body.imapHost || !body.imapPort || !body.authType) { + throw BAD_REQUEST('Missing required fields'); + } + + const authResult = await resolveAuth(user.id, body.authType, body.email, body.credentials); + if (!authResult.ok) throw BAD_REQUEST(authResult.error); + + const validation = await validateImapConnection({ + host: body.imapHost, + port: body.imapPort, + secure: body.imapSecure, + user: body.email, + ...authResult.auth, + }); + + if (!validation.ok) throw BAD_REQUEST(`IMAP connection failed: ${validation.error}`); + + const account = await createEmailAccount({ + userId: user.id, + provider: body.provider, + email: body.email, + displayName: body.displayName, + imapHost: body.imapHost, + imapPort: body.imapPort, + imapSecure: body.imapSecure, + authType: body.authType, + credentials: body.credentials, + }); + + return ctx.json({ id: account.id, provider: account.provider, email: account.email }, 201); +}); + +accountsRouter.delete('/:id', async (ctx) => { + const user = ctx.get('user'); + const id = Number(ctx.req.param('id')); + const deleted = await deleteEmailAccount(id, user.id); + if (!deleted) throw NOT_FOUND('Account not found'); + return ctx.json({ ok: true }); +}); + +accountsRouter.post('/:id/sync', async (ctx) => { + const user = ctx.get('user'); + const id = Number(ctx.req.param('id')); + + const account = await getEmailAccount(id); + if (!account || account.userId !== user.id) throw NOT_FOUND('Account not found'); + + if (account.status === 'queued') throw BAD_REQUEST('Sync is already queued'); + if (account.status === 'syncing') throw BAD_REQUEST('Account is already syncing'); + + // Resolve auth before enqueueing + const authResult = await resolveAuth(user.id, account.authType, account.email, account.credentials as Record); + if (!authResult.ok) throw BAD_REQUEST(authResult.error); + + // Set status immediately so the UI reflects the queued state + await updateEmailAccountStatus(id, 'queued'); + + const job = await sidecar.enqueueJob({ + lane: 'email', + type: 'email-sync', + userId: user.email, + meta: { + emailAccountId: 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: { user: account.email, ...authResult.auth }, + }, + }); + + return ctx.json({ ok: true, jobId: job.id }, 201); +}); + +accountsRouter.post('/validate', async (ctx) => { + const user = ctx.get('user'); + const body = ctx.get('body') as ValidateBody; + + if (!body.imapHost || !body.imapPort || !body.authType || !body.email) { + throw BAD_REQUEST('Missing required fields'); + } + + const authResult = await resolveAuth(user.id, body.authType, body.email, body.credentials); + if (!authResult.ok) return ctx.json({ ok: false, error: authResult.error }); + + const result = await validateImapConnection({ + host: body.imapHost, + port: body.imapPort, + secure: body.imapSecure, + user: body.email, + ...authResult.auth, + }); + + return ctx.json(result); +}); + +type AuthResult = { ok: true; auth: { pass?: string; accessToken?: string } } | { ok: false; error: string }; + +async function resolveAuth( + userId: number, + authType: string, + email: string, + credentials: Record, +): Promise { + if (authType === 'oauth') { + const integrationId = credentials.userIntegrationId as number | undefined; + if (!integrationId) return { ok: false, error: 'Missing userIntegrationId for OAuth' }; + + const integration = await getUserIntegration(userId, 'google'); + if (!integration) return { ok: false, error: 'Google integration not found' }; + + const config = integration.config as Record; + const accessToken = config.accessToken as string | undefined; + if (!accessToken) return { ok: false, error: 'No access token available — reconnect Google account' }; + + return { ok: true, auth: { accessToken } }; + } + + if (authType === 'password') { + const pass = credentials.password as string | undefined; + if (!pass) return { ok: false, error: 'Missing password' }; + return { ok: true, auth: { pass } }; + } + + return { ok: false, error: `Unknown auth type: ${authType}` }; +} diff --git a/src/servers/api/email/email.ts b/src/servers/api/email/email.ts index a0c6f7ac..9ab1d122 100644 --- a/src/servers/api/email/email.ts +++ b/src/servers/api/email/email.ts @@ -4,9 +4,12 @@ import type { EmailMessage } from 'types'; import { createRouter } from '../../create-router'; import { DATA_PATH } from '@@/data-path'; import { openEmailDb, rowToSummary, getSyncMeta } from './email-db'; +import { accountsRouter } from './accounts'; export const emailRouter = createRouter(); +emailRouter.route('/accounts', accountsRouter); + emailRouter.get('/messages', async (ctx) => { const email = ctx.get('user').email; const page = Number(ctx.req.query('page') ?? '1'); @@ -18,7 +21,9 @@ emailRouter.get('/messages', async (ctx) => { const db = openEmailDb(email); try { - const rows = db.query(`SELECT * FROM emails WHERE ${folderWhere} ORDER BY date DESC LIMIT ? OFFSET ?`).all(limit, offset) as Record[]; + const rows = db + .query(`SELECT * FROM emails WHERE ${folderWhere} ORDER BY date DESC LIMIT ? OFFSET ?`) + .all(limit, offset) as Record[]; const countRow = db.query(`SELECT COUNT(*) as total FROM emails WHERE ${folderWhere}`).get() as { total: number }; const messages = rows.map(rowToSummary); return ctx.json({ messages, total: countRow.total }); @@ -36,11 +41,11 @@ emailRouter.get('/messages/:id', async (ctx) => { const row = db.query('SELECT * FROM emails WHERE id = ? AND deleted = 0').get(id) as Record | null; if (!row) return ctx.text('Not found', 404); - const attachmentRows = db.query('SELECT * FROM attachments WHERE email_id = ? ORDER BY idx').all(id) as Array>; + const attachmentRows = db.query('SELECT * FROM attachments WHERE email_id = ? ORDER BY idx').all(id) as Array< + Record + >; - const from = row.from_name - ? `${row.from_name} <${row.from_address}>` - : (row.from_address as string); + const from = row.from_name ? `${row.from_name} <${row.from_address}>` : (row.from_address as string); const message: EmailMessage = { id: row.id as string, @@ -75,7 +80,10 @@ emailRouter.post('/messages/:id/attachments/:index/extract', async (ctx) => { const db = openEmailDb(email); try { - const row = db.query('SELECT filename, content FROM attachments WHERE email_id = ? AND idx = ?').get(id, index) as { filename: string; content: string | null } | null; + const row = db.query('SELECT filename, content FROM attachments WHERE email_id = ? AND idx = ?').get(id, index) as { + filename: string; + content: string | null; + } | null; if (!row || !row.content) return ctx.text('Attachment not found', 404); const fileName = row.filename ?? 'unknown'; @@ -144,9 +152,18 @@ emailRouter.get('/stats', async (ctx) => { const db = openEmailDb(email); try { - const total = (db.query(`SELECT COUNT(*) as count FROM emails WHERE ${folderWhere}`).get() 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 }>; - 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 }>; + const total = (db.query(`SELECT COUNT(*) as count FROM emails WHERE ${folderWhere}`).get() 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 }>; + 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 }>; return ctx.json({ total, byDomain, bySender }); } finally { @@ -159,7 +176,9 @@ emailRouter.get('/labels', async (ctx) => { const db = openEmailDb(email); try { - const rows = db.query('SELECT labels FROM emails WHERE deleted = 0 AND labels IS NOT NULL').all() as Array<{ labels: string }>; + const rows = db.query('SELECT labels FROM emails WHERE deleted = 0 AND labels IS NOT NULL').all() as Array<{ + labels: string; + }>; const counts = new Map(); for (const row of rows) { @@ -169,9 +188,7 @@ emailRouter.get('/labels', async (ctx) => { } } - const labels = [...counts.entries()] - .map(([label, count]) => ({ label, count })) - .sort((a, b) => b.count - a.count); + const labels = [...counts.entries()].map(([label, count]) => ({ label, count })).sort((a, b) => b.count - a.count); return ctx.json({ labels }); } finally { diff --git a/src/servers/api/email/imap-validate.ts b/src/servers/api/email/imap-validate.ts new file mode 100644 index 00000000..def5c082 --- /dev/null +++ b/src/servers/api/email/imap-validate.ts @@ -0,0 +1,54 @@ +type ValidateImapParams = { + host: string; + port: number; + secure: boolean; + user: string; + pass?: string; + accessToken?: string; +}; + +type ValidateImapResult = { ok: true; folderCount: number } | { ok: false; error: string }; + +export async function validateImapConnection(params: ValidateImapParams): Promise { + const { ImapFlow } = await import('imapflow'); + + const auth: { user: string; pass?: string; accessToken?: string } = { user: params.user }; + if (params.accessToken) { + auth.accessToken = params.accessToken; + } else if (params.pass) { + auth.pass = params.pass; + } else { + return { ok: false, error: 'No authentication credentials provided' }; + } + + const client = new ImapFlow({ + host: params.host, + port: params.port, + secure: params.secure, + auth, + logger: false, + greetingTimeout: 60_000, + socketTimeout: 60_000, + }); + + try { + const result = await Promise.race([ + (async () => { + await client.connect(); + const folders = await client.list(); + await client.logout(); + return { ok: true as const, folderCount: folders.length }; + })(), + new Promise((_, reject) => + setTimeout(() => reject(new Error('Connection timed out')), 90_000), + ), + ]); + return result; + } catch (err) { + try { + client.close(); + } catch {} + const message = err instanceof Error ? err.message : 'Connection failed'; + return { ok: false, error: message }; + } +} diff --git a/src/servers/api/integrations/google-auth.ts b/src/servers/api/integrations/google-auth.ts new file mode 100644 index 00000000..e461db27 --- /dev/null +++ b/src/servers/api/integrations/google-auth.ts @@ -0,0 +1,34 @@ +import { getServerIntegration } from 'officerdb'; +import { PermanentError } from '../../queue/types'; + +type TokenRefreshResult = { accessToken: string; expiresAt: number }; + +export async function refreshGoogleAccessToken(refreshToken: string): Promise { + const serverGoogle = await getServerIntegration('google'); + const serverConfig = serverGoogle?.config as Record | undefined; + if (!serverConfig?.clientId || !serverConfig?.clientSecret) { + throw new PermanentError('Google OAuth not configured on server'); + } + + const response = await fetch('https://oauth2.googleapis.com/token', { + method: 'POST', + headers: { 'Content-Type': 'application/x-www-form-urlencoded' }, + body: new URLSearchParams({ + client_id: serverConfig.clientId as string, + client_secret: serverConfig.clientSecret as string, + refresh_token: refreshToken, + grant_type: 'refresh_token', + }), + }); + + if (!response.ok) { + const text = await response.text(); + throw new Error(`Google token refresh failed: ${text}`); + } + + const data = (await response.json()) as { access_token: string; expires_in: number }; + return { + accessToken: data.access_token, + expiresAt: Date.now() + data.expires_in * 1000, + }; +} diff --git a/src/servers/api/integrations/integrations.ts b/src/servers/api/integrations/integrations.ts index 02cfb312..f3317bc2 100644 --- a/src/servers/api/integrations/integrations.ts +++ b/src/servers/api/integrations/integrations.ts @@ -259,11 +259,16 @@ export const googleCallbackHandler = async (ctx: any) => { const serverIntegration = await getServerIntegration('google'); + // Merge with existing config to preserve fields like imapAppPassword + const existingIntegration = await getUserIntegration(dbUser.id, 'google'); + const existingConfig = (existingIntegration?.config as Record) ?? {}; + await upsertUserIntegration({ userId: dbUser.id, provider: 'google', serverIntegrationId: serverIntegration?.id, config: { + ...existingConfig, accessToken: tokens.access_token, refreshToken: tokens.refresh_token, expiresAt: Date.now() + tokens.expires_in * 1000, diff --git a/src/servers/queue/handlers/email-sync.ts b/src/servers/queue/handlers/email-sync.ts new file mode 100644 index 00000000..6e11706c --- /dev/null +++ b/src/servers/queue/handlers/email-sync.ts @@ -0,0 +1,381 @@ +import { createHash } from 'node:crypto'; +import type { JobHandler } from '../types'; +import { PermanentError } from '../types'; +import { registerHandler } from '../handler-registry'; +import { openEmailDb, upsertFromRawEml } from '../../api/email/email-db'; +import { refreshGoogleAccessToken } from '../../api/integrations/google-auth'; +import { + getEmailAccount, + getUserIntegration, + upsertUserIntegration, + getServerIntegration, + updateEmailAccountStatus, + updateEmailAccountSyncMeta, + getDockPaths, + setDockPaths, +} from 'officerdb'; + +// ── Types for job meta (passed by the API server at enqueue time) ── + +type EmailSyncMeta = { + emailAccountId: number; + userEmail: string; + account: { + id: number; + userId: number; + email: string; + imapHost: string; + imapPort: number; + imapSecure: boolean; + provider: string; + authType: string; + credentials: Record; + }; + imapAuth: { user: string; pass?: string; accessToken?: string }; + saved?: number; +}; + +// ── Stable ID from Message-Id header ── + +function messageIdToStableId(raw: string): string | null { + const match = raw.match(/^Message-Id:\s*\s]+)>?/im); + if (!match?.[1]) return null; + return createHash('sha1').update(match[1]).digest('hex').slice(0, 16); +} + +// ── IMAP folder → label mapping ── + +const SPECIAL_USE_LABEL_MAP: Record = { + '\\Inbox': 'inbox', + '\\Sent': 'sent', + '\\Drafts': 'draft', + '\\Flagged': 'starred', + '\\Trash': 'trash', + '\\Junk': 'spam', + '\\All': 'archive', +}; + +const SKIP_SPECIAL_USE = new Set(['\\Trash', '\\Junk', '\\All']); + +type FolderInfo = { specialUse?: string; path: string; flags: Set; status?: { uidNext?: number; uidValidity?: number } }; + +function shouldSkipFolder(folder: FolderInfo): boolean { + if (folder.flags.has('\\Noselect') || folder.flags.has('\\NonExistent')) return true; + if (folder.specialUse && SKIP_SPECIAL_USE.has(folder.specialUse)) return true; + return false; +} + +function folderToLabel(folder: FolderInfo): string { + if (folder.specialUse && SPECIAL_USE_LABEL_MAP[folder.specialUse]) { + return SPECIAL_USE_LABEL_MAP[folder.specialUse]!; + } + if (folder.path === 'INBOX') return 'inbox'; + return folder.path.toLowerCase(); +} + +// ── OAuth token refresh ── + +async function resolveImapAuth(meta: EmailSyncMeta): Promise<{ user: string; pass?: string; accessToken?: string }> { + if (meta.account.authType !== 'oauth') return meta.imapAuth; + + // Refresh OAuth token if expired + const userGoogle = await getUserIntegration(meta.account.userId, 'google'); + const config = userGoogle?.config as Record | undefined; + if (!config?.refreshToken) { + throw new PermanentError('Google OAuth not configured — reconnect your Google account'); + } + + const expiresAt = config.expiresAt as number | undefined; + const tokenExpired = !expiresAt || expiresAt < Date.now() + 60_000; + + if (!tokenExpired && config.accessToken) { + return { user: meta.account.email, accessToken: config.accessToken as string }; + } + + console.log('[email-sync] Refreshing OAuth token'); + const refreshed = await refreshGoogleAccessToken(config.refreshToken as string); + + const serverGoogle = await getServerIntegration('google'); + await upsertUserIntegration({ + userId: meta.account.userId, + provider: 'google', + serverIntegrationId: serverGoogle?.id, + config: { ...config, accessToken: refreshed.accessToken, expiresAt: refreshed.expiresAt }, + }); + + return { user: meta.account.email, accessToken: refreshed.accessToken }; +} + +// ── Handler ── + +const emailSyncHandler: JobHandler = { + type: 'email-sync', + retry: { delayMs: 15 * 60 * 1000, maxRetries: 5 }, + steps: [ + { + name: 'Sync emails', + run: async (ctx) => { + const { ImapFlow } = await import('imapflow'); + const meta = ctx.meta as unknown as EmailSyncMeta; + const { account, userEmail } = meta; + + // Resolve auth (refreshes OAuth token if needed) + const imapAuth = await resolveImapAuth(meta); + + // Load syncMeta from DB (always fresh, not from job meta) + const freshAccount = await getEmailAccount(account.id); + if (!freshAccount) throw new PermanentError(`Email account ${account.id} not found`); + const syncMeta = (freshAccount.syncMeta ?? {}) as Record; + const isIncremental = !!syncMeta.last_sync_at; + + await updateEmailAccountStatus(account.id, 'syncing'); + console.log(`[email-sync] ${isIncremental ? 'Incremental' : 'Initial'} sync for ${account.email}`); + + const MAX_RECONNECTS = 10; + const RECONNECT_DELAY_MS = 5_000; + let reconnects = 0; + let saved = 0; + let skipped = 0; + let errors = 0; + let allDone = false; + + const db = openEmailDb(userEmail); + + // Load existing IDs for dedup (once, shared across reconnections) + const existingIds = new Set(); + const rows = db.query('SELECT id FROM emails').all() as Array<{ id: string }>; + for (const row of rows) existingIds.add(row.id); + + try { + while (!allDone && reconnects <= MAX_RECONNECTS) { + if (reconnects > 0) { + console.log(`[email-sync] Reconnecting (${reconnects}/${MAX_RECONNECTS}) after ${RECONNECT_DELAY_MS / 1000}s...`); + await ctx.updateProgress({ current: saved + skipped, total: 0, label: `Reconnecting (${reconnects}/${MAX_RECONNECTS})...` }); + await new Promise((r) => setTimeout(r, RECONNECT_DELAY_MS)); + + // Re-read syncMeta from DB to get latest saved UIDs + const updated = await getEmailAccount(account.id); + if (updated?.syncMeta) { + Object.assign(syncMeta, updated.syncMeta as Record); + } + } else { + await ctx.updateProgress({ current: 0, total: 0, label: 'Connecting...' }); + } + + const client = new ImapFlow({ + host: account.imapHost, + port: account.imapPort, + secure: account.imapSecure, + auth: imapAuth, + logger: false, + socketTimeout: 30 * 60 * 1000, + }); + + const connState = { error: null as Error | null }; + client.on('error', (err: Error) => { + console.log(`[email-sync] IMAP connection error: ${err.message}`); + connState.error = err; + }); + + try { + await client.connect(); + console.log(`[email-sync] IMAP connected${reconnects > 0 ? ` (reconnect ${reconnects})` : ''}`); + + const folders = await client.list({ statusQuery: { uidNext: true, uidValidity: true } }) as FolderInfo[]; + + // Filter to folders that still need syncing + const foldersToSync: Array<{ folder: FolderInfo; lastUid: number }> = []; + for (const folder of folders) { + if (shouldSkipFolder(folder)) continue; + + const uidValidityKey = `imap_uidvalidity:${folder.path}`; + const lastUidKey = `imap_lastuid:${folder.path}`; + const storedUidValidity = syncMeta[uidValidityKey] as string | undefined; + const storedLastUid = syncMeta[lastUidKey] as string | undefined; + + const uidValidity = folder.status?.uidValidity ? String(folder.status.uidValidity) : null; + const uidNext = folder.status?.uidNext ?? 0; + let lastUid = storedUidValidity && uidValidity === storedUidValidity && storedLastUid ? parseInt(storedLastUid, 10) : 0; + + if (storedUidValidity && uidValidity && storedUidValidity !== uidValidity) { + console.log(`[email-sync] UIDVALIDITY changed for ${folder.path} — will re-scan`); + lastUid = 0; + } + + if (lastUid > 0 && uidNext <= lastUid + 1) continue; + + foldersToSync.push({ folder, lastUid }); + } + + if (foldersToSync.length === 0) { + console.log('[email-sync] All folders synced'); + allDone = true; + await client.logout().catch(() => {}); + break; + } + + console.log(`[email-sync] ${foldersToSync.length} folder(s) to sync`); + + let connectionLost = false; + + for (let fi = 0; fi < foldersToSync.length; fi++) { + const { folder, lastUid } = foldersToSync[fi]!; + + if (connState.error) { connectionLost = true; break; } + + console.log(`[email-sync] Syncing folder ${fi + 1}/${foldersToSync.length}: ${folder.path}`); + + let lock; + try { + lock = await client.getMailboxLock(folder.path); + } catch (err) { + const errMsg = err instanceof Error ? err.message : String(err); + if (errMsg.includes('not available') || errMsg.includes('timeout') || errMsg.includes('closed')) { + console.log(`[email-sync] Connection lost at folder ${folder.path}`); + connectionLost = true; + break; + } + console.log(`[email-sync] Skipping folder ${folder.path}: ${errMsg}`); + continue; + } + + try { + const mailbox = client.mailbox; + if (!mailbox) continue; + + const uidValidity = String(mailbox.uidValidity); + const effectiveLastUid = syncMeta[`imap_uidvalidity:${folder.path}`] === uidValidity ? lastUid : 0; + let maxUid = effectiveLastUid; + const label = folderToLabel(folder); + const range = effectiveLastUid > 0 ? `${effectiveLastUid + 1}:*` : '1:*'; + + try { + const fetchOpts: Record = { source: true, uid: true }; + if (account.provider === 'gmail') fetchOpts.labels = true; + + for await (const msg of client.fetch(range, fetchOpts, { uid: true })) { + if (msg.uid <= effectiveLastUid) continue; + + maxUid = Math.max(maxUid, msg.uid); + + if (!msg.source) { errors++; continue; } + + const raw = msg.source.toString('utf-8'); + const id = messageIdToStableId(raw); + if (!id) { errors++; continue; } + + if (existingIds.has(id)) { skipped++; continue; } + + const labels = [label]; + try { + upsertFromRawEml({ db, id, raw, integration: account.provider, emailAccount: account.email, labels }); + existingIds.add(id); + saved++; + } catch { + errors++; + } + + if ((saved + skipped) % 50 === 0) { + const progressLabel = `${folder.path} — ${saved.toLocaleString()} saved, ${skipped.toLocaleString()} skipped`; + console.log(`[email-sync] ${progressLabel}`); + await ctx.updateProgress({ current: saved + skipped, total: 0, label: progressLabel }); + + if (maxUid > effectiveLastUid) { + syncMeta[`imap_lastuid:${folder.path}`] = String(maxUid); + await updateEmailAccountSyncMeta(account.id, syncMeta as Record); + } + } + } + } catch (fetchErr) { + const errMsg = fetchErr instanceof Error ? fetchErr.message : String(fetchErr); + if (errMsg.includes('Nothing to fetch')) { + // No messages in range — normal + } else if (errMsg.includes('not available') || errMsg.includes('timeout') || errMsg.includes('closed')) { + console.log(`[email-sync] Connection lost during fetch in ${folder.path}: ${errMsg}`); + connectionLost = true; + } else { + console.log(`[email-sync] Fetch error in ${folder.path}: ${errMsg}`); + errors++; + } + } + + syncMeta[`imap_uidvalidity:${folder.path}`] = uidValidity; + if (maxUid > effectiveLastUid) { + syncMeta[`imap_lastuid:${folder.path}`] = String(maxUid); + } + + console.log(`[email-sync] ${folder.path} done: ${saved} saved, ${skipped} skipped, ${errors} errors`); + await ctx.updateProgress({ current: saved + skipped, total: 0, label: `${folder.path}: ${saved} saved` }); + await updateEmailAccountSyncMeta(account.id, syncMeta as Record); + } finally { + lock.release(); + } + + if (connectionLost) break; + } + + await client.logout().catch(() => {}); + + if (connectionLost) { + console.log(`[email-sync] Connection lost after saving ${saved} emails — will reconnect`); + reconnects++; + continue; + } + + allDone = true; + } catch (err) { + await client.logout().catch(() => {}); + if (reconnects < MAX_RECONNECTS) { + console.log(`[email-sync] Error: ${err instanceof Error ? err.message : String(err)} — will reconnect`); + reconnects++; + continue; + } + throw err; + } + } + + if (!allDone) { + throw new Error(`IMAP sync incomplete after ${MAX_RECONNECTS} reconnection attempts`); + } + + // Mark sync complete + syncMeta.last_sync_at = new Date().toISOString(); + await updateEmailAccountSyncMeta(account.id, syncMeta as Record); + } finally { + db.close(); + } + + console.log(`[email-sync] Done: saved ${saved}, skipped ${skipped}, errors ${errors}, reconnects ${reconnects}`); + ctx.meta.saved = saved; + await ctx.updateProgress({ current: 1, total: 1, label: `Done — ${saved} new emails` }); + }, + }, + { + name: 'Finalize', + run: async (ctx) => { + const meta = ctx.meta as unknown as EmailSyncMeta; + const saved = meta.saved ?? 0; + + await updateEmailAccountStatus(meta.account.id, 'synced'); + + if (saved > 0) { + try { + const paths = await getDockPaths(meta.account.userId); + if (!paths) { + const defaults = ['/', '/files', '/automation', '/projects', '/dashboards', '/chat']; + await setDockPaths(meta.account.userId, [...defaults, '/email']); + } else if (!paths.includes('/email')) { + await setDockPaths(meta.account.userId, [...paths, '/email']); + } + } catch { + // Non-fatal + } + } + + console.log(`[email-sync] ${meta.account.email} status set to synced`); + }, + }, + ], +}; + +registerHandler(emailSyncHandler); diff --git a/src/servers/queue/handlers/gmail-sync.ts b/src/servers/queue/handlers/gmail-sync.ts index 316eb2db..c77fae14 100644 --- a/src/servers/queue/handlers/gmail-sync.ts +++ b/src/servers/queue/handlers/gmail-sync.ts @@ -4,28 +4,59 @@ import { join } from 'node:path'; import { readdir, readFile, writeFile, unlink, mkdir, stat } from 'node:fs/promises'; import { type JobHandler, PermanentError } from '../types'; import { registerHandler } from '../handler-registry'; -import { openEmailDb, upsertFromRawEml, setSyncMeta } from '../../api/email/email-db'; -import { getUserByEmail, getUserIntegration, getDockPaths, setDockPaths } from 'officerdb'; +import { openEmailDb, upsertFromRawEml, setSyncMeta, getSyncMeta } from '../../api/email/email-db'; +import { + getUserByEmail, + getUserIntegration, + upsertUserIntegration, + getServerIntegration, + getDockPaths, + setDockPaths, +} from 'officerdb'; import { getMaildirPath } from '@@/data-path'; +import { refreshGoogleAccessToken } from '@@/api/integrations/google-auth'; // ── Credentials ── -type ImapCredentials = { email: string; appPassword: string }; +type GmailCredentials = { + email: string; + userId: number; + appPassword?: string; + accessToken?: string; + refreshToken?: string; + expiresAt?: number; +}; -async function loadImapCredentials(userEmail: string): Promise { +async function loadGmailCredentials(userEmail: string): Promise { const dbUser = await getUserByEmail(userEmail); if (!dbUser) throw new PermanentError('User not found'); const userGoogle = await getUserIntegration(dbUser.id, 'google'); const config = userGoogle?.config as Record | undefined; + + const gmailEmail = (config?.email as string) ?? userEmail; + + // Prefer OAuth tokens, fall back to app password + if (config?.accessToken && config?.refreshToken) { + return { + email: gmailEmail, + userId: dbUser.id, + accessToken: config.accessToken as string, + refreshToken: config.refreshToken as string, + expiresAt: config.expiresAt as number | undefined, + appPassword: config.imapAppPassword as string | undefined, + }; + } + if (!config?.imapAppPassword) { throw new PermanentError('Gmail App Password not configured — set it in Settings → Integrations'); } - const gmailEmail = (config.email as string) ?? userEmail; - return { email: gmailEmail, appPassword: config.imapAppPassword as string }; + return { email: gmailEmail, userId: dbUser.id, appPassword: config.imapAppPassword as string }; } +// refreshGoogleAccessToken is imported from @@/api/integrations/google-auth + // ── mbsync config ── function buildMbsyncConfig(email: string, appPassword: string, maildirPath: string): string { @@ -48,7 +79,7 @@ SubFolders Verbatim Channel gmail Far :gmail-remote: Near :gmail-local: -Patterns * ![Gmail]/Trash ![Gmail]/Spam ![Gmail]/Bin ![Google Mail]/Trash ![Google Mail]/Spam ![Google Mail]/Bin +Patterns * ![Gmail]/Trash ![Gmail]/Spam ![Gmail]/Bin !"[Gmail]/All Mail" ![Google Mail]/Trash ![Google Mail]/Spam ![Google Mail]/Bin !"[Google Mail]/All Mail" Create Near Expunge None SyncState * @@ -99,21 +130,40 @@ function messageIdToStableId(raw: string): string | null { type ImportResult = { saved: number; skipped: number; errors: number }; -export async function importMaildir( - maildirPath: string, - emailAccount: string, - db: Database, - onProgress?: (saved: number, skipped: number) => void, -): Promise { +type ImportMaildirParams = { + maildirPath: string; + emailAccount: string; + db: Database; + lastSyncAt?: string | null; + onProgress?: (saved: number, skipped: number) => void; +}; + +export async function importMaildir({ + maildirPath, + emailAccount, + db, + lastSyncAt, + onProgress, +}: ImportMaildirParams): Promise { let saved = 0; let skipped = 0; let errors = 0; + // For incremental syncs, skip files older than last sync (with 60s buffer for clock skew) + const mtimeCutoff = lastSyncAt ? new Date(lastSyncAt).getTime() - 60_000 : 0; + const isIncremental = mtimeCutoff > 0; + // Load existing IDs for fast dedup const existingIds = new Set(); const rows = db.query('SELECT id FROM emails').all() as Array<{ id: string }>; for (const row of rows) existingIds.add(row.id); + if (isIncremental) { + console.log( + `[gmail-sync] Incremental import — only reading files newer than ${new Date(mtimeCutoff).toISOString()}`, + ); + } + // First pass: collect all message files with their folders to build label map const messageIdLabels = new Map>(); const messageFiles = new Map(); // id → first file path @@ -162,6 +212,12 @@ export async function importMaildir( for (const file of files) { const filePath = join(dirPath, file); try { + // Skip files older than last sync for incremental imports + if (isIncremental) { + const fileStat = await stat(filePath); + if (fileStat.mtimeMs < mtimeCutoff) continue; + } + const raw = await readFile(filePath, 'utf-8'); const id = messageIdToStableId(raw); if (!id) { @@ -238,6 +294,220 @@ async function countMaildirFiles(maildirPath: string): Promise { return count; } +// ── Gmail IMAP label → our label mapping ── + +const GMAIL_LABEL_MAP: Record = { + '\\Inbox': 'inbox', + '\\Sent': 'sent', + '\\Drafts': 'draft', + '\\Starred': 'starred', + '\\Important': 'important', + '\\All': 'archive', + '\\Trash': 'trash', + '\\Junk': 'spam', +}; + +function gmailLabelsToLabels(gmailLabels: Set): string[] { + const labels: string[] = []; + for (const gl of gmailLabels) { + const mapped = GMAIL_LABEL_MAP[gl]; + if (mapped) { + labels.push(mapped); + } else if (!gl.startsWith('\\')) { + // Custom label — lowercase it + labels.push(gl.toLowerCase()); + } + } + // If only "archive" and no specific folder, keep it; otherwise drop "archive" + if (labels.length > 1 && labels.includes('archive')) { + return labels.filter((l) => l !== 'archive'); + } + return labels; +} + +// ── IMAP special-use folders to skip ── + +const SKIP_SPECIAL_USE = new Set(['\\Trash', '\\Junk', '\\All']); + +function shouldSkipFolder(folder: { specialUse?: string; path: string; flags: Set }): boolean { + if (folder.flags.has('\\Noselect') || folder.flags.has('\\NonExistent')) return true; + if (folder.specialUse && SKIP_SPECIAL_USE.has(folder.specialUse)) return true; + const norm = normalizeGmailFolder(folder.path); + return norm === 'All Mail' || norm === 'Trash' || norm === 'Spam' || norm === 'Bin'; +} + +// ── Incremental IMAP sync ── + +type IncrementalSyncParams = { + creds: GmailCredentials; + db: Database; + onProgress?: (fetched: number, folder: string) => void; +}; + +type IncrementalSyncResult = { saved: number; skipped: number; errors: number }; + +async function incrementalImapSync({ creds, db, onProgress }: IncrementalSyncParams): Promise { + const { ImapFlow } = await import('imapflow'); + + // Determine auth method: prefer OAuth, fall back to app password + const auth: { user: string; pass?: string; accessToken?: string } = { user: creds.email }; + if (creds.accessToken) { + auth.accessToken = creds.accessToken; + } else if (creds.appPassword) { + auth.pass = creds.appPassword; + } else { + throw new PermanentError('No authentication method available for IMAP'); + } + + const client = new ImapFlow({ + host: 'imap.gmail.com', + port: 993, + secure: true, + auth, + logger: false, + }); + + let saved = 0; + let skipped = 0; + let errors = 0; + + try { + await client.connect(); + console.log('[gmail-sync] Incremental IMAP connected'); + + // Get all folders with status in a single LIST command + const folders = await client.list({ statusQuery: { uidNext: true, uidValidity: true } }); + + // Load existing IDs for dedup + const existingIds = new Set(); + const rows = db.query('SELECT id FROM emails').all() as Array<{ id: string }>; + for (const row of rows) existingIds.add(row.id); + + // Filter to only folders with new messages + const foldersToSync: typeof folders = []; + for (const folder of folders) { + if (shouldSkipFolder(folder)) continue; + + const folderPath = folder.path; + const uidValidityKey = `imap_uidvalidity:${folderPath}`; + const lastUidKey = `imap_lastuid:${folderPath}`; + const storedUidValidity = getSyncMeta(db, uidValidityKey); + const storedLastUid = getSyncMeta(db, lastUidKey); + + const uidValidity = folder.status?.uidValidity ? String(folder.status.uidValidity) : null; + const uidNext = folder.status?.uidNext ?? 0; + const lastUid = + storedUidValidity && uidValidity === storedUidValidity && storedLastUid ? parseInt(storedLastUid, 10) : 0; + + if (uidValidity) setSyncMeta(db, uidValidityKey, uidValidity); + + if (lastUid > 0 && uidNext <= lastUid + 1) continue; // no new messages + + if (storedUidValidity && uidValidity && storedUidValidity !== uidValidity) { + console.log(`[gmail-sync] UIDVALIDITY changed for ${folderPath} — will re-scan`); + } + + foldersToSync.push(folder); + } + + console.log(`[gmail-sync] ${foldersToSync.length} folder(s) with new messages`); + + for (const folder of foldersToSync) { + const folderPath = folder.path; + const uidValidityKey = `imap_uidvalidity:${folderPath}`; + const lastUidKey = `imap_lastuid:${folderPath}`; + const storedUidValidity = getSyncMeta(db, uidValidityKey); + const storedLastUid = getSyncMeta(db, lastUidKey); + + // Open folder and fetch new messages + let lock; + try { + lock = await client.getMailboxLock(folderPath); + } catch (err) { + console.log(`[gmail-sync] Skipping folder ${folderPath}: ${err instanceof Error ? err.message : err}`); + continue; + } + + try { + const mailbox = client.mailbox; + if (!mailbox) continue; + + const uidValidity = String(mailbox.uidValidity); + let lastUid = 0; + if (storedUidValidity === uidValidity && storedLastUid) { + lastUid = parseInt(storedLastUid, 10); + } + + const range = lastUid > 0 ? `${lastUid + 1}:*` : '1:*'; + let maxUid = lastUid; + let folderFetched = 0; + + try { + for await (const msg of client.fetch(range, { source: true, labels: true, uid: true }, { uid: true })) { + if (msg.uid <= lastUid) continue; + + maxUid = Math.max(maxUid, msg.uid); + folderFetched++; + + if (!msg.source) { + errors++; + continue; + } + + const raw = msg.source.toString('utf-8'); + const id = messageIdToStableId(raw); + if (!id) { + errors++; + continue; + } + + if (existingIds.has(id)) { + skipped++; + continue; + } + + // Gmail X-GM-EXT-1 labels don't include folder membership (e.g. \Inbox), + // so always use the folder we're fetching from as the base label + const folderLabel = folderToLabel(folderPath); + const gmailLabels = msg.labels ? gmailLabelsToLabels(msg.labels) : []; + const labels = folderLabel ? [...new Set([folderLabel, ...gmailLabels])] : gmailLabels; + + try { + upsertFromRawEml({ db, id, raw, integration: 'gmail', emailAccount: creds.email, labels }); + existingIds.add(id); + saved++; + } catch { + errors++; + } + } + } catch (fetchErr) { + const msg = fetchErr instanceof Error ? fetchErr.message : String(fetchErr); + if (!msg.includes('Nothing to fetch')) { + console.log(`[gmail-sync] Fetch error in ${folderPath}: ${msg}`); + errors++; + } + } + + if (maxUid > lastUid) { + setSyncMeta(db, lastUidKey, String(maxUid)); + } + setSyncMeta(db, uidValidityKey, uidValidity); + + if (folderFetched > 0) { + console.log(`[gmail-sync] ${folderPath}: fetched ${folderFetched}, saved ${saved}, skipped ${skipped}`); + onProgress?.(saved + skipped, folderPath); + } + } finally { + lock.release(); + } + } + } finally { + await client.logout().catch(() => {}); + } + + return { saved, skipped, errors }; +} + // ── Handler ── const gmailSyncHandler: JobHandler = { @@ -247,119 +517,219 @@ const gmailSyncHandler: JobHandler = { { name: 'Verify credentials', run: async (ctx) => { - const creds = await loadImapCredentials(ctx.job.userId); + const creds = await loadGmailCredentials(ctx.job.userId); + + // Determine sync mode: incremental if we have a previous sync + const db = openEmailDb(ctx.job.userId); + try { + const lastSyncAt = getSyncMeta(db, 'last_sync_at'); + ctx.meta.isIncremental = !!lastSyncAt; + } finally { + db.close(); + } + + // For incremental sync with OAuth, refresh token if expired + if (ctx.meta.isIncremental && creds.accessToken && creds.refreshToken) { + const tokenExpired = !creds.expiresAt || creds.expiresAt < Date.now() + 60_000; + if (tokenExpired) { + console.log('[gmail-sync] Refreshing OAuth token for incremental sync'); + try { + const refreshed = await refreshGoogleAccessToken(creds.refreshToken); + creds.accessToken = refreshed.accessToken; + creds.expiresAt = refreshed.expiresAt; + + // Persist refreshed token + const userGoogle = await getUserIntegration(creds.userId, 'google'); + const existingConfig = (userGoogle?.config as Record) ?? {}; + const serverGoogle = await getServerIntegration('google'); + await upsertUserIntegration({ + userId: creds.userId, + provider: 'google', + serverIntegrationId: serverGoogle?.id, + config: { ...existingConfig, accessToken: refreshed.accessToken, expiresAt: refreshed.expiresAt }, + }); + } catch (err) { + console.log(`[gmail-sync] OAuth refresh failed, will fall back to app password: ${err}`); + // Clear OAuth so we fall back to app password + creds.accessToken = undefined; + creds.refreshToken = undefined; + } + } + } + + ctx.meta.creds = creds; ctx.meta.email = creds.email; ctx.meta.appPassword = creds.appPassword; + + if (ctx.meta.isIncremental) { + console.log(`[gmail-sync] Incremental sync mode (${creds.accessToken ? 'OAuth' : 'App Password'})`); + } else { + console.log('[gmail-sync] Full sync mode (mbsync)'); + if (!creds.appPassword) { + throw new PermanentError('Gmail App Password required for initial sync'); + } + } }, }, { - name: 'Sync via mbsync', + name: 'Sync emails', run: async (ctx) => { - const email = ctx.meta.email as string; - const appPassword = ctx.meta.appPassword as string; - const maildirPath = getMaildirPath(ctx.job.userId); + if (ctx.meta.isIncremental) { + // ── Incremental: direct IMAP via imapflow ── + const creds = ctx.meta.creds as GmailCredentials; + await ctx.updateProgress({ current: 0, total: 0, label: 'Connecting to Gmail...' }); - // Ensure Maildir root exists - await mkdir(maildirPath, { recursive: true }); - - // Write temp config - const configPath = join(maildirPath, '.mbsyncrc'); - const config = buildMbsyncConfig(email, appPassword, maildirPath); - await writeFile(configPath, config, { mode: 0o600 }); - - await ctx.updateProgress({ current: 0, total: 0, label: 'Running mbsync...' }); - - try { - let proc: ReturnType; + const db = openEmailDb(ctx.job.userId); try { - proc = Bun.spawn(['mbsync', '-c', configPath, '-a', '-V'], { - stdout: 'pipe', - stderr: 'pipe', + const result = await incrementalImapSync({ + creds, + db, + onProgress: (fetched, folder) => { + ctx.updateProgress({ current: fetched, total: 0, label: `Syncing ${folder}...` }); + }, }); - } catch (spawnErr) { - const msg = spawnErr instanceof Error ? spawnErr.message : String(spawnErr); - throw new PermanentError(`Failed to start mbsync: ${msg}`); - } - // Periodically count downloaded emails and update progress - let emailCount = 0; - let counting = true; - const countLoop = (async () => { - while (counting) { - await new Promise((r) => setTimeout(r, 3000)); - if (!counting) break; + console.log( + `[gmail-sync] Incremental sync 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()); + + ctx.meta.syncResult = result; + await ctx.updateProgress({ current: 1, total: 1, label: `Done — ${result.saved} new emails` }); + } finally { + db.close(); + } + } else { + // ── Full sync: mbsync ── + const email = ctx.meta.email as string; + const appPassword = ctx.meta.appPassword as string; + const maildirPath = getMaildirPath(ctx.job.userId); + + await mkdir(maildirPath, { recursive: true }); + + const configPath = join(maildirPath, '.mbsyncrc'); + const config = buildMbsyncConfig(email, appPassword, maildirPath); + await writeFile(configPath, config, { mode: 0o600 }); + + await ctx.updateProgress({ current: 0, total: 0, label: 'Running mbsync...' }); + + try { + let proc: ReturnType; + try { + proc = Bun.spawn(['mbsync', '-c', configPath, '-a'], { + stdout: 'pipe', + stderr: 'pipe', + }); + } catch (spawnErr) { + const msg = spawnErr instanceof Error ? spawnErr.message : String(spawnErr); + throw new PermanentError(`Failed to start mbsync: ${msg}`); + } + + let emailCount = 0; + let counting = true; + const countLoop = (async () => { + while (counting) { + await new Promise((r) => setTimeout(r, 3000)); + if (!counting) break; + emailCount = await countMaildirFiles(maildirPath); + ctx.updateProgress({ + current: emailCount, + total: 0, + label: `Downloading — ${emailCount.toLocaleString()} emails`, + }); + } + })(); + + let stderrBuf = ''; + const reader = (proc.stderr as ReadableStream).getReader(); + const decoder = new TextDecoder(); + const readLoop = (async () => { + while (true) { + const { done, value } = await reader.read(); + if (done) break; + const chunk = decoder.decode(value, { stream: true }); + stderrBuf += chunk; + const lines = chunk.split('\n'); + for (const line of lines) { + const trimmed = line.trim(); + if (trimmed) console.log(`[gmail-sync] mbsync: ${trimmed}`); + } + } + })(); + + const exitCode = await proc.exited; + counting = false; + await readLoop; + await countLoop; + + if (exitCode !== 0) { + const isOverquota = stderrBuf.includes('OVERQUOTA'); + const isAuthFail = + stderrBuf.includes('AUTHENTICATIONFAILED') || stderrBuf.includes('Invalid credentials'); + + if (isOverquota) { + const emailCount = await countMaildirFiles(maildirPath); + console.log(`[gmail-sync] Gmail OVERQUOTA — proceeding to import ${emailCount} downloaded emails`); + ctx.meta.gmailSyncPartial = true; + ctx.meta.gmailSyncEmailCount = emailCount; + } else { + console.error(`[gmail-sync] mbsync failed`); + if (isAuthFail) { + const emailCount = await countMaildirFiles(maildirPath); + 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)}`); + } + } else { emailCount = await countMaildirFiles(maildirPath); - ctx.updateProgress({ + console.log(`[gmail-sync] mbsync completed successfully — ${emailCount} emails`); + await ctx.updateProgress({ current: emailCount, - total: 0, - label: `Downloading — ${emailCount.toLocaleString()} emails`, + total: emailCount, + label: `Download complete — ${emailCount.toLocaleString()} emails`, }); } - })(); - - // Stream stderr for logging - let stderrBuf = ''; - const reader = (proc.stderr as ReadableStream).getReader(); - const decoder = new TextDecoder(); - const readLoop = (async () => { - while (true) { - const { done, value } = await reader.read(); - if (done) break; - const chunk = decoder.decode(value, { stream: true }); - stderrBuf += chunk; - const lines = chunk.split('\n'); - for (const line of lines) { - const trimmed = line.trim(); - if (trimmed) console.log(`[gmail-sync] mbsync: ${trimmed}`); - } - } - })(); - - const exitCode = await proc.exited; - counting = false; - await readLoop; - await countLoop; - - if (exitCode !== 0) { - const isOverquota = stderrBuf.includes('OVERQUOTA'); - const isAuthFail = stderrBuf.includes('AUTHENTICATIONFAILED') || stderrBuf.includes('Invalid credentials'); - - if (isOverquota) { - // Gmail throttled us — don't retry, just import what we have - const emailCount = await countMaildirFiles(maildirPath); - console.log(`[gmail-sync] Gmail OVERQUOTA — proceeding to import ${emailCount} downloaded emails`); - ctx.meta.gmailSyncPartial = true; - ctx.meta.gmailSyncEmailCount = emailCount; - // Fall through to import step - } else { - console.error(`[gmail-sync] mbsync failed`); - if (isAuthFail) { - const emailCount = await countMaildirFiles(maildirPath); - 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)}`); - } - } else { - emailCount = await countMaildirFiles(maildirPath); - console.log(`[gmail-sync] mbsync completed successfully — ${emailCount} emails`); - await ctx.updateProgress({ - current: emailCount, - total: emailCount, - label: `Download complete — ${emailCount.toLocaleString()} emails`, - }); + } finally { + await unlink(configPath).catch(() => {}); } - } finally { - // Always clean up config (contains password) - await unlink(configPath).catch(() => {}); } }, }, { name: 'Import to database', run: async (ctx) => { + // Incremental sync already imported in the previous step + if (ctx.meta.isIncremental) { + const result = ctx.meta.syncResult as IncrementalSyncResult | undefined; + + // Auto-add /email to dock + if (result && result.saved > 0) { + try { + const dbUser = await getUserByEmail(ctx.job.userId); + if (dbUser) { + const paths = await getDockPaths(dbUser.id); + if (!paths) { + 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 + } + } + + await ctx.updateProgress({ current: 1, total: 1, label: `Done — ${result?.saved ?? 0} new emails` }); + return; + } + + // Full sync: import from Maildir const emailAccount = ctx.meta.email as string; const maildirPath = getMaildirPath(ctx.job.userId); @@ -367,12 +737,19 @@ const gmailSyncHandler: JobHandler = { const db = openEmailDb(ctx.job.userId); try { - const result = await importMaildir(maildirPath, emailAccount, db, (saved, skipped) => { - ctx.updateProgress({ - current: saved + skipped, - total: 0, - label: `Saved ${saved.toLocaleString()}, skipped ${skipped.toLocaleString()}`, - }); + const lastSyncAt = getSyncMeta(db, 'last_sync_at'); + const result = await importMaildir({ + maildirPath, + emailAccount, + db, + lastSyncAt, + onProgress: (saved, skipped) => { + ctx.updateProgress({ + current: saved + skipped, + total: 0, + label: `Saved ${saved.toLocaleString()}, skipped ${skipped.toLocaleString()}`, + }); + }, }); console.log( @@ -389,7 +766,6 @@ const gmailSyncHandler: JobHandler = { 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')) { diff --git a/src/servers/queue/handlers/index.ts b/src/servers/queue/handlers/index.ts index 70376d89..fae81d29 100644 --- a/src/servers/queue/handlers/index.ts +++ b/src/servers/queue/handlers/index.ts @@ -1 +1,2 @@ import './gmail-sync'; +import './email-sync'; diff --git a/src/servers/sidecar/email-cron.ts b/src/servers/sidecar/email-cron.ts new file mode 100644 index 00000000..1e6dd86f --- /dev/null +++ b/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 | 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 | 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 = { user: account.email }; + if (account.authType === 'oauth') { + const integration = await getUserIntegration(account.userId, 'google'); + const config = integration?.config as Record | 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; + 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; + } +} diff --git a/src/servers/sidecar/index.ts b/src/servers/sidecar/index.ts index 44d77a62..b3e32eba 100644 --- a/src/servers/sidecar/index.ts +++ b/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>(); @@ -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); diff --git a/src/servers/sidecar/queue-runner.ts b/src/servers/sidecar/queue-runner.ts index c0b34755..d2811387 100644 --- a/src/servers/sidecar/queue-runner.ts +++ b/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(); 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 = { ...(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); } }