Files
picpeak/backend/src/services/emailIntakeService.js
T
Luca ee46cf2125 feat(messages): Phase 2 — customer (hello@) mailbox + inbound body capture
Second inbound mailbox and real message bodies for the Messages viewer.

Backend:
- Migration 154: mail_accounts table (additional inbound mailboxes beyond the
  primary accounting IMAP) + received_emails.{account_key,to_address,body_html,
  body_text}. Additive/guarded.
- emailIntakeService now polls the accounting mailbox AND every enabled
  mail_accounts row. Extracted pollAccountOnce(cfg, {accountKey, routeToExpenses});
  accounting keeps its exact attachment->expenses behavior, customer mail is
  logged with its body and NOT routed to accounting. Inbound HTML is sanitized
  server-side (sanitize-html) on ingest.
- adminEmail: /received gains an account filter + returns account_key/to_address
  (bodies excluded from the list); new GET /received/:id returns the body;
  GET/POST /accounts + /accounts/test manage the extra mailboxes.

Frontend:
- Customers inbox now pulls the hello@ mailbox; reading pane renders the
  sanitized body in a strict (script-less, no same-origin) sandboxed iframe.
  Accounting inbox shows bodies too. Toolbar context keys off the mailbox.
- CustomerMailboxCard in Settings -> Email (behind the messaging flag) to
  configure + test the hello@ IMAP box.

No behavior change to the existing accounting inbound flow. Frontend build +
migration boot verified.
2026-07-07 10:11:58 +02:00

471 lines
20 KiB
JavaScript
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
/**
* Incoming-mail intake (migration 128). Polls the configured IMAP mailbox
* every minute, parses each unseen message, and drops PDF/image attachments
* into the incoming-invoices inbox (inbound_documents, source='email').
*
* Gated by the `incomingMail` feature flag. Idempotent: each message is logged
* in received_emails keyed by message-id (skip if seen); duplicate attachments
* are caught downstream by the inbound_documents SHA-256 dedup. Handles
* forwarded messages because mailparser flattens nested attachments.
*/
const fsp = require('fs').promises;
const path = require('path');
const { ImapFlow } = require('imapflow');
const { simpleParser } = require('mailparser');
const { db } = require('../database/db');
const logger = require('../utils/logger');
const { getStoragePath } = require('../config/storage');
const expenseService = require('./expenseService');
const sanitizeHtml = require('sanitize-html');
const { isUniqueViolation } = require('../utils/dbErrors');
const ALLOWED_MIME = ['application/pdf', 'image/jpeg', 'image/png'];
let polling = false;
// Fail fast instead of hanging on a wrong host/port (e.g. IMAP pointed at an
// 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 });
}
/** Connect with a hard ceiling, so a stuck TLS handshake can't hang forever. */
async function connectWithTimeout(client, ms = 12000) {
let timer;
const timeout = new Promise((_, reject) => { timer = setTimeout(() => reject(new Error('IMAP connection timed out')), ms); });
try {
await Promise.race([client.connect(), timeout]);
} catch (err) {
// Best-effort teardown if connect lost the race but is still pending.
try { await client.logout(); } catch (_) { /* noop */ }
try { client.close(); } catch (_) { /* noop */ }
throw err;
} finally {
clearTimeout(timer);
}
}
async function isEnabled() {
const flag = await db('feature_flags').where({ key: 'incomingMail' }).first();
return !!(flag && (flag.value === true || flag.value === 1 || flag.value === '1'));
}
async function getImapConfig() {
const c = await db('email_configs').first();
if (!c || !c.imap_host || !c.imap_user) return null;
return {
host: c.imap_host,
port: c.imap_port || 993,
secure: c.imap_secure !== false && c.imap_secure !== 0,
auth: { user: c.imap_user, pass: c.imap_pass || '' },
folder: c.imap_folder || 'INBOX',
};
}
async function saveAttachment(att) {
const year = new Date().getFullYear();
const dir = path.join(getStoragePath(), 'business-docs', 'inbound', String(year));
await fsp.mkdir(dir, { recursive: true });
const ext = path.extname(att.filename || '')
|| (att.contentType === 'application/pdf' ? '.pdf' : att.contentType === 'image/png' ? '.png' : '.jpg');
const filePath = path.join(dir, `email-${Date.now()}-${Math.floor(Math.random() * 1e6)}${ext}`);
await fsp.writeFile(filePath, att.content);
return filePath;
}
/**
* List the mailbox folders on the IMAP server so the UI can offer a
* dropdown instead of a free-text path. Uses the saved config; an
* `override` ({ host, port, secure, user, pass }) lets the admin detect
* folders BEFORE saving. A masked/blank override password falls back to
* the stored one. Returns [{ path, name, specialUse }] (specialUse like
* '\\Inbox' lets the caller auto-select the inbox).
*/
async function listFolders(override) {
let cfg;
if (override && override.host && override.user) {
cfg = {
host: override.host,
port: override.port || 993,
secure: override.secure !== false && override.secure !== 0,
auth: { user: override.user, pass: override.pass || '' },
};
if (!cfg.auth.pass || cfg.auth.pass === '********') {
const stored = await getImapConfig();
cfg.auth.pass = stored?.auth?.pass || '';
}
} else {
cfg = await getImapConfig();
}
if (!cfg) return [];
const client = makeImapClient(cfg);
await connectWithTimeout(client);
try {
const list = await client.list();
return (list || []).map((m) => ({ path: m.path, name: m.name, specialUse: m.specialUse || null }));
} finally {
await client.logout().catch(() => {});
}
}
/**
* Test the IMAP connection: log in, open the configured folder, and report
* the message + unread counts. Non-destructive (marks nothing seen, ingests
* nothing) — proves host/port/user/pass AND that the chosen folder opens.
* Accepts an `override` ({ host, port, secure, user, pass, folder }) so the
* admin can test before saving; a masked/blank password falls back to stored.
*/
async function testConnection(override) {
let cfg; let folder;
if (override && override.host && override.user) {
cfg = {
host: override.host,
port: override.port || 993,
secure: override.secure !== false && override.secure !== 0,
auth: { user: override.user, pass: override.pass || '' },
};
folder = override.folder || 'INBOX';
if (!cfg.auth.pass || cfg.auth.pass === '********') {
const stored = await getImapConfig();
cfg.auth.pass = stored?.auth?.pass || '';
}
} else {
const c = await getImapConfig();
if (!c) return { ok: false, error: 'unconfigured' };
cfg = { host: c.host, port: c.port, secure: c.secure, auth: c.auth };
folder = c.folder;
}
const client = makeImapClient(cfg);
await connectWithTimeout(client);
try {
const status = await client.status(folder, { messages: true, unseen: true });
return { ok: true, folder, messages: status.messages || 0, unseen: status.unseen || 0 };
} finally {
await client.logout().catch(() => {});
}
}
/**
* End-to-end round-trip test: send a uniquely-tagged email through the saved
* SMTP (outgoing) config TO the IMAP mailbox, then poll IMAP until it arrives.
* Proves the whole pipeline (outgoing delivery → incoming reception) in one
* click. Uses SAVED config for both sides (real passwords needed to send +
* read). Cleans up: the test message is deleted once found, so it never
* reaches the accounting inbox.
*
* Returns { ok, seconds, recipient } on success, or { ok:false, sent, reason }.
*/
async function roundTripTest({ timeoutMs = 30000, intervalMs = 3000 } = {}) {
const nodemailer = require('nodemailer');
const crypto = require('crypto');
const c = await db('email_configs').first();
if (!c || !c.smtp_host || !c.smtp_port) return { ok: false, sent: false, reason: 'smtp_unconfigured' };
if (!c.imap_host || !c.imap_user) return { ok: false, sent: false, reason: 'imap_unconfigured' };
// Recipient = the mailbox we poll. imap_user is the mailbox address in the
// typical setup (e.g. rechnungen@…). NOT hardcoded — but some hosts use a
// non-email IMAP login, in which case we can't auto-address the test.
const recipient = c.imap_user;
if (!/^[^\s@]+@[^\s@]+\.[^\s@]+$/.test(recipient || '')) {
return { ok: false, sent: false, reason: 'recipient_not_email', recipient };
}
const token = `ppk-rt-${Date.now()}-${crypto.randomBytes(5).toString('hex')}`;
const subject = `picpeak round-trip test ${token}`;
// 1) Send via the saved SMTP config (mirror the /test route's transport).
const transporter = nodemailer.createTransport({
host: c.smtp_host,
port: parseInt(c.smtp_port, 10),
secure: c.smtp_secure === true || c.smtp_secure === 1,
auth: c.smtp_user && c.smtp_pass ? { user: c.smtp_user, pass: c.smtp_pass } : undefined,
tls: { rejectUnauthorized: c.tls_reject_unauthorized !== false },
});
try {
await transporter.sendMail({
from: `${c.from_name || 'picpeak'} <${c.from_email || c.smtp_user}>`,
to: recipient,
subject,
text: `This is an automated picpeak round-trip test. Token: ${token}. Safe to ignore — it is deleted automatically.`,
});
} catch (err) {
return { ok: false, sent: false, reason: 'send_failed', error: err.message };
}
// 2) Poll IMAP for the tagged message until timeout.
const cfg = await getImapConfig();
const folder = cfg?.folder || 'INBOX';
const client = makeImapClient(cfg);
await connectWithTimeout(client);
const started = Date.now();
// Backoff (PR #622 nit 4): some IMAP servers throttle frequent SELECT/SEARCH.
// Grow the gap ×1.5 (cap 8s) so a 30s test does ~5 polls, not ~10.
let delay = intervalMs;
try {
// eslint-disable-next-line no-constant-condition
while (true) {
const lock = await client.getMailboxLock(folder);
try {
const uids = await client.search({ subject: token }, { uid: true });
if (uids && uids.length) {
await client.messageDelete(uids, { uid: true }).catch(() => {});
return { ok: true, seconds: Math.round((Date.now() - started) / 1000), recipient };
}
} finally {
lock.release();
}
if (Date.now() - started > timeoutMs) {
return { ok: false, sent: true, reason: 'not_received', recipient };
}
// eslint-disable-next-line no-await-in-loop
await new Promise((r) => setTimeout(r, delay));
delay = Math.min(Math.round(delay * 1.5), 8000);
}
} finally {
await client.logout().catch(() => {});
}
}
// Sanitize an inbound HTML body before storing it. Inbound mail is untrusted,
// so this strips scripts/handlers/unknown schemes (the viewer ALSO renders it
// in a script-less sandboxed iframe — defense in depth). Remote images are kept
// (many legit emails embed them) but that is the only tracking-vector allowed.
function sanitizeBody(html) {
if (!html) return null;
try {
return sanitizeHtml(html, {
allowedTags: sanitizeHtml.defaults.allowedTags.concat(['img']),
allowedAttributes: {
...sanitizeHtml.defaults.allowedAttributes,
img: ['src', 'alt', 'width', 'height'],
'*': ['style'],
},
allowedSchemes: ['http', 'https', 'mailto', 'cid'],
});
} catch (_) {
return null;
}
}
/**
* Poll ONE mailbox once and return the count of newly-processed messages.
* `opts.accountKey` tags each received_emails row; `opts.routeToExpenses`
* controls whether PDF/image attachments are dropped into the accounting inbox
* (true for the primary rechnungen@ mailbox) or only logged with the body
* (customer mail, e.g. hello@). The claim/dedup/stale-recovery logic is
* identical for every mailbox.
*/
async function pollAccountOnce(cfg, { accountKey = 'accounting', routeToExpenses = true } = {}) {
const client = makeImapClient(cfg);
let processed = 0;
try {
await connectWithTimeout(client);
const lock = await client.getMailboxLock(cfg.folder);
/* eslint-disable no-await-in-loop */
try {
// 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;
let claimKey = null;
let claimed = false;
try {
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;
// Claim key: a no-Message-ID mail still needs a non-null, per-message
// key so two pollers converge — fall back to the mailbox uid.
claimKey = messageId || `nomsgid-${cand.uid}`;
// Fast-path: already processed. Recover a row left 'processing' by a
// worker that crashed mid-ingest (>10 min) so the attachment isn't
// orphaned — otherwise skip + mark seen.
const existing = await db('received_emails').where({ message_id: claimKey }).first();
if (existing) {
const staleProcessing = existing.status === 'processing'
&& existing.created_at
&& (Date.now() - new Date(existing.created_at).getTime() > 10 * 60 * 1000);
if (!staleProcessing) { await client.messageFlagsAdd(cand.uid, ['\\Seen'], { uid: true }); continue; }
await db('received_emails').where({ id: existing.id }).del();
}
// CLAIM the message atomically BEFORE any ingest. The message_id UNIQUE
// index (migration 128) makes this the real guard: if a second poller
// (multi-replica / rolling deploy) already claimed it, the insert hits
// the unique constraint and we skip cleanly — no double-ingest.
try {
await db('received_emails').insert({
message_id: claimKey,
account_key: accountKey,
status: 'processing',
attachment_count: 0,
received_at: new Date(),
created_at: new Date(),
});
claimed = true;
} catch (ce) {
if (isUniqueViolation(ce)) { await client.messageFlagsAdd(cand.uid, ['\\Seen'], { uid: true }); continue; }
throw ce;
}
// Attachment handling. The accounting mailbox drops PDF/image
// attachments into the incoming-invoices inbox (isolated so one bad
// file can't prevent the audit row). Customer mailboxes only COUNT
// attachments — they aren't supplier invoices.
let inboundId = null;
let count = 0;
const attErrors = [];
if (routeToExpenses) {
const atts = (parsed.attachments || []).filter((a) => ALLOWED_MIME.includes(a.contentType));
for (const att of atts) {
try {
const filePath = await saveAttachment(att);
const doc = await expenseService.recordInboundDocument({ source: 'email', filePath, originalFilename: att.filename || 'attachment', mimeType: att.contentType }, null);
inboundId = doc.id; count += 1;
} catch (ae) {
attErrors.push(ae.message);
logger.error?.(`emailIntake: attachment "${att.filename}" failed: ${ae.message}`);
}
}
} else {
count = (parsed.attachments || []).length;
}
// A malformed Date: header yields an Invalid Date, which throws on a
// Postgres timestamp insert — coerce to now.
const receivedAt = (parsed.date instanceof Date && !Number.isNaN(parsed.date.getTime())) ? parsed.date : new Date();
const status = routeToExpenses
? (count > 0 ? 'ingested' : (attErrors.length ? 'error' : 'no_attachment'))
: 'received';
// Finalise the claimed row — every processed message ends up in the
// Received log with its (sanitized) body, even attachment-less ones.
await db('received_emails').where({ message_id: claimKey }).update({
from_address: ((parsed.from && parsed.from.text) || '').slice(0, 512) || null,
to_address: ((parsed.to && parsed.to.text) || '').slice(0, 512) || null,
subject: parsed.subject || null,
received_at: receivedAt,
attachment_count: count,
status,
inbound_document_id: inboundId,
body_html: sanitizeBody(parsed.html || null),
body_text: parsed.text || null,
error: attErrors.length ? attErrors.join('; ').slice(0, 2000) : null,
});
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 ${cand.uid} (${messageId}) failed: ${e.message}`);
try {
if (claimed && claimKey) {
// We already claimed the row — mark it errored rather than orphan it.
await db('received_emails').where({ message_id: claimKey })
.update({ status: 'error', error: String(e.message).slice(0, 2000) });
} else {
await db('received_emails').insert({ message_id: `err-${cand.uid}-${Date.now()}`, account_key: accountKey, 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}`);
}
}
}
} finally {
lock.release();
}
/* eslint-enable no-await-in-loop */
await client.logout();
} catch (e) {
logger.error?.(`emailIntake: poll failed (${accountKey}): ${e.message}`);
try { await client.close(); } catch (_e) { /* ignore */ }
}
return processed;
}
/**
* Poll ALL configured inbound mailboxes once: the primary accounting IMAP
* (email_configs) plus every enabled row in mail_accounts (e.g. hello@).
* Safe to call repeatedly; self-skips when busy/off.
*/
async function pollOnce() {
if (polling) return { skipped: 'busy' };
if (!(await isEnabled())) return { skipped: 'disabled' };
polling = true;
let processed = 0;
let anyConfigured = false;
try {
// 1) Primary accounting mailbox — routes attachments to the invoices inbox.
const acctCfg = await getImapConfig();
if (acctCfg) {
anyConfigured = true;
processed += await pollAccountOnce(acctCfg, { accountKey: 'accounting', routeToExpenses: true });
}
// 2) Additional mailboxes (customers/hello@) — body captured, no expense
// routing. Guarded so a pre-migration DB simply polls the accounting box.
let extras = [];
try {
if (await db.schema.hasTable('mail_accounts')) {
extras = await db('mail_accounts').where({ enabled: true });
}
} catch (_) { extras = []; }
for (const a of extras) {
if (!a.imap_host || !a.imap_user) continue;
anyConfigured = true;
const cfg = {
host: a.imap_host,
port: a.imap_port || 993,
secure: a.imap_secure !== false && a.imap_secure !== 0,
auth: { user: a.imap_user, pass: a.imap_pass || '' },
folder: a.imap_folder || 'INBOX',
};
// eslint-disable-next-line no-await-in-loop
processed += await pollAccountOnce(cfg, { accountKey: a.account_key, routeToExpenses: false });
}
} finally {
polling = false;
}
if (!anyConfigured) return { skipped: 'unconfigured' };
return { processed };
}
/** Start the 1-minute poll loop (mirrors the outgoing queue cadence). */
function startIncomingMailPoller() {
const run = () => pollOnce().catch((e) => logger.error?.(`emailIntake: ${e.message}`));
setTimeout(run, 15000); // first run shortly after boot
setInterval(run, 60 * 1000);
logger.info?.('Incoming-mail poller started (every 60s when enabled)');
}
module.exports = { pollOnce, startIncomingMailPoller, listFolders, testConnection, roundTripTest, _internal: { getImapConfig, isEnabled, saveAttachment } };