Merge pull request #5940 from Countly/topEventJob

[TopEventsJob] Job Improvements

Author: ArtursKadikis

api/jobs/topEvents.js

@@ -19,8 +19,8 @@ class TopEventsJob extends job.Job {
     /**
      * TopEvents initialize function
      */
-    init() {
-        this.getAllApps();
+    async init() {
+        return this.getAllApps();
     }
 
     /**
@@ -144,6 +144,7 @@ class TopEventsJob extends job.Job {
         }
         catch (error) {
             log.e("TopEvents Job has a error: ", error);
+            throw error;
         }
     }
 
@@ -157,7 +158,18 @@ class TopEventsJob extends job.Job {
         const encodedData = this.encodeEvents(data);
         const timeSecond = this.timeSecond();
         const currentPeriood = this.mutatePeriod(period);
-        await new Promise((res, rej) => common.db.collection(TopEventsJob.COLLECTION_NAME).insert({ app_id: _id, ts: timeSecond, period: currentPeriood, data: encodedData, totalCount: totalCount, prevTotalCount: prevTotalCount, totalSum: totalSum, prevTotalSum: prevTotalSum, totalDuration: totalDuration, prevTotalDuration: prevTotalDuration, prevSessionCount: sessionData.prevSessionCount, totalSessionCount: sessionData.totalSessionCount, prevUsersCount: usersData.prevUsersCount, totalUsersCount: usersData.totalUsersCount }, (error, records) => !error && records ? res(records) : rej(error)));
+        await new Promise((res, rej) => common.db.collection(TopEventsJob.COLLECTION_NAME).findOneAndReplace(
+            {
+                app_id: _id, period: currentPeriood,
+            },
+            {
+                app_id: _id, ts: timeSecond, period: currentPeriood, data: encodedData, totalCount: totalCount, prevTotalCount: prevTotalCount, totalSum: totalSum, prevTotalSum: prevTotalSum, totalDuration: totalDuration, prevTotalDuration: prevTotalDuration, prevSessionCount: sessionData.prevSessionCount, totalSessionCount: sessionData.totalSessionCount, prevUsersCount: usersData.prevUsersCount, totalUsersCount: usersData.totalUsersCount
+            },
+            {
+                upsert: true
+            },
+            (error, records) => !error && records ? res(records) : rej(error))
+        );
     }
 
     /**
@@ -169,7 +181,6 @@ class TopEventsJob extends job.Job {
         const getEvents = await new Promise((res, rej) => common.db.collection("events").findOne({ _id: app._id }, (errorEvents, result) => errorEvents ? rej(errorEvents) : res(result)));
         if (getEvents && 'list' in getEvents) {
             const eventMap = this.eventsFilter(getEvents.list);
-            await new Promise((res, rej) => common.db.collection(TopEventsJob.COLLECTION_NAME).remove({ app_id: app._id }, (error, result) => error ? rej(error) : res(result)));
             if (eventMap && eventMap instanceof Array) {
                 for (const period of TopEventsJob.PERIODS) {
                     const data = {};
@@ -211,9 +222,14 @@ class TopEventsJob extends job.Job {
      * @param {Db} db connection
      * @param {done} done callback
      */
-    run(db, done) {
-        this.init();
-        done();
+    async run(db, done) {
+        try {
+            await this.init();
+            done();
+        }
+        catch (error) {
+            done(error);
+        }
     }
 }