paul/picpeak
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

fix(email): log all received mail, not just unseen (90-day lookback + dedup)

Browse Source 
Cause of 'not all received emails listed': the poller fetched {seen:false}
only, so any message already read in another client was never pulled or logged.

Now the poller scans a LOOKBACK_DAYS (90) window regardless of \Seen via a
cheap envelope-only pass, dedups by message-id against received_emails, and only
downloads + processes (fetchOne source) messages not yet logged — so the
Received tab is complete while each poll stays light. Marks processed messages
seen; re-checks the parsed message-id before insert.
This commit is contained in:
Luca
2026-06-12 16:06:40 +02:00
parent cee692c10a
commit c36797db2d
1 changed files with 45 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/services/emailIntakeService.js
+45 -12
View File
@@ -24,6 +24,10 @@ let polling = false;
// SMTP port). Without these, ImapFlow waits indefinitely and the HTTP request
// dies at the proxy as a 502 with no useful message.
const IMAP_TIMEOUTS = { connectionTimeout: 10000, greetingTimeout: 10000, socketTimeout: 30000 };
// Look back this far so the Received log captures mail already read in another
// client (the unseen-only fetch missed those). Dedup by message-id keeps each
// poll cheap — only un-logged messages are downloaded + processed.
const LOOKBACK_DAYS = 90;
function makeImapClient(cfg) {
return new ImapFlow({ host: cfg.host, port: cfg.port, secure: cfg.secure, auth: cfg.auth, logger: false, ...IMAP_TIMEOUTS });
@@ -234,15 +238,45 @@ async function pollOnce() {
try {
await connectWithTimeout(client);
const lock = await client.getMailboxLock(cfg.folder);
/* eslint-disable no-await-in-loop */
try {
// eslint-disable-next-line no-restricted-syntax
for await (const msg of client.fetch({ seen: false }, { source: true, uid: true })) {
let messageId = `uid-${cfg.folder}-${msg.uid}`;
// 1) Candidate UIDs within the lookback window — regardless of \Seen, so
// mail already read elsewhere is still logged. Fall back to unseen-only
// if the server rejects a SINCE search.
const since = new Date(Date.now() - LOOKBACK_DAYS * 86400000);
let uids = [];
try { uids = (await client.search({ since }, { uid: true })) || []; } catch (_) { uids = []; }
if (!uids.length) { try { uids = (await client.search({ seen: false }, { uid: true })) || []; } catch (_) { uids = []; } }
// 2) Cheap envelope-only pass → uid + message-id (no source download).
const candidates = [];
if (uids.length) {
// eslint-disable-next-line no-restricted-syntax
for await (const m of client.fetch(uids, { uid: true, envelope: true }, { uid: true })) {
candidates.push({ uid: m.uid, messageId: (m.envelope && m.envelope.messageId) || `uid-${cfg.folder}-${m.uid}` });
}
}
// 3) Drop ones we've already logged (so each poll only does new work).
const logged = new Set();
for (let i = 0; i < candidates.length; i += 500) {
const chunk = candidates.slice(i, i + 500).map((c) => c.messageId);
const rows = await db('received_emails').whereIn('message_id', chunk).select('message_id');
rows.forEach((r) => logged.add(r.message_id));
}
const fresh = candidates.filter((c) => !logged.has(c.messageId));
// 4) Download + process each fresh message.
for (const cand of fresh) {
let messageId = cand.messageId;
try {
const parsed = await simpleParser(msg.source);
messageId = parsed.messageId || messageId;
const seen = await db('received_emails').where({ message_id: messageId }).first();
if (seen) { await client.messageFlagsAdd(msg.uid, ['\\Seen'], { uid: true }); continue; }
const one = await client.fetchOne(String(cand.uid), { source: true }, { uid: true });
if (!one || !one.source) continue;
const parsed = await simpleParser(one.source);
messageId = parsed.messageId || cand.messageId;
// Re-check with the parsed id (can differ from the envelope's).
const dup = await db('received_emails').where({ message_id: messageId }).first();
if (dup) { await client.messageFlagsAdd(cand.uid, ['\\Seen'], { uid: true }); continue; }
// Ingest attachments. Isolate each so one bad file can't prevent the
// audit row (the symptom: doc lands in Incoming invoices but the
@@ -253,9 +287,7 @@ async function pollOnce() {
const attErrors = [];
for (const att of atts) {
try {
// eslint-disable-next-line no-await-in-loop
const filePath = await saveAttachment(att);
// eslint-disable-next-line no-await-in-loop
const doc = await expenseService.recordInboundDocument({ source: 'email', filePath, originalFilename: att.filename || 'attachment', mimeType: att.contentType }, null);
inboundId = doc.id; count += 1;
} catch (ae) {
@@ -281,14 +313,14 @@ async function pollOnce() {
error: attErrors.length ? attErrors.join('; ').slice(0, 2000) : null,
created_at: new Date(),
});
await client.messageFlagsAdd(msg.uid, ['\\Seen'], { uid: true });
await client.messageFlagsAdd(cand.uid, ['\\Seen'], { uid: true });
processed += 1;
} catch (e) {
// Loud: this is exactly where a silent failure would hide a missing
// Received row.
logger.error?.(`emailIntake: message uid ${msg.uid} (${messageId}) failed: ${e.message}`);
logger.error?.(`emailIntake: message uid ${cand.uid} (${messageId}) failed: ${e.message}`);
try {
await db('received_emails').insert({ message_id: `err-${msg.uid}-${Date.now()}`, status: 'error', error: e.message, attachment_count: 0, received_at: new Date(), created_at: new Date() });
await db('received_emails').insert({ message_id: `err-${cand.uid}-${Date.now()}`, status: 'error', error: e.message, attachment_count: 0, received_at: new Date(), created_at: new Date() });
} catch (ie) {
logger.error?.(`emailIntake: could not even write the error row (received_emails insert failing): ${ie.message}`);
}
@@ -297,6 +329,7 @@ async function pollOnce() {
} finally {
lock.release();
}
/* eslint-enable no-await-in-loop */
await client.logout();
} catch (e) {
logger.error?.(`emailIntake: poll failed: ${e.message}`);