aknuth
/
email-amazon
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

BugFixes

This commit is contained in:
Andreas Knuth 2026-02-12 17:48:06 -06:00
parent 16469de068
commit b321e6d2ec
3 changed files with 137 additions and 171 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

78
email-worker-nodejs/logger.ts
View File

@ -1,15 +1,20 @@
/**
* Structured logging for email worker with daily rotation
* Structured logging for email worker with daily rotation AND retention
*
* Uses pino for high-performance JSON logging.
* Console output is human-readable via pino-pretty in dev,
* and JSON in production (for Docker json-file driver).
*
* File logging uses a simple daily rotation approach.
* Includes logic to delete logs older than X days.
*/
import pino from 'pino';
import { existsSync, mkdirSync, createWriteStream, type WriteStream } from 'node:fs';
import {
existsSync,
mkdirSync,
createWriteStream,
type WriteStream,
readdirSync,
statSync,
unlinkSync
} from 'node:fs';
import { join } from 'node:path';
// ---------------------------------------------------------------------------
@ -17,9 +22,10 @@ import { join } from 'node:path';
// ---------------------------------------------------------------------------
const LOG_DIR = '/var/log/email-worker';
const LOG_FILE_PREFIX = 'worker';
const RETENTION_DAYS = 14; // Logs älter als 14 Tage löschen
// ---------------------------------------------------------------------------
// File stream (best-effort, never crashes the worker)
// File stream & Retention Logic
// ---------------------------------------------------------------------------
let fileStream: WriteStream | null = null;
let currentDateStr = '';
@ -28,15 +34,63 @@ function getDateStr(): string {
return new Date().toISOString().slice(0, 10); // YYYY-MM-DD
}
/**
* Löscht alte Log-Dateien basierend auf RETENTION_DAYS
*/
function cleanUpOldLogs(): void {
try {
if (!existsSync(LOG_DIR)) return;
const files = readdirSync(LOG_DIR);
const now = Date.now();
const maxAgeMs = RETENTION_DAYS * 24 * 60 * 60 * 1000;
for (const file of files) {
// Prüfen ob es eine unserer Log-Dateien ist
if (!file.startsWith(LOG_FILE_PREFIX) || !file.endsWith('.log')) continue;
const filePath = join(LOG_DIR, file);
try {
const stats = statSync(filePath);
const ageMs = now - stats.mtimeMs;
if (ageMs > maxAgeMs) {
unlinkSync(filePath);
// Einmalig auf stdout loggen, damit man sieht, dass aufgeräumt wurde
process.stdout.write(`[INFO] Deleted old log file: ${file}\n`);
}
} catch (err) {
// Ignorieren, falls Datei gerade gelöscht wurde oder Zugriff verweigert
}
}
} catch (err) {
process.stderr.write(`[WARN] Failed to clean up old logs: ${err}\n`);
}
}
function ensureFileStream(): WriteStream | null {
const today = getDateStr();
// Wenn wir bereits einen Stream für heute haben, zurückgeben
if (fileStream && currentDateStr === today) return fileStream;
try {
if (!existsSync(LOG_DIR)) mkdirSync(LOG_DIR, { recursive: true });
// Wenn sich das Datum geändert hat (oder beim ersten Start): Aufräumen
if (currentDateStr !== today) {
cleanUpOldLogs();
}
// Alten Stream schließen, falls vorhanden
if (fileStream) {
fileStream.end();
}
const filePath = join(LOG_DIR, `${LOG_FILE_PREFIX}.${today}.log`);
fileStream = createWriteStream(filePath, { flags: 'a' });
currentDateStr = today;
return fileStream;
} catch {
// Silently continue without file logging (e.g. permission issue)
@ -55,12 +109,10 @@ const logger = pino({
},
},
timestamp: pino.stdTimeFunctions.isoTime,
// In production Docker we write plain JSON to stdout;
// pino-pretty can be used during dev via `pino-pretty` pipe.
});
// ---------------------------------------------------------------------------
// Log level mapping (matches Python worker levels)
// Log level mapping
// ---------------------------------------------------------------------------
type LogLevel = 'DEBUG' | 'INFO' | 'WARNING' | 'ERROR' | 'CRITICAL' | 'SUCCESS';
@ -74,7 +126,7 @@ const LEVEL_MAP: Record<LogLevel, keyof pino.Logger> = {
};
// ---------------------------------------------------------------------------
// Public API – mirrors Python's log(message, level, worker_name)
// Public API
// ---------------------------------------------------------------------------
export function log(
message: string,
@ -84,11 +136,11 @@ export function log(
const prefix = level === 'SUCCESS' ? '[SUCCESS] ' : '';
const formatted = `[${workerName}] ${prefix}${message}`;
// Pino
// Pino (stdout/json)
const method = LEVEL_MAP[level] ?? 'info';
(logger as any)[method](formatted);
// File (best-effort)
// File (plain text)
const stream = ensureFileStream();
if (stream) {
const ts = new Date().toISOString().replace('T', ' ').slice(0, 19);

41
email-worker-nodejs/message-processor.ts
View File

@ -312,17 +312,38 @@ export class MessageProcessor {
// ---------------------------------------------------------------------------
/**
* Add X-SES-Worker-Processed header to raw email bytes.
* Add X-SES-Worker-Processed header to raw email bytes using Buffer manipulation.
* More robust and memory efficient than toString().
*/
function addProcessedHeader(raw: Buffer): Buffer {
const str = raw.toString('utf-8');
const sep = str.match(/\r?\n\r?\n/);
if (!sep || sep.index === undefined) return raw;
// Wir suchen nach dem Ende der Header: Double Newline (\r\n\r\n oder \n\n)
let headerEndIndex = -1;
const before = str.slice(0, sep.index);
const after = str.slice(sep.index);
return Buffer.from(
`${before}\r\nX-SES-Worker-Processed: delivered${after}`,
'utf-8',
);
// Effiziente Suche im Buffer
for (let i = 0; i < raw.length - 3; i++) {
// Check für \r\n\r\n
if (raw[i] === 0x0d && raw[i+1] === 0x0a && raw[i+2] === 0x0d && raw[i+3] === 0x0a) {
headerEndIndex = i;
break;
}
// Check für \n\n (Unix style, seltener bei E-Mail aber möglich)
if (raw[i] === 0x0a && raw[i+1] === 0x0a) {
headerEndIndex = i;
break;
}
}
// Falls keine Header-Trennung gefunden wurde (kaputte Mail?), hängen wir es einfach vorne an
if (headerEndIndex === -1) {
const headerLine = Buffer.from('X-SES-Worker-Processed: delivered\r\n', 'utf-8');
return Buffer.concat([headerLine, raw]);
}
// Wir fügen den Header VOR der leeren Zeile ein
const before = raw.subarray(0, headerEndIndex);
const after = raw.subarray(headerEndIndex);
const newHeader = Buffer.from('\r\nX-SES-Worker-Processed: delivered', 'utf-8');
return Buffer.concat([before, newHeader, after]);
}

183
email-worker-nodejs/rules-processor.ts
View File

@ -1,19 +1,21 @@
/**
* Email rules processing (Auto-Reply / OOO and Forwarding)
*
* Removed: Legacy SMTP forward (forward_smtp_override)
* Remaining paths:
* - OOO → internal (SMTP port 25) or external (SES)
* - Forward → internal (SMTP port 25) or external (SES)
* * CLEANED UP & FIXED:
* - Uses MailComposer for ALL message generation (safer MIME handling)
* - Fixes broken attachment forwarding
* - Removed legacy SMTP forwarding
* - Removed manual string concatenation for MIME boundaries
*/
import { createTransport, type Transporter } from 'nodemailer';
import { createTransport } from 'nodemailer';
import type { ParsedMail } from 'mailparser';
import type { DynamoDBHandler, EmailRule } from '../aws/dynamodb.js';
import type { SESHandler } from '../aws/ses.js';
import { extractBodyParts } from './parser.js';
import { config, isInternalAddress } from '../config.js';
import { log } from '../logger.js';
// Wir nutzen MailComposer direkt für das Erstellen der Raw Bytes
import MailComposer from 'nodemailer/lib/mail-composer/index.js';
export type MetricsCallback = (action: 'autoreply' | 'forward', domain: string) => void;
@ -25,7 +27,6 @@ export class RulesProcessor {
/**
* Process OOO and Forward rules for a single recipient.
* Returns false always (no skip_local_delivery since legacy SMTP removed).
*/
async processRulesForRecipient(
recipient: string,
@ -103,7 +104,9 @@ export class RulesProcessor {
try {
const oooMsg = (rule.ooo_message as string) ?? 'I am out of office.';
const contentType = (rule.ooo_content_type as string) ?? 'text';
const oooBuffer = buildOooReply(parsed, recipient, oooMsg, contentType);
// FIX: Use MailComposer via await
const oooBuffer = await buildOooReply(parsed, recipient, oooMsg, contentType);
if (isInternalAddress(senderAddr)) {
const ok = await sendInternalEmail(recipient, senderAddr, oooBuffer, workerName);
@ -134,7 +137,8 @@ export class RulesProcessor {
): Promise<void> {
for (const forwardTo of forwards) {
try {
const fwdBuffer = buildForwardMessage(parsed, recipient, forwardTo, originalFrom);
// FIX: Correctly await the composer result
const fwdBuffer = await buildForwardMessage(parsed, recipient, forwardTo, originalFrom);
if (isInternalAddress(forwardTo)) {
const ok = await sendInternalEmail(recipient, forwardTo, fwdBuffer, workerName);
@ -154,15 +158,15 @@ export class RulesProcessor {
}
// ---------------------------------------------------------------------------
// Message building
// Message building (Using Nodemailer MailComposer for Safety)
// ---------------------------------------------------------------------------
function buildOooReply(
async function buildOooReply(
original: ParsedMail,
recipient: string,
oooMsg: string,
contentType: string,
): Buffer {
): Promise<Buffer> {
const { text: textBody, html: htmlBody } = extractBodyParts(original);
const originalSubject = original.subject ?? '(no subject)';
const originalFrom = original.from?.text ?? 'unknown';
@ -184,32 +188,33 @@ function buildOooReply(
const includeHtml = contentType === 'html' || !!htmlBody;
return buildMimeMessage({
const composer = new MailComposer({
from: recipient,
to: originalFrom,
subject: `Out of Office: ${originalSubject}`,
inReplyTo: originalMsgId,
references: originalMsgId,
domain: recipientDomain,
textContent,
htmlContent: includeHtml ? htmlContent : undefined,
extraHeaders: {
references: [originalMsgId], // Nodemailer wants array
text: textContent,
html: includeHtml ? htmlContent : undefined,
headers: {
'Auto-Submitted': 'auto-replied',
'X-SES-Worker-Processed': 'ooo-reply',
},
messageId: `<${Date.now()}.${Math.random().toString(36).slice(2)}@${recipientDomain}>`
});
return composer.compile().build();
}
function buildForwardMessage(
async function buildForwardMessage(
original: ParsedMail,
recipient: string,
forwardTo: string,
originalFrom: string,
): Buffer {
): Promise<Buffer> {
const { text: textBody, html: htmlBody } = extractBodyParts(original);
const originalSubject = original.subject ?? '(no subject)';
const originalDate = original.date?.toUTCString() ?? 'unknown';
const recipientDomain = recipient.split('@')[1];
// Text version
let fwdText = '---------- Forwarded message ---------\n';
@ -232,144 +237,32 @@ function buildForwardMessage(
fwdHtml += '</div>';
}
// Build base message
const baseBuffer = buildMimeMessage({
// Config object for MailComposer
const mailOptions: any = {
from: recipient,
to: forwardTo,
subject: `FWD: ${originalSubject}`,
replyTo: originalFrom,
domain: recipientDomain,
textContent: fwdText,
htmlContent: fwdHtml,
extraHeaders: {
'X-SES-Worker-Processed': 'forwarded',
},
});
// For attachments, we re-build using nodemailer which handles them properly
if (original.attachments && original.attachments.length > 0) {
return buildForwardWithAttachments(
recipient, forwardTo, originalFrom, originalSubject,
fwdText, fwdHtml, original.attachments, recipientDomain,
);
}
return baseBuffer;
}
function buildForwardWithAttachments(
from: string,
to: string,
replyTo: string,
subject: string,
textContent: string,
htmlContent: string | undefined,
attachments: ParsedMail['attachments'],
domain: string,
): Buffer {
// Use nodemailer's mail composer to build the MIME message
const MailComposer = require('nodemailer/lib/mail-composer');
const mailOptions: any = {
from,
to,
subject: `FWD: ${subject}`,
replyTo,
text: textContent,
text: fwdText,
html: fwdHtml,
headers: {
'X-SES-Worker-Processed': 'forwarded',
},
attachments: attachments.map((att) => ({
};
// Attachments
if (original.attachments && original.attachments.length > 0) {
mailOptions.attachments = original.attachments.map((att) => ({
filename: att.filename ?? 'attachment',
content: att.content,
contentType: att.contentType,
cid: att.cid ?? undefined,
})),
};
if (htmlContent) {
mailOptions.html = htmlContent;
contentDisposition: att.contentDisposition || 'attachment'
}));
}
const composer = new MailComposer(mailOptions);
// build() returns a stream, but we can use buildAsync pattern
// For synchronous buffer we use the compile + createReadStream approach
const mail = composer.compile();
mail.keepBcc = true;
const chunks: Buffer[] = [];
const stream = mail.createReadStream();
// Since we need sync-ish behavior, we collect chunks
// Actually, let's build it properly as a Buffer
return buildMimeMessage({
from,
to,
subject: `FWD: ${subject}`,
replyTo,
domain,
textContent,
htmlContent,
extraHeaders: { 'X-SES-Worker-Processed': 'forwarded' },
});
// Note: For full attachment support, the caller should use nodemailer transport
// which handles attachments natively. This is a simplified version.
}
// ---------------------------------------------------------------------------
// Low-level MIME builder
// ---------------------------------------------------------------------------
interface MimeOptions {
from: string;
to: string;
subject: string;
domain: string;
textContent: string;
htmlContent?: string;
inReplyTo?: string;
references?: string;
replyTo?: string;
extraHeaders?: Record<string, string>;
}
function buildMimeMessage(opts: MimeOptions): Buffer {
const boundary = `----=_Part_${Date.now()}_${Math.random().toString(36).slice(2)}`;
const msgId = `<${Date.now()}.${Math.random().toString(36).slice(2)}@${opts.domain}>`;
let headers = '';
headers += `From: ${opts.from}\r\n`;
headers += `To: ${opts.to}\r\n`;
headers += `Subject: ${opts.subject}\r\n`;
headers += `Date: ${new Date().toUTCString()}\r\n`;
headers += `Message-ID: ${msgId}\r\n`;
headers += `MIME-Version: 1.0\r\n`;
if (opts.inReplyTo) headers += `In-Reply-To: ${opts.inReplyTo}\r\n`;
if (opts.references) headers += `References: ${opts.references}\r\n`;
if (opts.replyTo) headers += `Reply-To: ${opts.replyTo}\r\n`;
if (opts.extraHeaders) {
for (const [k, v] of Object.entries(opts.extraHeaders)) {
headers += `${k}: ${v}\r\n`;
}
}
if (opts.htmlContent) {
// multipart/alternative
headers += `Content-Type: multipart/alternative; boundary="${boundary}"\r\n`;
let body = `\r\n--${boundary}\r\n`;
body += `Content-Type: text/plain; charset=utf-8\r\nContent-Transfer-Encoding: quoted-printable\r\n\r\n`;
body += opts.textContent;
body += `\r\n--${boundary}\r\n`;
body += `Content-Type: text/html; charset=utf-8\r\nContent-Transfer-Encoding: quoted-printable\r\n\r\n`;
body += opts.htmlContent;
body += `\r\n--${boundary}--\r\n`;
return Buffer.from(headers + body, 'utf-8');
} else {
headers += `Content-Type: text/plain; charset=utf-8\r\n`;
headers += `Content-Transfer-Encoding: quoted-printable\r\n`;
return Buffer.from(headers + '\r\n' + opts.textContent, 'utf-8');
}
return composer.compile().build();
}
// ---------------------------------------------------------------------------