'use strict'; const log = require('../lib/log'); const config = require('config'); const net = require('net'); const campaigns = require('../models/campaigns'); const contextHelpers = require('../lib/context-helpers'); const { SubscriptionStatus } = require('../../shared/lists'); const bluebird = require('bluebird'); const seenIds = new Set(); let remainder = ''; let reading = false; async function readNextChunks() { if (reading) { return false; } reading = true; while (true) { const chunk = socket.read(); if (chunk === null) { reading = false; return; } const lines = (remainder + chunk.toString()).split(/\r?\n/); remainder = lines.pop(); for (const line of lines) { try { const match = /\bstatus=(bounced|sent)\b/.test(line) && line.match(/\bpostfix\/\w+\[\d+\]:\s*([^:]+).*?status=(\w+)/); if (match) { const queueId = match[1]; let queued = ''; let queuedAs = ''; if (!seenIds.has(queueId)) { seenIds.add(queueId); // Losacno: Check for local requeue const status = match[2]; log.verbose('POSTFIXBOUNCE', 'Checking message %s for local requeue (status: %s)', queueId, status); if (status === 'sent') { // Save new queueId to update message's previous queueId (thanks @mfechner ) queued = / relay=/.test(line) && line.match(/status=sent \((.*)\)/); if (queued) { queued = queued[1]; queuedAs = queued.match(/ queued as (\w+)/); if (queuedAs) { queuedAs = queuedAs[1]; } else { queuedAs = ''; } } } const message = await campaigns.getMessageByResponseId(queueId); if (message) { if (queuedAs || status === 'sent') { log.verbose('POSTFIXBOUNCE', 'Message %s locally requeued as %s', queueId, queuedAs); // Update message's previous queueId (thanks @mfechner ) campaigns.updateMessageResponse(contextHelpers.getAdminContext(), message, queued, queuedAs); log.verbose('POSTFIXBOUNCE', 'Successfully changed message queueId to %s', queuedAs); } else { campaigns.changeStatusByMessage(contextHelpers.getAdminContext(), message, SubscriptionStatus.BOUNCED, true); log.verbose('POSTFIXBOUNCE', 'Marked message %s as bounced', queueId); } // No need to keep in memory... free it ( thanks @witzig ) seenIds.delete(queueId); } } } } catch (err) { log.error('POSTFIXBOUNCE', err && err.stack); } } } } function start(callback) { if (!config.postfixBounce.enabled) { return setImmediate(callback); } let started = false; // Not sure why all this magic around "started". But it was there this way in Mailtrain v1, so we kept it. const server = net.createServer(socket => { socket.on('readable', readNextChunks); }); server.on('error', err => { const port = config.postfixBounce.port; const bind = typeof port === 'string' ? 'Pipe ' + port : 'Port ' + port; switch (err.code) { case 'EACCES': log.error('POSTFIXBOUNCE', '%s requires elevated privileges.', bind); break; case 'EADDRINUSE': log.error('POSTFIXBOUNCE', '%s is already in use', bind); break; default: log.error('POSTFIXBOUNCE', err); } if (!started) { started = true; return callback(err); } }); server.listen(config.postfixBounce.port, config.postfixBounce.host, () => { if (started) { return server.close(); } started = true; log.info('POSTFIXBOUNCE', 'Server listening on port %s', config.postfixBounce.port); setImmediate(callback); }); } module.exports.start = bluebird.promisify(start);