fix(platformservice): Refactor module structure to update import paths and file organization
This commit is contained in:
623
ts/mta/classes.emailsendjob.ts
Normal file
623
ts/mta/classes.emailsendjob.ts
Normal file
@ -0,0 +1,623 @@
|
||||
import * as plugins from '../plugins.js';
|
||||
import * as paths from '../paths.js';
|
||||
import { Email } from './classes.email.js';
|
||||
import { EmailSignJob } from './classes.emailsignjob.js';
|
||||
import type { MtaService } from './classes.mta.js';
|
||||
|
||||
// Configuration options for email sending
|
||||
export interface IEmailSendOptions {
|
||||
maxRetries?: number;
|
||||
retryDelay?: number; // in milliseconds
|
||||
connectionTimeout?: number; // in milliseconds
|
||||
tlsOptions?: plugins.tls.ConnectionOptions;
|
||||
debugMode?: boolean;
|
||||
}
|
||||
|
||||
// Email delivery status
|
||||
export enum DeliveryStatus {
|
||||
PENDING = 'pending',
|
||||
SENDING = 'sending',
|
||||
DELIVERED = 'delivered',
|
||||
FAILED = 'failed',
|
||||
DEFERRED = 'deferred' // Temporary failure, will retry
|
||||
}
|
||||
|
||||
// Detailed information about delivery attempts
|
||||
export interface DeliveryInfo {
|
||||
status: DeliveryStatus;
|
||||
attempts: number;
|
||||
error?: Error;
|
||||
lastAttempt?: Date;
|
||||
nextAttempt?: Date;
|
||||
mxServer?: string;
|
||||
deliveryTime?: Date;
|
||||
logs: string[];
|
||||
}
|
||||
|
||||
export class EmailSendJob {
|
||||
mtaRef: MtaService;
|
||||
private email: Email;
|
||||
private socket: plugins.net.Socket | plugins.tls.TLSSocket = null;
|
||||
private mxServers: string[] = [];
|
||||
private currentMxIndex = 0;
|
||||
private options: IEmailSendOptions;
|
||||
public deliveryInfo: DeliveryInfo;
|
||||
|
||||
constructor(mtaRef: MtaService, emailArg: Email, options: IEmailSendOptions = {}) {
|
||||
this.email = emailArg;
|
||||
this.mtaRef = mtaRef;
|
||||
|
||||
// Set default options
|
||||
this.options = {
|
||||
maxRetries: options.maxRetries || 3,
|
||||
retryDelay: options.retryDelay || 300000, // 5 minutes
|
||||
connectionTimeout: options.connectionTimeout || 30000, // 30 seconds
|
||||
tlsOptions: options.tlsOptions || { rejectUnauthorized: true },
|
||||
debugMode: options.debugMode || false
|
||||
};
|
||||
|
||||
// Initialize delivery info
|
||||
this.deliveryInfo = {
|
||||
status: DeliveryStatus.PENDING,
|
||||
attempts: 0,
|
||||
logs: []
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Send the email with retry logic
|
||||
*/
|
||||
async send(): Promise<DeliveryStatus> {
|
||||
try {
|
||||
// Check if the email is valid before attempting to send
|
||||
this.validateEmail();
|
||||
|
||||
// Resolve MX records for the recipient domain
|
||||
await this.resolveMxRecords();
|
||||
|
||||
// Try to send the email
|
||||
return await this.attemptDelivery();
|
||||
} catch (error) {
|
||||
this.log(`Critical error in send process: ${error.message}`);
|
||||
this.deliveryInfo.status = DeliveryStatus.FAILED;
|
||||
this.deliveryInfo.error = error;
|
||||
|
||||
// Save failed email for potential future retry or analysis
|
||||
await this.saveFailed();
|
||||
return DeliveryStatus.FAILED;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Validate the email before sending
|
||||
*/
|
||||
private validateEmail(): void {
|
||||
if (!this.email.to || this.email.to.length === 0) {
|
||||
throw new Error('No recipients specified');
|
||||
}
|
||||
|
||||
if (!this.email.from) {
|
||||
throw new Error('No sender specified');
|
||||
}
|
||||
|
||||
const fromDomain = this.email.getFromDomain();
|
||||
if (!fromDomain) {
|
||||
throw new Error('Invalid sender domain');
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve MX records for the recipient domain
|
||||
*/
|
||||
private async resolveMxRecords(): Promise<void> {
|
||||
const domain = this.email.getPrimaryRecipient()?.split('@')[1];
|
||||
if (!domain) {
|
||||
throw new Error('Invalid recipient domain');
|
||||
}
|
||||
|
||||
this.log(`Resolving MX records for domain: ${domain}`);
|
||||
try {
|
||||
const addresses = await this.resolveMx(domain);
|
||||
|
||||
// Sort by priority (lowest number = highest priority)
|
||||
addresses.sort((a, b) => a.priority - b.priority);
|
||||
|
||||
this.mxServers = addresses.map(mx => mx.exchange);
|
||||
this.log(`Found ${this.mxServers.length} MX servers: ${this.mxServers.join(', ')}`);
|
||||
|
||||
if (this.mxServers.length === 0) {
|
||||
throw new Error(`No MX records found for domain: ${domain}`);
|
||||
}
|
||||
} catch (error) {
|
||||
this.log(`Failed to resolve MX records: ${error.message}`);
|
||||
throw new Error(`MX lookup failed for ${domain}: ${error.message}`);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Attempt to deliver the email with retries
|
||||
*/
|
||||
private async attemptDelivery(): Promise<DeliveryStatus> {
|
||||
while (this.deliveryInfo.attempts < this.options.maxRetries) {
|
||||
this.deliveryInfo.attempts++;
|
||||
this.deliveryInfo.lastAttempt = new Date();
|
||||
this.deliveryInfo.status = DeliveryStatus.SENDING;
|
||||
|
||||
try {
|
||||
this.log(`Delivery attempt ${this.deliveryInfo.attempts} of ${this.options.maxRetries}`);
|
||||
|
||||
// Try each MX server in order of priority
|
||||
while (this.currentMxIndex < this.mxServers.length) {
|
||||
const currentMx = this.mxServers[this.currentMxIndex];
|
||||
this.deliveryInfo.mxServer = currentMx;
|
||||
|
||||
try {
|
||||
this.log(`Attempting delivery to MX server: ${currentMx}`);
|
||||
await this.connectAndSend(currentMx);
|
||||
|
||||
// If we get here, email was sent successfully
|
||||
this.deliveryInfo.status = DeliveryStatus.DELIVERED;
|
||||
this.deliveryInfo.deliveryTime = new Date();
|
||||
this.log(`Email delivered successfully to ${currentMx}`);
|
||||
|
||||
// Save successful email record
|
||||
await this.saveSuccess();
|
||||
return DeliveryStatus.DELIVERED;
|
||||
} catch (error) {
|
||||
this.log(`Error with MX ${currentMx}: ${error.message}`);
|
||||
|
||||
// Clean up socket if it exists
|
||||
if (this.socket) {
|
||||
this.socket.destroy();
|
||||
this.socket = null;
|
||||
}
|
||||
|
||||
// Try the next MX server
|
||||
this.currentMxIndex++;
|
||||
|
||||
// If this is a permanent failure, don't try other MX servers
|
||||
if (this.isPermanentFailure(error)) {
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// If we've tried all MX servers without success, throw an error
|
||||
throw new Error('All MX servers failed');
|
||||
} catch (error) {
|
||||
// Check if this is a permanent failure
|
||||
if (this.isPermanentFailure(error)) {
|
||||
this.log(`Permanent failure: ${error.message}`);
|
||||
this.deliveryInfo.status = DeliveryStatus.FAILED;
|
||||
this.deliveryInfo.error = error;
|
||||
|
||||
// Save failed email for analysis
|
||||
await this.saveFailed();
|
||||
return DeliveryStatus.FAILED;
|
||||
}
|
||||
|
||||
// This is a temporary failure, we can retry
|
||||
this.log(`Temporary failure: ${error.message}`);
|
||||
|
||||
// If this is the last attempt, mark as failed
|
||||
if (this.deliveryInfo.attempts >= this.options.maxRetries) {
|
||||
this.deliveryInfo.status = DeliveryStatus.FAILED;
|
||||
this.deliveryInfo.error = error;
|
||||
|
||||
// Save failed email for analysis
|
||||
await this.saveFailed();
|
||||
return DeliveryStatus.FAILED;
|
||||
}
|
||||
|
||||
// Schedule the next retry
|
||||
const nextRetryTime = new Date(Date.now() + this.options.retryDelay);
|
||||
this.deliveryInfo.status = DeliveryStatus.DEFERRED;
|
||||
this.deliveryInfo.nextAttempt = nextRetryTime;
|
||||
this.log(`Will retry at ${nextRetryTime.toISOString()}`);
|
||||
|
||||
// Wait before retrying
|
||||
await this.delay(this.options.retryDelay);
|
||||
|
||||
// Reset MX server index for the next attempt
|
||||
this.currentMxIndex = 0;
|
||||
}
|
||||
}
|
||||
|
||||
// If we get here, all retries failed
|
||||
this.deliveryInfo.status = DeliveryStatus.FAILED;
|
||||
await this.saveFailed();
|
||||
return DeliveryStatus.FAILED;
|
||||
}
|
||||
|
||||
/**
|
||||
* Connect to a specific MX server and send the email
|
||||
*/
|
||||
private async connectAndSend(mxServer: string): Promise<void> {
|
||||
return new Promise((resolve, reject) => {
|
||||
let commandTimeout: NodeJS.Timeout;
|
||||
|
||||
// Function to clear timeouts and remove listeners
|
||||
const cleanup = () => {
|
||||
clearTimeout(commandTimeout);
|
||||
if (this.socket) {
|
||||
this.socket.removeAllListeners();
|
||||
}
|
||||
};
|
||||
|
||||
// Function to set a timeout for each command
|
||||
const setCommandTimeout = () => {
|
||||
clearTimeout(commandTimeout);
|
||||
commandTimeout = setTimeout(() => {
|
||||
this.log('Connection timed out');
|
||||
cleanup();
|
||||
if (this.socket) {
|
||||
this.socket.destroy();
|
||||
this.socket = null;
|
||||
}
|
||||
reject(new Error('Connection timed out'));
|
||||
}, this.options.connectionTimeout);
|
||||
};
|
||||
|
||||
// Connect to the MX server
|
||||
this.log(`Connecting to ${mxServer}:25`);
|
||||
setCommandTimeout();
|
||||
|
||||
this.socket = plugins.net.connect(25, mxServer);
|
||||
|
||||
this.socket.on('error', (err) => {
|
||||
this.log(`Socket error: ${err.message}`);
|
||||
cleanup();
|
||||
reject(err);
|
||||
});
|
||||
|
||||
// Set up the command sequence
|
||||
this.socket.once('data', async (data) => {
|
||||
try {
|
||||
const greeting = data.toString();
|
||||
this.log(`Server greeting: ${greeting.trim()}`);
|
||||
|
||||
if (!greeting.startsWith('220')) {
|
||||
throw new Error(`Unexpected server greeting: ${greeting}`);
|
||||
}
|
||||
|
||||
// EHLO command
|
||||
const fromDomain = this.email.getFromDomain();
|
||||
await this.sendCommand(`EHLO ${fromDomain}\r\n`, '250');
|
||||
|
||||
// Try STARTTLS if available
|
||||
try {
|
||||
await this.sendCommand('STARTTLS\r\n', '220');
|
||||
this.upgradeToTLS(mxServer, fromDomain);
|
||||
// The TLS handshake and subsequent commands will continue in the upgradeToTLS method
|
||||
// resolve will be called from there if successful
|
||||
} catch (error) {
|
||||
this.log(`STARTTLS failed or not supported: ${error.message}`);
|
||||
this.log('Continuing with unencrypted connection');
|
||||
|
||||
// Continue with unencrypted connection
|
||||
await this.sendEmailCommands();
|
||||
cleanup();
|
||||
resolve();
|
||||
}
|
||||
} catch (error) {
|
||||
cleanup();
|
||||
reject(error);
|
||||
}
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Upgrade the connection to TLS
|
||||
*/
|
||||
private upgradeToTLS(mxServer: string, fromDomain: string): void {
|
||||
this.log('Starting TLS handshake');
|
||||
|
||||
const tlsOptions = {
|
||||
...this.options.tlsOptions,
|
||||
socket: this.socket,
|
||||
servername: mxServer
|
||||
};
|
||||
|
||||
// Create TLS socket
|
||||
this.socket = plugins.tls.connect(tlsOptions);
|
||||
|
||||
// Handle TLS connection
|
||||
this.socket.once('secureConnect', async () => {
|
||||
try {
|
||||
this.log('TLS connection established');
|
||||
|
||||
// Send EHLO again over TLS
|
||||
await this.sendCommand(`EHLO ${fromDomain}\r\n`, '250');
|
||||
|
||||
// Send the email
|
||||
await this.sendEmailCommands();
|
||||
|
||||
this.socket.destroy();
|
||||
this.socket = null;
|
||||
} catch (error) {
|
||||
this.log(`Error in TLS session: ${error.message}`);
|
||||
this.socket.destroy();
|
||||
this.socket = null;
|
||||
}
|
||||
});
|
||||
|
||||
this.socket.on('error', (err) => {
|
||||
this.log(`TLS error: ${err.message}`);
|
||||
this.socket.destroy();
|
||||
this.socket = null;
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Send SMTP commands to deliver the email
|
||||
*/
|
||||
private async sendEmailCommands(): Promise<void> {
|
||||
// MAIL FROM command
|
||||
await this.sendCommand(`MAIL FROM:<${this.email.from}>\r\n`, '250');
|
||||
|
||||
// RCPT TO command for each recipient
|
||||
for (const recipient of this.email.getAllRecipients()) {
|
||||
await this.sendCommand(`RCPT TO:<${recipient}>\r\n`, '250');
|
||||
}
|
||||
|
||||
// DATA command
|
||||
await this.sendCommand('DATA\r\n', '354');
|
||||
|
||||
// Create the email message with DKIM signature
|
||||
const message = await this.createEmailMessage();
|
||||
|
||||
// Send the message content
|
||||
await this.sendCommand(message);
|
||||
await this.sendCommand('\r\n.\r\n', '250');
|
||||
|
||||
// QUIT command
|
||||
await this.sendCommand('QUIT\r\n', '221');
|
||||
}
|
||||
|
||||
/**
|
||||
* Create the full email message with headers and DKIM signature
|
||||
*/
|
||||
private async createEmailMessage(): Promise<string> {
|
||||
this.log('Preparing email message');
|
||||
|
||||
const messageId = `<${plugins.uuid.v4()}@${this.email.getFromDomain()}>`;
|
||||
const boundary = '----=_NextPart_' + plugins.uuid.v4();
|
||||
|
||||
// Prepare headers
|
||||
const headers = {
|
||||
'Message-ID': messageId,
|
||||
'From': this.email.from,
|
||||
'To': this.email.to.join(', '),
|
||||
'Subject': this.email.subject,
|
||||
'Content-Type': `multipart/mixed; boundary="${boundary}"`,
|
||||
'Date': new Date().toUTCString()
|
||||
};
|
||||
|
||||
// Add CC header if present
|
||||
if (this.email.cc && this.email.cc.length > 0) {
|
||||
headers['Cc'] = this.email.cc.join(', ');
|
||||
}
|
||||
|
||||
// Add custom headers
|
||||
for (const [key, value] of Object.entries(this.email.headers || {})) {
|
||||
headers[key] = value;
|
||||
}
|
||||
|
||||
// Add priority header if not normal
|
||||
if (this.email.priority && this.email.priority !== 'normal') {
|
||||
const priorityValue = this.email.priority === 'high' ? '1' : '5';
|
||||
headers['X-Priority'] = priorityValue;
|
||||
}
|
||||
|
||||
// Create body
|
||||
let body = '';
|
||||
|
||||
// Text part
|
||||
body += `--${boundary}\r\nContent-Type: text/plain; charset=utf-8\r\n\r\n${this.email.text}\r\n`;
|
||||
|
||||
// HTML part if present
|
||||
if (this.email.html) {
|
||||
body += `--${boundary}\r\nContent-Type: text/html; charset=utf-8\r\n\r\n${this.email.html}\r\n`;
|
||||
}
|
||||
|
||||
// Attachments
|
||||
for (const attachment of this.email.attachments) {
|
||||
body += `--${boundary}\r\nContent-Type: ${attachment.contentType}; name="${attachment.filename}"\r\n`;
|
||||
body += 'Content-Transfer-Encoding: base64\r\n';
|
||||
body += `Content-Disposition: attachment; filename="${attachment.filename}"\r\n`;
|
||||
|
||||
// Add Content-ID for inline attachments if present
|
||||
if (attachment.contentId) {
|
||||
body += `Content-ID: <${attachment.contentId}>\r\n`;
|
||||
}
|
||||
|
||||
body += '\r\n';
|
||||
body += attachment.content.toString('base64') + '\r\n';
|
||||
}
|
||||
|
||||
// End of message
|
||||
body += `--${boundary}--\r\n`;
|
||||
|
||||
// Create DKIM signature
|
||||
const dkimSigner = new EmailSignJob(this.mtaRef, {
|
||||
domain: this.email.getFromDomain(),
|
||||
selector: 'mta',
|
||||
headers: headers,
|
||||
body: body,
|
||||
});
|
||||
|
||||
// Build the message with headers
|
||||
let headerString = '';
|
||||
for (const [key, value] of Object.entries(headers)) {
|
||||
headerString += `${key}: ${value}\r\n`;
|
||||
}
|
||||
let message = headerString + '\r\n' + body;
|
||||
|
||||
// Add DKIM signature header
|
||||
let signatureHeader = await dkimSigner.getSignatureHeader(message);
|
||||
message = `${signatureHeader}${message}`;
|
||||
|
||||
return message;
|
||||
}
|
||||
|
||||
/**
|
||||
* Send a command to the SMTP server and wait for the expected response
|
||||
*/
|
||||
private sendCommand(command: string, expectedResponseCode?: string): Promise<string> {
|
||||
return new Promise((resolve, reject) => {
|
||||
if (!this.socket) {
|
||||
return reject(new Error('Socket not connected'));
|
||||
}
|
||||
|
||||
// Debug log for commands (except DATA which can be large)
|
||||
if (this.options.debugMode && !command.startsWith('--')) {
|
||||
const logCommand = command.length > 100
|
||||
? command.substring(0, 97) + '...'
|
||||
: command;
|
||||
this.log(`Sending: ${logCommand.replace(/\r\n/g, '<CRLF>')}`);
|
||||
}
|
||||
|
||||
this.socket.write(command, (error) => {
|
||||
if (error) {
|
||||
this.log(`Write error: ${error.message}`);
|
||||
return reject(error);
|
||||
}
|
||||
|
||||
// If no response is expected, resolve immediately
|
||||
if (!expectedResponseCode) {
|
||||
return resolve('');
|
||||
}
|
||||
|
||||
// Set a timeout for the response
|
||||
const responseTimeout = setTimeout(() => {
|
||||
this.log('Response timeout');
|
||||
reject(new Error('Response timeout'));
|
||||
}, this.options.connectionTimeout);
|
||||
|
||||
// Wait for the response
|
||||
this.socket.once('data', (data) => {
|
||||
clearTimeout(responseTimeout);
|
||||
const response = data.toString();
|
||||
|
||||
if (this.options.debugMode) {
|
||||
this.log(`Received: ${response.trim()}`);
|
||||
}
|
||||
|
||||
if (response.startsWith(expectedResponseCode)) {
|
||||
resolve(response);
|
||||
} else {
|
||||
const error = new Error(`Unexpected server response: ${response.trim()}`);
|
||||
this.log(error.message);
|
||||
reject(error);
|
||||
}
|
||||
});
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Determine if an error represents a permanent failure
|
||||
*/
|
||||
private isPermanentFailure(error: Error): boolean {
|
||||
if (!error || !error.message) return false;
|
||||
|
||||
const message = error.message.toLowerCase();
|
||||
|
||||
// Check for permanent SMTP error codes (5xx)
|
||||
if (message.match(/^5\d\d/)) return true;
|
||||
|
||||
// Check for specific permanent failure messages
|
||||
const permanentFailurePatterns = [
|
||||
'no such user',
|
||||
'user unknown',
|
||||
'domain not found',
|
||||
'invalid domain',
|
||||
'rejected',
|
||||
'denied',
|
||||
'prohibited',
|
||||
'authentication required',
|
||||
'authentication failed',
|
||||
'unauthorized'
|
||||
];
|
||||
|
||||
return permanentFailurePatterns.some(pattern => message.includes(pattern));
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve MX records for a domain
|
||||
*/
|
||||
private resolveMx(domain: string): Promise<plugins.dns.MxRecord[]> {
|
||||
return new Promise((resolve, reject) => {
|
||||
plugins.dns.resolveMx(domain, (err, addresses) => {
|
||||
if (err) {
|
||||
reject(err);
|
||||
} else {
|
||||
resolve(addresses);
|
||||
}
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Add a log entry
|
||||
*/
|
||||
private log(message: string): void {
|
||||
const timestamp = new Date().toISOString();
|
||||
const logEntry = `[${timestamp}] ${message}`;
|
||||
this.deliveryInfo.logs.push(logEntry);
|
||||
|
||||
if (this.options.debugMode) {
|
||||
console.log(`EmailSendJob: ${logEntry}`);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Save a successful email for record keeping
|
||||
*/
|
||||
private async saveSuccess(): Promise<void> {
|
||||
try {
|
||||
plugins.smartfile.fs.ensureDirSync(paths.sentEmailsDir);
|
||||
const emailContent = await this.createEmailMessage();
|
||||
const fileName = `${Date.now()}_success_${this.email.getPrimaryRecipient()}.eml`;
|
||||
plugins.smartfile.memory.toFsSync(emailContent, plugins.path.join(paths.sentEmailsDir, fileName));
|
||||
|
||||
// Save delivery info
|
||||
const infoFileName = `${Date.now()}_success_${this.email.getPrimaryRecipient()}.json`;
|
||||
plugins.smartfile.memory.toFsSync(
|
||||
JSON.stringify(this.deliveryInfo, null, 2),
|
||||
plugins.path.join(paths.sentEmailsDir, infoFileName)
|
||||
);
|
||||
} catch (error) {
|
||||
console.error('Error saving successful email:', error);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Save a failed email for potential retry
|
||||
*/
|
||||
private async saveFailed(): Promise<void> {
|
||||
try {
|
||||
plugins.smartfile.fs.ensureDirSync(paths.failedEmailsDir);
|
||||
const emailContent = await this.createEmailMessage();
|
||||
const fileName = `${Date.now()}_failed_${this.email.getPrimaryRecipient()}.eml`;
|
||||
plugins.smartfile.memory.toFsSync(emailContent, plugins.path.join(paths.failedEmailsDir, fileName));
|
||||
|
||||
// Save delivery info
|
||||
const infoFileName = `${Date.now()}_failed_${this.email.getPrimaryRecipient()}.json`;
|
||||
plugins.smartfile.memory.toFsSync(
|
||||
JSON.stringify(this.deliveryInfo, null, 2),
|
||||
plugins.path.join(paths.failedEmailsDir, infoFileName)
|
||||
);
|
||||
} catch (error) {
|
||||
console.error('Error saving failed email:', error);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Simple delay function
|
||||
*/
|
||||
private delay(ms: number): Promise<void> {
|
||||
return new Promise(resolve => setTimeout(resolve, ms));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user