NodeJS Push Notification Routines

push.android.js

var config = require('../config.js');
var gcm = require("node-gcm");
var gcmConnection = new gcm.Sender(config.GCM_KEY);
global.Promise = require('bluebird');
var zip = require('lodash/zip');

var feedbackHandlers = [];

function send(pushTokens, message, payload, expiry, unreadCounter){
	return new Promise((resolve, reject) => {
		var notification = {
			collapseKey: "tr-" + Date.now(),
			delayWhileIdle: true,
			timeToLive: 60 * 60 * 10, // expiry ? Math.floor(expiracio.getTime()/1000 - (new Date()).getTime()/1000) : 172800,
			data: {
				title: config.APP_NAME,
				message: message,
				// sound: "push",
				badge: unreadCounter
			}
		};

		Object.assign(notification.data, payload);

		// delivery
		gcmConnection.send(new gcm.Message(notification), pushTokens, 3, (err, result) => {
			if(err) return reject(err || "Notificació no enviada");
			
			// RESULT (in case of error)
			//  { multicast_id: 5215101923310065000,
			//   success: 0,
			//   failure: 1,
			//   canonical_ids: 0,
			//   results: [ { error: 'NotRegistered' } ]
			// }

			resolve(result);
		});
	})
	.then(res => {
		if(!res) return;

		// regrouping like [ [tokenStr1, resultObj1], [tokenStr2, resultObj2], ... ]
		const grouped = zip(pushTokens, res.results);

		// map like { ...result, token: "..." }
		return grouped.map(tuple => {
			var result = tuple[1] || {};
			result.token = tuple[0];
			return result;
		});
	})
	.then(results => {
		var tokensToUpdate = [], tokensToRemove = [];

		const failures = results.reduce((prev, result) => {
			// puchToken cleanup
			if(result.registration_id)
				tokensToUpdate.push({from: result.token, to: result.registration_id});
			else if (result.error === 'InvalidRegistration' || result.error === 'NotRegistered')
				tokensToRemove.push(result.token);

			// count failures
			return prev + (result.failure || 0);
		}, 0);
		
		feedbackHandlers.forEach(func => {
			func(tokensToUpdate, tokensToRemove);
		});

		return {failures};
	});
}

function addFeedbackHandler(func){
	if(typeof func !== 'function')
		throw new Error("Not a valid function");

	feedbackHandlers.push(func);
}

module.exports = {
    send: send,
    addFeedbackHandler: addFeedbackHandler
}

push.ios.js

var config = require('../config.js');
var apn = require('apn');
global.Promise = require('bluebird');

var apnConfig = {
	// buffersNotifications: true,
	fastMode: true,
	cert: config.APN_CERT_FILE,
	key: config.APN_KEY_FILE,
	production: !config.DEBUG
};
var feedbackConfig = {
	cert: config.APN_CERT_FILE,
	key: config.APN_KEY_FILE,
	production: !config.DEBUG,
	batchFeedback: true,
  interval: 300
};
var apnConnection = new apn.Connection(apnConfig);
apnConnection.on('transmissionError', onTransmissionError);

var apnFeedback = new apn.Feedback(feedbackConfig)
apnFeedback.on('feedback', onFeedback);

var feedbackHandlers = [];

function send(pushTokens, message, payload, expiry, unreadCounter){
	return Promise.try(function(){
		var notification = new apn.Notification();
		notification.expiry = expiry ? Math.floor(expiry.getTime() / 1000) : ((new Date()).getTime()/1000) + 172800;
		notification.badge = unreadCounter;
		notification.truncateAtWordEnd = true;

		// notification.sound = "www/push.caf";
		// notification.defered = defered;
		notification.alert = message;
		notification.priority = 10;  // 10 => asap, 5 => chillout

		// payload
		notification.payload = payload;

		apnConnection.pushNotification(notification, pushTokens);
	});
}

function addFeedbackHandler(func){
	if(typeof func !== 'function') throw "Not a valid function";

	feedbackHandlers.push(func);
}

///////////////////////////////////////////////////////////////////////////////
// HELPERS
///////////////////////////////////////////////////////////////////////////////

function onFeedback(deviceInfos) {
	// console.log('Feedback service, number of devices to remove: ' + deviceInfos.length);

	if (deviceInfos.length == 0) return;
	var tokensToRemove = deviceInfos.map(function(deviceInfo) {
    return deviceInfo.device.token.toString('hex');
  });
	feedbackHandlers.forEach(function(func){
		func([/* no tokens to update */], tokensToRemove);
	});
}

function onTransmissionError(errorCode, notification, recipient) {
  console.error('Error while pushing to APN: ' + errorCode);

  if(errorCode === 8 && recipient.token) {
		// Invalid token => remove device
    // console.log('Invalid token: removing device ' + token);

    var token = recipient.token.toString('hex').toUpperCase();

    feedbackHandlers.forEach(function(func){
			func([token]);
		});
  }
}


module.exports = {
    send: send,
    addFeedbackHandler: addFeedbackHandler
}

push.js

var config = require('../config.js');
global.Promise = require("bluebird");

var pushAndroid = require('./push.android');
var pushIos = require('./push.ios');

var Tramesa = require('../models/tramesa.js');
var Missatge = require('../models/missatge.js');
var Usuari = require('../models/usuari.js');

pushAndroid.addFeedbackHandler(onDeviceError);
pushIos.addFeedbackHandler(onDeviceError);

function log(...args){
	console.log((new Date()).toJSON(), '|', ...args);
}

function logError(...args){
	console.error((new Date()).toJSON(), '|', ...args);
}

////////////////////////////
// DELIVERY
////////////////////////////

function deliverTramesa(tramesa, notificacions){
	return Promise.try(function(){
		if(!tramesa) throw "No s'ha pogut accedir a la tramesa";

		return Tramesa.findByIdAndUpdate(tramesa._id, {estat: 'enviant'}).exec()
	})
	.then(function(){
		if(notificacions && notificacions.length) return notificacions;
		// WARNING idTramesa in some places
		return Missatge.find({tramesa: tramesa._id, enviat: null, llegit: null, retries: {$gt: 0}}).lean().exec()
	})
	.then(function(notificacions){
		if(!notificacions) throw new Error("No s'ha pogut accedir a la tramesa");
		var connectionErrors = 0;
		var payload = {idmissatge : tramesa._id + ""};

		return Promise.map(notificacions, function(notificacio){
			if(!notificacio) return;

			switch(notificacio.platform){
				case 'android':
				case 'Android':
					return pushAndroid.send([notificacio.pushToken], tramesa.missatge, payload, tramesa.expiracio, 1)
					.then(res => {
							if(res.failures) {
								return Missatge.findByIdAndUpdate(notificacio._id, {enviat: null, error: new Date(), retries: 0}).exec();
							}
							return Missatge.findByIdAndUpdate(notificacio._id, {enviat: new Date()}).exec()
							/* Usuari.findByIdAndUpdate(missatge.idReceptor, {$inc: {unreadCounter: 1}}) */
						}
					)
					.catch(err => {
						connectionErrors++;
						logError("Error d'entrega:", err, 'Tramesa', tramesa._id);
						return Missatge.findByIdAndUpdate(notificacio._id, {enviat: null, error: new Date(), $inc: {retries: -1}}).exec();
					});

				case 'iphone':
				case 'iPhone':
				case 'ios':
				case 'iOS':
					return pushIos.send([notificacio.pushToken], tramesa.missatge, payload, tramesa.expiracio, 1)
					.then(() => {
							return Missatge.findByIdAndUpdate(notificacio._id, {enviat: new Date()}).exec()
							/* Usuari.findByIdAndUpdate(missatge.idReceptor, {$inc: {unreadCounter: 1}}) */
						}
					)
					.catch(err => {
						connectionErrors++;
						logError("Error d'entrega:", err, 'Tramesa', tramesa._id);
						return Missatge.findByIdAndUpdate(notificacio._id, {enviat: null, error: new Date(), $inc: {retries: -1}}).exec();
					});

				default:
					throw new Error('Plataforma no suportada');
			}
		}, {concurrency: 30})
		.then(function(){
			return connectionErrors;
		});
	})
	.then(function(connectionErrors){
		// DONE
		if(connectionErrors > 0)
			return Tramesa.findByIdAndUpdate(tramesa._id, {estat: 'incompleta'}).lean().exec();
		else
			return Tramesa.findByIdAndUpdate(tramesa._id, {estat: 'enviat'}).lean().exec();
	});
}

function onDeviceError(tokensToUpdate, tokensToRemove){
	Promise.map(tokensToRemove || [], token => Usuari.findOneAndUpdate({pushToken: new RegExp(token)}, {pushToken: null}).lean().exec() )
	.then(() => log("Removed user tokens", tokensToRemove) )
	.catch(err => logError("tokenRemove Error", err) );

	Promise.all(tokensToUpdate || [], upd => {
		return Usuari.findOne({pushToken: new RegExp(upd.from)}).lean().exec()
		.then(user => Usuari.findByIdAndUpdate(user._id, {pushToken: user.pushToken.substr(0, 4) + upd.to}).lean().exec() )
	})
	.then(() => tokensToUpdate.length > 0 && log("Updated user tokens", tokensToUpdate) )
	.catch(err => logError("tokenUpdate Error", err) );
}


////////////////////////////
// LIFECYCLE
////////////////////////////

function pushInit(){
	Tramesa.find({estat: 'enviant'}).exec()
	.then(function(trameses){
		if(!trameses || typeof trameses != "object") return log("Tramesa.find retorna", typeof trameses );

		return Promise.all(trameses.map(function(tram){
			log("Iniciant entregues de la tramesa", tram._id);

			return deliverTramesa(tram);
		}));
	})
	.then(function(){
		log("Sistema de notificacions push iniciat [ OK ]");
	})
	.catch(function(err){
		return log("ERROR CONTINUANT ELS ENVIAMENTS PENDENTS", err);
	});

	// Periodic
	setInterval(resendIncomplete, config.PUSH_RETRY_INTERVAL);
	resendIncomplete();

	// AUTO CLEAN OLD
	setInterval(cleanTrameses, config.TRAMESA_CLEAN_INTERVAL);
	cleanTrameses();
}

function resendIncomplete(){
	Tramesa.find({estat: 'incompleta'}).exec()
	.then(function(trameses){
		if(!trameses || typeof trameses != "object") return logError("Tramesa.find retorna", typeof trameses );
		else if(!trameses.length) return;

		return Promise.map(trameses, function(tram){
			log("Reintentant entregues de la tramesa", tram._id);

			return deliverTramesa(tram);
		}, {concurrency: 30})
		.then(function(){
			log("Notificacions pendents reintentades [ OK ]");
		});
	})
	.catch(function(err){
		return log("ERROR REINTENTANT MISSATGES PENDENTS", err);
	});
}

function cleanTrameses(){
	// REMOVE
	Tramesa.find({expiracio: {$lt: new Date()}}).select('_id').exec()
	.then(function(trams){
		return Promise.map(trams, function(t){
			return Missatge.remove({tramesa: t._id});  // WARNING idTramesa in some places
		})
		.then(function(){
			return Tramesa.remove({_id: {$in: trams.map(function(tram){return tram._id;})}});
		});
	})
	.catch(function(err){
		return log("ERROR NETEJANT MISSATGES ANTICS", err); // ??
	});
}

// INIT

pushInit();

module.exports = deliverTramesa;