Email reading with node js

emailreader.js

var Imap = require('imap');

var config = require('../config')(process.env['NODE_ENV']);

var winston = require('winston');
var path = require("path");
var logger = new (winston.Logger)({
	transports: [
		new (winston.transports.File)({
			filename: path.resolve(__dirname + '/../logs/all-logs.log'),
			json: true
		})
	],
	exitOnError: false
});

if (config){
	var imap = new Imap(config.email.imap);

	var searchInOpenBox = function() {
		imap.search( [ 'ALL' ], function(err, results) {

			if (err) {
				logger.log("error", "Search error", err);
				return;
			}
			logger.log("debug", "fetching emails", results);
			
			if ( results.length ) {
				var totalMsg = 0;
				var processedMsg = 0;

				var fetcher = imap.fetch(results, {
					bodies: '',
					struct: true,
					markSeen : true
				});
				fetcher.on('message', function( msg, seqno ) {
					totalMsg += 1;

					var mAttr;
					var buffer = '';
					msg.on('body', function(stream, info) {
						stream.on('data', function(chunk) {
							buffer += chunk.toString('utf8');
						});
						stream.once('end', function() {
							logger.log("debug", "email body received", seqno);
						});
					});
					msg.once('attributes', function(attrs) {
						logger.log("debug", "email attributes received", attrs);
						/*
							uid - integer - A 32-bit ID that uniquely identifies this message within its mailbox.
							flags - array - A list of flags currently set on this message.
							date - Date - The internal server date for the message.
							struct - array - The message's body structure (only set if requested with fetch()). See below for an explanation of the format of this property.
							size - integer - The RFC822 message size (only set if requested with fetch()).
						*/
						mAttr = attrs;
					});
					msg.once('end', function() {

						var MailParser = require("mailparser").MailParser;
						var mp = new MailParser({
							streamAttachments: true
						});

						mp.on("headers", function(headers){
							logger.log("debug", "header parsed", headers);
						});

						mp.on("attachment", function(attachment, mail){
							logger.log("debug", "streaming attachments", attachment);

							var path = require("path");
							var fs = require("fs");
							var dir = path.resolve(__dirname + '/../assets/' + mAttr.uid);

							//TODO async not working because of bug
							try{
								fs.mkdirSync(dir);
							} catch (err) {
								if (err.code == 'EEXIST'){

								} else {
									logger.log("error", "Directory create error", err);
								}
							};
							attachment.fullpath = path.join( dir + '/' + attachment.generatedFileName );
							var writeOpts = {highWaterMark: Math.pow(2,16)};
							var output = fs.createWriteStream(attachment.fullpath, writeOpts);
							attachment.stream.pipe(output);
						});

						mp.on("end", function(mail_object){
							/*
								headers - unprocessed headers in the form of - {key: value} - if there were multiple fields with the same key then the value is an array
								from - an array of parsed From addresses - [{address:'[email protected]',name:'Sender Name'}] (should be only one though)
								to - an array of parsed To addresses
								cc - an array of parsed Cc addresses
								bcc - an array of parsed 'Bcc' addresses
								subject - the subject line
								references - an array of reference message id values (not set if no reference values present)
								inReplyTo - an array of In-Reply-To message id values (not set if no in-reply-to values present)
								priority - priority of the e-mail, always one of the following: normal (default), high, low
								text - text body
								html - html body
								date - date field as a Date() object. If date could not be resolved or is not found this field is not set. Check the original date string from headers.date
								attachments - an array of attachments
							*/
							var request = require('request');
							
							var dataToPost = {};
							dataToPost.uId = mAttr.uid;
							if ( mail_object.headers['message-id'] ){
								dataToPost.messageID = mail_object.headers['message-id'];
							}
							if ( mail_object.date )
								dataToPost.receivedDate = mail_object.date.toString();
							dataToPost.headers = JSON.stringify(mail_object.headers);
							dataToPost.from = JSON.stringify(mail_object.from);
							dataToPost.to = JSON.stringify(mail_object.to);
							if ( mail_object.cc )
								dataToPost.cc = JSON.stringify(mail_object.cc);
							if ( mail_object.bcc )
								dataToPost.bcc = JSON.stringify(mail_object.bcc);
							dataToPost.subject = mail_object.subject;
							if ( mail_object.inReplyTo )
								dataToPost.inReplyTo = JSON.stringify(mail_object.inReplyTo);
							if ( mail_object.priority )
								dataToPost.priority = mail_object.priority;
							if ( mail_object.text )
								dataToPost.text = mail_object.text;
							if ( mail_object.html )
								dataToPost.html = mail_object.html;
							if ( mail_object.attachments )
								dataToPost.attachments = JSON.stringify(mail_object.attachments);

							request.post(
								{
									url : config.sslpath + config.email.uri,
									form : dataToPost
								},
								function (err, httpResponse, body) {
									if (err) {
										logger.log("error", "problem with httpRequest", err);
										return;
									}
									if(httpResponse.statusCode !== 200){
										logger.log("error", "problem with httpResponse", body);
										return;
									}
									try {
										var resp = JSON.parse(body);
										if(resp.SUCCESS){
											if (config.email.deleteafter) {
												imap.addFlags( mAttr.uid, ['\\Deleted'], function(err){
													if (err) {
														logger.log("error", "Delete mark error", err);
													} else {
														imap.expunge( mAttr.uid , function(err){
															if (err) {
																logger.log("error", "Delete error", err);
															} else {
																fetcher.emit("deleted", seqno, mAttr.uid);
															}
														});
													}
												});
											}
										} else {
											logger.log("error", "Server side error", resp);
										}
									} catch (err) {
										logger.log("error", "Server side error", { error : err, body : body} );
									}
								}
							);
						});
						mp.write(buffer);
						mp.end();
					});
				});
				fetcher.once('error', function(err) {
					logger.log("error", "Fetch error", err);
				});
				fetcher.on('deleted', function(seqno, uid) {
					logger.log("debug", "message processed successfuly", {seqno:seqno, uid:uid});
					processedMsg += 1;
					if (processedMsg == totalMsg){
						searchInOpenBox();
					}
				});
				fetcher.once('end', function() {
					//we don't need this since we use custom deleted event
					logger.log("debug", "done with fetching all email");
				});
			} else {
				imap.once('mail', function( cnt ){
					if (cnt > 0){
						searchInOpenBox();
					}
				});
			}
		});
	}
	imap.once('mail', function( cnt ){
		if (cnt > 0){
			searchInOpenBox();
		}
	});

	imap.once('ready', function() {
		imap.openBox(config.email.folder, false, function(err, box) {
			if (err){
				logger.log("error", "Box open error", err);
			}
		});
	});

	imap.on('alert', function(message){
		// Emitted when the server issues an alert (e.g. "the server is going down for maintenance").
		logger.log("info", "Server message", message);
	});

	imap.on('expunge', function(seqno){
		/*
			Emitted when a message was expunged externally. seqno is the sequence number (instead of the unique UID) 
				of the message that was expunged. If you are caching sequence numbers, all sequence numbers 
				higher than this value MUST be decremented by 1 in order to stay synchronized with the server and to keep correct continuity.

		*/
		logger.log("debug", "message was expunged externally", seqno);
	});

	imap.on('uidvalidity', function(uidvalidity){
		//Emitted if the UID validity value for the currently open mailbox changes during the current session.
		logger.log("info", "UID validity value changed", uidvalidity);
	});

	imap.on('update', function(seqno, info){
		//Emitted when message metadata (e.g. flags) changes externally.
		logger.log("debug", "Message Updated", {seqno : seqno, info : info});
	});

	imap.on('error', function(err) {
		//Emitted when an error occurs. The 'source' property will be set to indicate where the error originated from.
		logger.log("error", "Connection error", err);
	});

	imap.on('close', function( hadError ) {
		//Emitted when the connection has completely closed
		if (hadError) {
			logger.log("error", "Connection closed", {hadError : hadError});

			setTimeout(function(){
				imap.connect();
			}, 15000);
		} else {
			logger.log("info", "Connection closed", {hadError : hadError});
		}
	});

	imap.on('end', function() {
		logger.log("debug", "Connection ended");
	});

	imap.connect();
} else {
	logger.log('error', "Missing config");
}