[job] to fix user merges

Author: ar2rsawseen

api/api.js

@@ -177,6 +177,7 @@ if (cluster.isMaster) {
         jobs.job('api:ping').replace().schedule('every 1 day');
         jobs.job('api:clear').replace().schedule('every 1 day');
         jobs.job('api:clearTokens').replace().schedule('every 1 day');
+        jobs.job('api:userMerge').replace().schedule('every 1 hour on the 10th min');
     }, 10000);
 } else {
     const taskManager = require('./utils/taskmanager.js');

api/jobs/userMerge.js

@@ -0,0 +1,43 @@
+'use strict';
+
+const job = require('../parts/jobs/job.js'),
+    async = require('async'),
+    moment = require('moment'),
+    plugins = require('../../plugins/pluginManager.js'),
+    log = require('../utils/log.js')('job:userMerge');
+
+class UserMergeJob extends job.Job {
+    run (db, done) {
+        log.d('Merging users ...');
+        var startTime = moment().subtract(1, 'hour').startOf('hour');
+        var endTime = moment(startTime).endOf("hour");
+        log.d("query from", startTime, "to", endTime);
+        function handleMerge(app, done){
+            db.collection('app_user_merges' + app._id).find({cd:{$gte: startTime.toDate(), $lte: endTime.toDate()}}).toArray(function(err, res){
+                if(!err && res && res.length){
+                    log.d('Found merges for '+app._id+' ', res);
+                    for(var i = 0; i < res.length; i++){
+                        if(res[i].merged_to){
+                            log.d('Dispatching', {app_id:app._id+"", oldUser:{uid:res[i]._id}, newUser:{uid:res[i].merged_to}});
+                            plugins.dispatch("/i/device_id", {app_id:app._id+"", oldUser:{uid:res[i]._id}, newUser:{uid:res[i].merged_to}});
+                        }
+                    }
+                }
+                done();
+            });
+        }
+        db.collection('apps').find({}).toArray(function (err, apps) {
+            if(!err && apps && apps.length){
+                async.forEach(apps, handleMerge, function(){
+                    log.d('Merging users finished ...');
+                    done();
+                });
+            }
+            else{
+                done(err);
+            }
+        });
+    }
+}
+
+module.exports = UserMergeJob;

api/parts/mgmt/apps.js

@@ -140,7 +140,7 @@ var appsApi = {},
             common.db.collection('app_users' + app.ops[0]._id).ensureIndex({"lac":1, "ls":1}, { background: true },function(err,res){});
             common.db.collection('app_users' + app.ops[0]._id).ensureIndex({"tsd":1}, { background: true },function(err,res){});
             common.db.collection('app_users' + app.ops[0]._id).ensureIndex({"did":1}, { background: true },function(err,res){});
-            common.db.collection('app_user_merges' + app.ops[0]._id).ensureIndex({cd: 1}, {expireAfterSeconds: 60*60*2, background: true},function(err,res){});
+            common.db.collection('app_user_merges' + app.ops[0]._id).ensureIndex({cd: 1}, {expireAfterSeconds: 60*60*3, background: true},function(err,res){});
             common.db.collection('metric_changes' + app.ops[0]._id).ensureIndex({ts:-1}, { background: true },function(err,res){});
             common.db.collection('metric_changes' + app.ops[0]._id).ensureIndex({uid:1}, { background: true },function(err,res){});
 			plugins.dispatch("/i/apps/create", {params:params, appId:app.ops[0]._id, data:newApp});
@@ -326,7 +326,7 @@ var appsApi = {},
                     common.db.collection('metric_changes' + appId).ensureIndex({uid:1}, { background: true },function(err,res){});
                 });
                 common.db.collection('app_user_merges' + appId).drop(function() {
-                    common.db.collection('app_user_merges' + appId).ensureIndex({cd: 1}, {expireAfterSeconds: 60*60*2, background: true},function(err,res){});
+                    common.db.collection('app_user_merges' + appId).ensureIndex({cd: 1}, {expireAfterSeconds: 60*60*3, background: true},function(err,res){});
                 });
                 if(params.qstring.args.period == "reset"){
                     plugins.dispatch("/i/apps/reset", {params:params, appId:appId, data:app}, deleteEvents);

bin/scripts/add_indexes.js

@@ -25,7 +25,7 @@ countlyDb.collection('apps').find({}).toArray(function (err, apps) {
         countlyDb.collection('app_users' + app._id).ensureIndex({"lac":1, "ls":1}, { background: true },cb);
         countlyDb.collection('app_users' + app._id).ensureIndex({"tsd":1}, { background: true },cb);
         countlyDb.collection('app_users' + app._id).ensureIndex({"did":1}, { background: true },cb);
-        countlyDb.collection('app_user_merges' + app._id).ensureIndex({cd: 1}, {expireAfterSeconds: 60*60*2, background: true},cb);
+        countlyDb.collection('app_user_merges' + app._id).ensureIndex({cd: 1}, {expireAfterSeconds: 60*60*3, background: true},cb);
         countlyDb.collection('metric_changes' + app._id).ensureIndex({ts:-1}, { background: true },cb);
         countlyDb.collection('metric_changes' + app._id).ensureIndex({uid:1}, { background: true },cb);
 	}