inline email resync instead of job queue, refresh list on completion
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -73,18 +73,28 @@ export const EmailList = () => {
|
|||||||
setPage(1);
|
setPage(1);
|
||||||
};
|
};
|
||||||
|
|
||||||
|
const [syncing, setSyncing] = useState(false);
|
||||||
|
|
||||||
const handleSync = async () => {
|
const handleSync = async () => {
|
||||||
if (!syncableAccount) return;
|
if (!syncableAccount) return;
|
||||||
|
setSyncing(true);
|
||||||
try {
|
try {
|
||||||
await client.post(`/email/accounts/${syncableAccount.id}/sync`, {});
|
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');
|
toast.success('Sync started');
|
||||||
|
}
|
||||||
refetchAccounts();
|
refetchAccounts();
|
||||||
} catch (err) {
|
} 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);
|
const prevSyncing = useRef(false);
|
||||||
useEffect(() => {
|
useEffect(() => {
|
||||||
if (isSyncing) {
|
if (isSyncing) {
|
||||||
@@ -149,8 +159,8 @@ export const EmailList = () => {
|
|||||||
) : syncableAccount ? (
|
) : syncableAccount ? (
|
||||||
<div className="flex flex-col items-center gap-2">
|
<div className="flex flex-col items-center gap-2">
|
||||||
<span>No emails synced yet</span>
|
<span>No emails synced yet</span>
|
||||||
<Button variant="outline" size="sm" onClick={handleSync} disabled={isSyncing}>
|
<Button variant="outline" size="sm" onClick={handleSync} disabled={syncing || isSyncing}>
|
||||||
{isSyncing ? <Loader2 className="mr-2 h-3.5 w-3.5 animate-spin" /> : <RefreshCw className="mr-2 h-3.5 w-3.5" />}
|
{syncing || isSyncing ? <Loader2 className="mr-2 h-3.5 w-3.5 animate-spin" /> : <RefreshCw className="mr-2 h-3.5 w-3.5" />}
|
||||||
Sync Now
|
Sync Now
|
||||||
</Button>
|
</Button>
|
||||||
</div>
|
</div>
|
||||||
@@ -189,7 +199,7 @@ export const EmailList = () => {
|
|||||||
})}
|
})}
|
||||||
</div>
|
</div>
|
||||||
<span className="text-xs opacity-50">{total}</span>
|
<span className="text-xs opacity-50">{total}</span>
|
||||||
{isSyncing ? (
|
{syncing || isSyncing ? (
|
||||||
<Loader2 className="h-3.5 w-3.5 animate-spin shrink-0 opacity-50" title="Syncing..." />
|
<Loader2 className="h-3.5 w-3.5 animate-spin shrink-0 opacity-50" title="Syncing..." />
|
||||||
) : syncableAccount ? (
|
) : syncableAccount ? (
|
||||||
<button
|
<button
|
||||||
|
|||||||
+7
-3
@@ -158,12 +158,16 @@ export const EmailAccounts = () => {
|
|||||||
const handleSync = async (id: number) => {
|
const handleSync = async (id: number) => {
|
||||||
setSyncing(id);
|
setSyncing(id);
|
||||||
try {
|
try {
|
||||||
await client.post(`/email/accounts/${id}/sync`, {});
|
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');
|
toast.success('Sync started');
|
||||||
// Update local state immediately
|
|
||||||
setAccounts((prev) => prev.map((a) => (a.id === id ? { ...a, status: 'queued' } : a)));
|
setAccounts((prev) => prev.map((a) => (a.id === id ? { ...a, status: 'queued' } : a)));
|
||||||
|
}
|
||||||
|
fetchAccounts();
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
toast.error(err instanceof Error ? err.message : 'Failed to start sync');
|
toast.error(err instanceof Error ? err.message : 'Failed to sync');
|
||||||
} finally {
|
} finally {
|
||||||
setSyncing(null);
|
setSyncing(null);
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -10,6 +10,8 @@ import {
|
|||||||
} from 'officerdb';
|
} from 'officerdb';
|
||||||
import { validateImapConnection } from './imap-validate';
|
import { validateImapConnection } from './imap-validate';
|
||||||
import { enqueueJob, listAllJobs } from '../../queue/init';
|
import { enqueueJob, listAllJobs } from '../../queue/init';
|
||||||
|
import { openEmailDb, getSyncMeta } from './email-db';
|
||||||
|
import { performResync } from './resync';
|
||||||
|
|
||||||
type CreateAccountBody = {
|
type CreateAccountBody = {
|
||||||
provider: string;
|
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 === 'queued') throw BAD_REQUEST('Sync is already queued');
|
||||||
if (account.status === 'syncing') throw BAD_REQUEST('Account is already syncing');
|
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<string, unknown>;
|
||||||
|
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(
|
const authResult = await resolveAuth(
|
||||||
user.id,
|
user.id,
|
||||||
account.authType,
|
account.authType,
|
||||||
@@ -145,7 +174,6 @@ accountsRouter.post('/:id/sync', async (ctx) => {
|
|||||||
);
|
);
|
||||||
if (!authResult.ok) throw BAD_REQUEST(authResult.error);
|
if (!authResult.ok) throw BAD_REQUEST(authResult.error);
|
||||||
|
|
||||||
// Set status immediately so the UI reflects the queued state
|
|
||||||
await updateEmailAccountStatus(id, 'queued');
|
await updateEmailAccountStatus(id, 'queued');
|
||||||
|
|
||||||
const jobType = account.provider === 'gmail' ? 'gmail-sync' : 'email-sync';
|
const jobType = account.provider === 'gmail' ? 'gmail-sync' : 'email-sync';
|
||||||
|
|||||||
@@ -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<GmailCredentials> {
|
||||||
|
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<string, unknown>) ?? {};
|
||||||
|
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<ResyncResult> {
|
||||||
|
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<string, unknown>;
|
||||||
|
};
|
||||||
|
|
||||||
|
type FolderInfo = {
|
||||||
|
specialUse?: string;
|
||||||
|
path: string;
|
||||||
|
flags: Set<string>;
|
||||||
|
status?: { uidNext?: number; uidValidity?: number };
|
||||||
|
};
|
||||||
|
|
||||||
|
const SPECIAL_USE_LABEL_MAP: Record<string, string> = {
|
||||||
|
'\\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<string, unknown> | 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<ResyncResult> {
|
||||||
|
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<string, unknown>;
|
||||||
|
|
||||||
|
const db = openEmailDb(userEmail);
|
||||||
|
const existingIds = new Set<string>();
|
||||||
|
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<string, unknown> = { 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<string, unknown>);
|
||||||
|
|
||||||
|
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<ResyncResult> {
|
||||||
|
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<string, unknown>,
|
||||||
|
},
|
||||||
|
userEmail,
|
||||||
|
);
|
||||||
|
}
|
||||||
@@ -17,7 +17,7 @@ import { refreshGoogleAccessToken } from '@@/api/integrations/google-auth';
|
|||||||
|
|
||||||
// ── Credentials ──
|
// ── Credentials ──
|
||||||
|
|
||||||
type GmailCredentials = {
|
export type GmailCredentials = {
|
||||||
email: string;
|
email: string;
|
||||||
userId: number;
|
userId: number;
|
||||||
appPassword?: string;
|
appPassword?: string;
|
||||||
@@ -26,7 +26,7 @@ type GmailCredentials = {
|
|||||||
expiresAt?: number;
|
expiresAt?: number;
|
||||||
};
|
};
|
||||||
|
|
||||||
async function loadGmailCredentials(userEmail: string): Promise<GmailCredentials> {
|
export async function loadGmailCredentials(userEmail: string): Promise<GmailCredentials> {
|
||||||
const dbUser = await getUserByEmail(userEmail);
|
const dbUser = await getUserByEmail(userEmail);
|
||||||
if (!dbUser) throw new PermanentError('User not found');
|
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';
|
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<string, string>): Promise<Response> {
|
async function gmailApiFetch(accessToken: string, path: string, params?: Record<string, string>): Promise<Response> {
|
||||||
const url = new URL(`${GMAIL_API}${path}`);
|
const url = new URL(`${GMAIL_API}${path}`);
|
||||||
@@ -219,13 +219,13 @@ async function fetchRawMessage(accessToken: string, messageId: string): Promise<
|
|||||||
|
|
||||||
// ── Gmail API sync (for resyncs) ──
|
// ── Gmail API sync (for resyncs) ──
|
||||||
|
|
||||||
type GmailApiSyncParams = {
|
export type GmailApiSyncParams = {
|
||||||
creds: GmailCredentials;
|
creds: GmailCredentials;
|
||||||
db: Database;
|
db: Database;
|
||||||
onProgress?: (saved: number, total: number) => void;
|
onProgress?: (saved: number, total: number) => void;
|
||||||
};
|
};
|
||||||
|
|
||||||
async function gmailApiSync({ creds, db, onProgress }: GmailApiSyncParams): Promise<GmailApiResult> {
|
export async function gmailApiSync({ creds, db, onProgress }: GmailApiSyncParams): Promise<GmailApiResult> {
|
||||||
if (!creds.accessToken) throw new PermanentError('OAuth access token required for Gmail API sync');
|
if (!creds.accessToken) throw new PermanentError('OAuth access token required for Gmail API sync');
|
||||||
|
|
||||||
let saved = 0;
|
let saved = 0;
|
||||||
|
|||||||
@@ -1,74 +1,28 @@
|
|||||||
import type { Job, EnqueueParams } from '../../queue/types';
|
import { getAllSyncedAccounts, getUserById } from 'officerdb';
|
||||||
import { getAllSyncedAccounts, getUserById, getUserIntegration } from 'officerdb';
|
import { performResync } from '../../api/email/resync';
|
||||||
|
|
||||||
const INTERVAL_MS = 10 * 60 * 1000; // 10 minutes
|
const INTERVAL_MS = 10 * 60 * 1000; // 10 minutes
|
||||||
|
|
||||||
let timer: ReturnType<typeof setInterval> | null = null;
|
let timer: ReturnType<typeof setInterval> | null = null;
|
||||||
let enqueueFn: ((params: EnqueueParams) => Promise<Job>) | null = null;
|
|
||||||
let listJobsFn: (() => Promise<Job[]>) | null = null;
|
|
||||||
|
|
||||||
async function tick() {
|
async function tick() {
|
||||||
if (!enqueueFn || !listJobsFn) return;
|
|
||||||
|
|
||||||
try {
|
try {
|
||||||
const accounts = await getAllSyncedAccounts();
|
const accounts = await getAllSyncedAccounts();
|
||||||
if (accounts.length === 0) return;
|
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<string, unknown> | undefined)?.emailAccountId),
|
|
||||||
);
|
|
||||||
|
|
||||||
for (const account of accounts) {
|
for (const account of accounts) {
|
||||||
if (activeEmailSyncIds.has(account.id)) continue;
|
|
||||||
|
|
||||||
const user = await getUserById(account.userId);
|
const user = await getUserById(account.userId);
|
||||||
if (!user) continue;
|
if (!user) continue;
|
||||||
|
|
||||||
// Resolve IMAP auth
|
|
||||||
const imapAuth: Record<string, unknown> = { user: account.email };
|
|
||||||
if (account.authType === 'oauth') {
|
|
||||||
const integration = await getUserIntegration(account.userId, 'google');
|
|
||||||
const config = integration?.config as Record<string, unknown> | undefined;
|
|
||||||
const accessToken = config?.accessToken as string | undefined;
|
|
||||||
if (!accessToken) {
|
|
||||||
console.log(`[email-cron] Skipping ${account.email}: no OAuth access token`);
|
|
||||||
continue;
|
|
||||||
}
|
|
||||||
imapAuth.accessToken = accessToken;
|
|
||||||
} else {
|
|
||||||
const creds = account.credentials as Record<string, unknown>;
|
|
||||||
imapAuth.pass = creds.password;
|
|
||||||
}
|
|
||||||
|
|
||||||
try {
|
try {
|
||||||
await enqueueFn({
|
console.log(`[email-cron] Resyncing ${account.email}`);
|
||||||
lane: 'email',
|
const result = await performResync({ accountId: account.id, userEmail: user.email, userId: user.id });
|
||||||
type: 'email-sync',
|
if (result.saved > 0) {
|
||||||
userId: user.email,
|
console.log(`[email-cron] ${account.email}: ${result.saved} new emails`);
|
||||||
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) {
|
} catch (err) {
|
||||||
console.error(
|
console.error(
|
||||||
`[email-cron] Failed to enqueue sync for ${account.email}:`,
|
`[email-cron] Failed to resync ${account.email}:`,
|
||||||
err instanceof Error ? err.message : err,
|
err instanceof Error ? err.message : err,
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
@@ -78,18 +32,10 @@ async function tick() {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
type EmailCronDeps = {
|
export function initEmailCron() {
|
||||||
enqueue: (params: EnqueueParams) => Promise<Job>;
|
|
||||||
listJobs: () => Promise<Job[]>;
|
|
||||||
};
|
|
||||||
|
|
||||||
export function initEmailCron(deps: EmailCronDeps) {
|
|
||||||
if (timer) return;
|
if (timer) return;
|
||||||
enqueueFn = deps.enqueue;
|
|
||||||
listJobsFn = deps.listJobs;
|
|
||||||
console.log(`[email-cron] Starting email sync cron (every ${INTERVAL_MS / 60_000} min)`);
|
console.log(`[email-cron] Starting email sync cron (every ${INTERVAL_MS / 60_000} min)`);
|
||||||
timer = setInterval(tick, INTERVAL_MS);
|
timer = setInterval(tick, INTERVAL_MS);
|
||||||
// Run first tick after a short delay
|
|
||||||
setTimeout(tick, 30_000);
|
setTimeout(tick, 30_000);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -83,7 +83,7 @@ const connection = createSidecarConnector({
|
|||||||
},
|
},
|
||||||
onConnected() {
|
onConnected() {
|
||||||
// Start email cron once connected (so queue commands can reach API server)
|
// Start email cron once connected (so queue commands can reach API server)
|
||||||
// initEmailCron({ enqueue: enqueueViaWs, listJobs: listJobsViaWs }); // TODO: re-enable after testing
|
initEmailCron();
|
||||||
},
|
},
|
||||||
});
|
});
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user