diff --git a/src/apps/officer-web/Screens/Dashboard/Email/EmailList.tsx b/src/apps/officer-web/Screens/Dashboard/Email/EmailList.tsx
index 99e7fa47..3b92cede 100644
--- a/src/apps/officer-web/Screens/Dashboard/Email/EmailList.tsx
+++ b/src/apps/officer-web/Screens/Dashboard/Email/EmailList.tsx
@@ -73,18 +73,28 @@ export const EmailList = () => {
setPage(1);
};
+ const [syncing, setSyncing] = useState(false);
+
const handleSync = async () => {
if (!syncableAccount) return;
+ setSyncing(true);
try {
- await client.post(`/email/accounts/${syncableAccount.id}/sync`, {});
- toast.success('Sync started');
+ const result = await client.post<{ ok: boolean; saved?: number }>(`/email/accounts/${syncableAccount.id}/sync`, {});
+ if (result.saved !== undefined) {
+ toast.success(result.saved > 0 ? `${result.saved} new emails` : 'No new emails');
+ queryClient.invalidateQueries({ queryKey: ['email-messages'] });
+ } else {
+ toast.success('Sync started');
+ }
refetchAccounts();
} catch (err) {
- toast.error(err instanceof Error ? err.message : 'Failed to start sync');
+ toast.error(err instanceof Error ? err.message : 'Failed to sync');
+ } finally {
+ setSyncing(false);
}
};
- // Poll accounts while syncing, refresh email list when done
+ // Poll accounts while first sync is running (job-based), refresh email list when done
const prevSyncing = useRef(false);
useEffect(() => {
if (isSyncing) {
@@ -149,8 +159,8 @@ export const EmailList = () => {
) : syncableAccount ? (
No emails synced yet
-
@@ -189,7 +199,7 @@ export const EmailList = () => {
})}
{total}
- {isSyncing ? (
+ {syncing || isSyncing ? (
) : syncableAccount ? (
{
const handleSync = async (id: number) => {
setSyncing(id);
try {
- await client.post(`/email/accounts/${id}/sync`, {});
- toast.success('Sync started');
- // Update local state immediately
- setAccounts((prev) => prev.map((a) => (a.id === id ? { ...a, status: 'queued' } : a)));
+ const result = await client.post<{ ok: boolean; saved?: number }>(`/email/accounts/${id}/sync`, {});
+ if (result.saved !== undefined) {
+ toast.success(result.saved > 0 ? `${result.saved} new emails` : 'No new emails');
+ } else {
+ toast.success('Sync started');
+ setAccounts((prev) => prev.map((a) => (a.id === id ? { ...a, status: 'queued' } : a)));
+ }
+ fetchAccounts();
} catch (err) {
- toast.error(err instanceof Error ? err.message : 'Failed to start sync');
+ toast.error(err instanceof Error ? err.message : 'Failed to sync');
} finally {
setSyncing(null);
}
diff --git a/src/servers/api/email/accounts.ts b/src/servers/api/email/accounts.ts
index 652232fa..eb6a1f54 100644
--- a/src/servers/api/email/accounts.ts
+++ b/src/servers/api/email/accounts.ts
@@ -10,6 +10,8 @@ import {
} from 'officerdb';
import { validateImapConnection } from './imap-validate';
import { enqueueJob, listAllJobs } from '../../queue/init';
+import { openEmailDb, getSyncMeta } from './email-db';
+import { performResync } from './resync';
type CreateAccountBody = {
provider: string;
@@ -136,7 +138,34 @@ accountsRouter.post('/:id/sync', async (ctx) => {
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
+ // Determine if this is a first sync or resync
+ let isFirstSync = true;
+ if (account.provider === 'gmail') {
+ const db = openEmailDb(user.email);
+ try {
+ isFirstSync = !getSyncMeta(db, 'last_sync_at');
+ } finally {
+ db.close();
+ }
+ } else {
+ const syncMeta = (account.syncMeta ?? {}) as Record;
+ isFirstSync = !syncMeta.last_sync_at;
+ }
+
+ // Resync: call directly, no job queue
+ if (!isFirstSync) {
+ await updateEmailAccountStatus(id, 'syncing');
+ try {
+ const result = await performResync({ accountId: id, userEmail: user.email, userId: user.id });
+ await updateEmailAccountStatus(id, 'synced');
+ return ctx.json({ ok: true, saved: result.saved });
+ } catch (err) {
+ await updateEmailAccountStatus(id, 'synced').catch(() => {});
+ throw err;
+ }
+ }
+
+ // First sync: enqueue a job
const authResult = await resolveAuth(
user.id,
account.authType,
@@ -145,7 +174,6 @@ accountsRouter.post('/:id/sync', async (ctx) => {
);
if (!authResult.ok) throw BAD_REQUEST(authResult.error);
- // Set status immediately so the UI reflects the queued state
await updateEmailAccountStatus(id, 'queued');
const jobType = account.provider === 'gmail' ? 'gmail-sync' : 'email-sync';
diff --git a/src/servers/api/email/resync.ts b/src/servers/api/email/resync.ts
new file mode 100644
index 00000000..7333fb8e
--- /dev/null
+++ b/src/servers/api/email/resync.ts
@@ -0,0 +1,296 @@
+import { createHash } from 'node:crypto';
+import {
+ getUserIntegration,
+ upsertUserIntegration,
+ getServerIntegration,
+ getEmailAccount,
+ updateEmailAccountSyncMeta,
+} from 'officerdb';
+import { refreshGoogleAccessToken } from '@@/api/integrations/google-auth';
+import { openEmailDb, getSyncMeta, setSyncMeta, upsertFromRawEml } from './email-db';
+import {
+ type GmailCredentials,
+ loadGmailCredentials,
+ gmailApiSync,
+} from '../../queue/handlers/gmail-sync';
+
+export type ResyncResult = { saved: number; skipped: number; errors: number };
+
+// ── Gmail resync (REST API, history-based) ──
+
+async function refreshCredentials(creds: GmailCredentials): Promise {
+ if (!creds.accessToken || !creds.refreshToken) return creds;
+
+ const tokenExpired = !creds.expiresAt || creds.expiresAt < Date.now() + 60_000;
+ if (!tokenExpired) return creds;
+
+ console.log('[resync] Refreshing OAuth token');
+ const refreshed = await refreshGoogleAccessToken(creds.refreshToken);
+
+ 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 },
+ });
+
+ return { ...creds, accessToken: refreshed.accessToken, expiresAt: refreshed.expiresAt };
+}
+
+async function gmailResync(userEmail: string): Promise {
+ let creds = await loadGmailCredentials(userEmail);
+ if (!creds.accessToken) {
+ throw new Error('OAuth not configured — connect Google in Settings → Integrations for resyncs');
+ }
+
+ creds = await refreshCredentials(creds);
+
+ const db = openEmailDb(userEmail);
+ try {
+ const result = await gmailApiSync({ creds, db });
+
+ setSyncMeta(db, 'last_sync_date', new Date().toISOString().split('T')[0]!);
+ setSyncMeta(db, 'last_sync_at', new Date().toISOString());
+
+ console.log(`[resync] Gmail done: saved ${result.saved}, skipped ${result.skipped}, errors ${result.errors}`);
+ return result;
+ } finally {
+ db.close();
+ }
+}
+
+// ── Generic IMAP resync ──
+
+type ImapAccountInfo = {
+ id: number;
+ userId: number;
+ email: string;
+ imapHost: string;
+ imapPort: number;
+ imapSecure: boolean;
+ provider: string;
+ authType: string;
+ credentials: Record;
+};
+
+type FolderInfo = {
+ specialUse?: string;
+ path: string;
+ flags: Set;
+ status?: { uidNext?: number; uidValidity?: number };
+};
+
+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']);
+
+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();
+}
+
+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);
+}
+
+async function resolveImapAuth(
+ account: ImapAccountInfo,
+): Promise<{ user: string; pass?: string; accessToken?: string }> {
+ if (account.authType !== 'oauth') {
+ return { user: account.email, pass: account.credentials.password as string };
+ }
+
+ const userGoogle = await getUserIntegration(account.userId, 'google');
+ const config = userGoogle?.config as Record | undefined;
+ if (!config?.refreshToken) throw new Error('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: account.email, accessToken: config.accessToken as string };
+ }
+
+ const refreshed = await refreshGoogleAccessToken(config.refreshToken as string);
+ const serverGoogle = await getServerIntegration('google');
+ await upsertUserIntegration({
+ userId: account.userId,
+ provider: 'google',
+ serverIntegrationId: serverGoogle?.id,
+ config: { ...config, accessToken: refreshed.accessToken, expiresAt: refreshed.expiresAt },
+ });
+
+ return { user: account.email, accessToken: refreshed.accessToken };
+}
+
+async function imapResync(account: ImapAccountInfo, userEmail: string): Promise {
+ const { ImapFlow } = await import('imapflow');
+
+ const imapAuth = await resolveImapAuth(account);
+
+ const freshAccount = await getEmailAccount(account.id);
+ if (!freshAccount) throw new Error(`Email account ${account.id} not found`);
+ const syncMeta = (freshAccount.syncMeta ?? {}) as Record;
+
+ const db = openEmailDb(userEmail);
+ 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);
+
+ let saved = 0;
+ let skipped = 0;
+ let errors = 0;
+
+ const client = new ImapFlow({
+ host: account.imapHost,
+ port: account.imapPort,
+ secure: account.imapSecure,
+ auth: imapAuth,
+ logger: false,
+ socketTimeout: 5 * 60 * 1000,
+ });
+
+ try {
+ await client.connect();
+ console.log('[resync] IMAP connected');
+
+ const folders = (await client.list({ statusQuery: { uidNext: true, uidValidity: true } })) as FolderInfo[];
+
+ 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;
+ const lastUid = storedUidValidity && uidValidity === storedUidValidity && storedLastUid ? parseInt(storedLastUid, 10) : 0;
+
+ if (lastUid > 0 && uidNext <= lastUid + 1) continue;
+
+ let lock;
+ try {
+ lock = await client.getMailboxLock(folder.path);
+ } catch {
+ continue;
+ }
+
+ try {
+ const mailbox = client.mailbox;
+ if (!mailbox) continue;
+
+ const mbUidValidity = String(mailbox.uidValidity);
+ const effectiveLastUid = storedUidValidity === mbUidValidity ? 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; }
+
+ try {
+ upsertFromRawEml({ db, id, raw, integration: account.provider, emailAccount: account.email, labels: [label] });
+ existingIds.add(id);
+ saved++;
+ } catch {
+ errors++;
+ }
+ }
+ } catch (fetchErr) {
+ const errMsg = fetchErr instanceof Error ? fetchErr.message : String(fetchErr);
+ if (!errMsg.includes('Nothing to fetch')) {
+ console.log(`[resync] Fetch error in ${folder.path}: ${errMsg}`);
+ errors++;
+ }
+ }
+
+ syncMeta[uidValidityKey] = mbUidValidity;
+ if (maxUid > effectiveLastUid) {
+ syncMeta[lastUidKey] = String(maxUid);
+ }
+ } finally {
+ lock.release();
+ }
+ }
+
+ await client.logout().catch(() => {});
+ } catch (err) {
+ await client.logout().catch(() => {});
+ throw err;
+ } finally {
+ db.close();
+ }
+
+ syncMeta.last_sync_at = new Date().toISOString();
+ await updateEmailAccountSyncMeta(account.id, syncMeta as Record);
+
+ console.log(`[resync] IMAP done: saved ${saved}, skipped ${skipped}, errors ${errors}`);
+ return { saved, skipped, errors };
+}
+
+// ── Public API ──
+
+type ResyncParams = {
+ accountId: number;
+ userEmail: string;
+ userId: number;
+};
+
+export async function performResync({ accountId, userEmail, userId }: ResyncParams): Promise {
+ const account = await getEmailAccount(accountId);
+ if (!account || account.userId !== userId) throw new Error('Account not found');
+
+ if (account.provider === 'gmail') {
+ return gmailResync(userEmail);
+ }
+
+ return imapResync(
+ {
+ 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 as Record,
+ },
+ userEmail,
+ );
+}
diff --git a/src/servers/queue/handlers/gmail-sync.ts b/src/servers/queue/handlers/gmail-sync.ts
index 74cde140..9938c0d4 100644
--- a/src/servers/queue/handlers/gmail-sync.ts
+++ b/src/servers/queue/handlers/gmail-sync.ts
@@ -17,7 +17,7 @@ import { refreshGoogleAccessToken } from '@@/api/integrations/google-auth';
// ── Credentials ──
-type GmailCredentials = {
+export type GmailCredentials = {
email: string;
userId: number;
appPassword?: string;
@@ -26,7 +26,7 @@ type GmailCredentials = {
expiresAt?: number;
};
-async function loadGmailCredentials(userEmail: string): Promise {
+export async function loadGmailCredentials(userEmail: string): Promise {
const dbUser = await getUserByEmail(userEmail);
if (!dbUser) throw new PermanentError('User not found');
@@ -102,7 +102,7 @@ function shouldSkipMessage(labelIds: string[]): boolean {
const GMAIL_API = 'https://gmail.googleapis.com/gmail/v1/users/me';
-type GmailApiResult = { saved: number; skipped: number; errors: number };
+export type GmailApiResult = { saved: number; skipped: number; errors: number };
async function gmailApiFetch(accessToken: string, path: string, params?: Record): Promise {
const url = new URL(`${GMAIL_API}${path}`);
@@ -219,13 +219,13 @@ async function fetchRawMessage(accessToken: string, messageId: string): Promise<
// ── Gmail API sync (for resyncs) ──
-type GmailApiSyncParams = {
+export type GmailApiSyncParams = {
creds: GmailCredentials;
db: Database;
onProgress?: (saved: number, total: number) => void;
};
-async function gmailApiSync({ creds, db, onProgress }: GmailApiSyncParams): Promise {
+export async function gmailApiSync({ creds, db, onProgress }: GmailApiSyncParams): Promise {
if (!creds.accessToken) throw new PermanentError('OAuth access token required for Gmail API sync');
let saved = 0;
diff --git a/src/servers/sidecar/email/email-cron.ts b/src/servers/sidecar/email/email-cron.ts
index 2d17ddb8..c2f4463d 100644
--- a/src/servers/sidecar/email/email-cron.ts
+++ b/src/servers/sidecar/email/email-cron.ts
@@ -1,74 +1,28 @@
-import type { Job, EnqueueParams } from '../../queue/types';
-import { getAllSyncedAccounts, getUserById, getUserIntegration } from 'officerdb';
+import { getAllSyncedAccounts, getUserById } from 'officerdb';
+import { performResync } from '../../api/email/resync';
const INTERVAL_MS = 10 * 60 * 1000; // 10 minutes
let timer: ReturnType | null = null;
-let enqueueFn: ((params: EnqueueParams) => Promise) | null = null;
-let listJobsFn: (() => Promise) | null = null;
async function tick() {
- if (!enqueueFn || !listJobsFn) return;
-
try {
const accounts = await getAllSyncedAccounts();
if (accounts.length === 0) return;
- const allJobs = await listJobsFn();
- 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 enqueueFn({
- 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}`);
+ console.log(`[email-cron] Resyncing ${account.email}`);
+ const result = await performResync({ accountId: account.id, userEmail: user.email, userId: user.id });
+ if (result.saved > 0) {
+ console.log(`[email-cron] ${account.email}: ${result.saved} new emails`);
+ }
} catch (err) {
console.error(
- `[email-cron] Failed to enqueue sync for ${account.email}:`,
+ `[email-cron] Failed to resync ${account.email}:`,
err instanceof Error ? err.message : err,
);
}
@@ -78,18 +32,10 @@ async function tick() {
}
}
-type EmailCronDeps = {
- enqueue: (params: EnqueueParams) => Promise;
- listJobs: () => Promise;
-};
-
-export function initEmailCron(deps: EmailCronDeps) {
+export function initEmailCron() {
if (timer) return;
- enqueueFn = deps.enqueue;
- listJobsFn = deps.listJobs;
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
setTimeout(tick, 30_000);
}
diff --git a/src/servers/sidecar/email/index.ts b/src/servers/sidecar/email/index.ts
index ecde2cb3..e461ddcb 100644
--- a/src/servers/sidecar/email/index.ts
+++ b/src/servers/sidecar/email/index.ts
@@ -83,7 +83,7 @@ const connection = createSidecarConnector({
},
onConnected() {
// Start email cron once connected (so queue commands can reach API server)
- // initEmailCron({ enqueue: enqueueViaWs, listJobs: listJobsViaWs }); // TODO: re-enable after testing
+ initEmailCron();
},
});