Merge pull request #3219 from Countly/hotfix/push

Hotfix/push

Author: ar2rsawseen

plugins/push/api/api-auto.js

@@ -53,8 +53,8 @@ module.exports.autoOnCohort = function(entry, cohort, uids) {
                         logCohorts.d('processing %s %s, result: %j', typ, msg._id, result);
                         if (result.total) {
                             return msg.update({$inc: {'result.total': result.total}}, () => {
-                                result.total += result.total;
-                            }).then(() => Audience.resetQueue(result.next));
+                                msg.result.total += result.total;
+                            });
                         }
                     }).then(() => {
                         logCohorts.d('done processing %s %s', typ, msg._id);

plugins/push/api/jobs/util/connector.js

@@ -1,5 +1,5 @@
 const { DoFinish } = require('./do_finish'),
-    { Message, State, Status, Creds, pools, FRAME, PushError, SendError, ERROR } = require('../../send'),
+    { Message, State, Status, Creds, pools, FRAME, PushError, SendError, ERROR, MAX_RUNS } = require('../../send'),
     { FRAME_NAME } = require('../../send/proto');
 
 /**
@@ -21,6 +21,7 @@ class Connector extends DoFinish {
         this.db = db;
         this.state = state;
         this.limit = state.cfg.pool.pushes;
+        this.connects = [];
         this.resetErrors();
     }
 
@@ -31,6 +32,7 @@ class Connector extends DoFinish {
         this.noApp = new SendError('NoApp', ERROR.DATA_COUNTLY);
         this.noCreds = new SendError('NoCredentials', ERROR.DATA_COUNTLY);
         this.expiredCreds = new SendError('ExpiredCreds', ERROR.CONNECTION_PROVIDER);
+        this.tooLateToSend = new SendError('TooLateToSend', ERROR.DATA_COUNTLY);
         this.noMessage = {}; // {mid: [push, push, push, ...]}
         this.noMessageBytes = 0;
     }
@@ -141,10 +143,17 @@ class Connector extends DoFinish {
         else { // all good with credentials, now let's check messages
             if (!message.is(State.Streaming)) {
                 let date = new Date(),
-                    update = {$set: {status: Status.Sending, 'info.startedLast': date}, $bit: {state: {or: State.Streaming}}};
+                    update = {$set: {status: Status.Sending, 'info.startedLast': date}, $bit: {state: {or: State.Streaming}}},
+                    run = message.result.startRun(new Date());
                 if (!message.info.started) {
                     update.$set['info.started'] = date;
                 }
+                if (message.result.lastRuns.length === MAX_RUNS) {
+                    update.$set['result.lastRuns'] = message.result.lastRuns;
+                }
+                else {
+                    update.$push = {'result.lastRuns': run};
+                }
                 message.updateAtomically({_id: message._id, state: message.state}, update)
                     .then(ok => {
                         if (ok) {
@@ -155,6 +164,11 @@ class Connector extends DoFinish {
                         }
                     }, callback);
             }
+            else if (push._id.getTimestamp().getTime() < Date.now() - 3600000) {
+                this.tooLateToSend.addAffected(push._id, 1);
+                this.do_flush(callback, true);
+                return;
+            }
             else {
                 let creds = app.creds[push.p],
                     pid = pools.id(creds.hash, push.p, push.f);
@@ -169,21 +183,24 @@ class Connector extends DoFinish {
                 }
                 else { // no connection yet
                     this.log.i('Connecting %s', pid);
-                    pools.connect(push.a, push.p, push.f, creds, this.state, this.state.cfg).then(valid => {
+                    let connect = pools.connect(push.a, push.p, push.f, creds, this.state, this.state.cfg).then(valid => {
                         if (valid) {
                             this.log.i('Connected %s', pid);
                         }
                         else {
                             app.creds[push.p] = null;
                         }
+                        this.connects = this.connects.filter(c => c !== connect);
                         callback(null, push);
                     }, err => {
                         this.log.i('Failed to connect %s', pid, err);
                         app.creds[push.p] = null;
+                        this.connects = this.connects.filter(c => c !== connect);
                         this.do_transform(push, encoding, callback);
                         // this.discardedByAppOrCreds.push(push._id);
                         // this.do_flush(callback, true);
                     });
+                    this.connects.push(connect);
                 }
             }
         }
@@ -196,7 +213,7 @@ class Connector extends DoFinish {
      * @param {boolean} ifNeeded true if we only need to flush `discarded` when it's length is over `limit`
      */
     do_flush(callback, ifNeeded) {
-        let total = this.noMessageBytes + this.noApp.affectedBytes + this.noCreds.affectedBytes + this.expiredCreds.affectedBytes;
+        let total = this.noMessageBytes + this.noApp.affectedBytes + this.noCreds.affectedBytes + this.expiredCreds.affectedBytes + this.tooLateToSend.affectedBytes;
         this.log.d('in connector do_flush, total', total);
 
         if (ifNeeded && !this.flushed && (!total || total < this.limit)) {
@@ -270,6 +287,10 @@ class Connector extends DoFinish {
             this.push({frame: FRAME.RESULTS | FRAME.ERROR, payload: this.expiredCreds});
         }
 
+        if (this.tooLateToSend.hasAffected) {
+            this.push({frame: FRAME.RESULTS | FRAME.ERROR, payload: this.tooLateToSend});
+        }
+
         this.resetErrors();
 
         callback();
@@ -375,7 +396,7 @@ class Connector extends DoFinish {
      * @param {function} callback callback function
      */
     do_final(callback) {
-        callback();
+        Promise.all(this.connects).then(() => callback(), () => callback());
     }
 
 

plugins/push/api/jobs/util/resultor.js

@@ -113,6 +113,8 @@ class Resultor extends DoFinish {
 
                         if (msg) {
                             result = msg.result;
+                            result.lastRun.processed++;
+                            result.lastRun.errored++;
                         }
                         else {
                             result = this.noMessage[m] || (this.noMessage[m] = new Result());
@@ -166,6 +168,7 @@ class Resultor extends DoFinish {
 
                     if (m) {
                         result = m.result;
+                        result.lastRun.processed++;
                     }
                     else {
                         result = this.noMessage[m] || (this.noMessage[m] = new Result());
@@ -262,6 +265,13 @@ class Resultor extends DoFinish {
                     result = this.noMessage[m] || (this.noMessage[m] = new Result());
                 }
                 result.processed[m] += mids[mid];
+
+                let run = result.lastRun;
+                if (run) {
+                    run.processed += mids[mid];
+                    run.errored += mids[mid];
+                }
+
                 result.pushError(error);
                 this.data.decSending(mid);
             }
@@ -286,6 +296,8 @@ class Resultor extends DoFinish {
 
         let updates = {},
             promises = this.data.messages().map(m => {
+                m.result.lastRun.ended = new Date();
+
                 if (this.data.isSending(m.id)) {
                     this.log.d('message %s is still in processing (%d out of %d)', m.id, m.result.processed, m.result.total);
                     return m.save();

plugins/push/api/send/data/index.js

@@ -5,7 +5,7 @@ const { S, State, Status, STATUSES, TriggerKind, MEDIA_MIME_ALL, MEDIA_MIME_IOS,
     { Creds } = require('./creds'),
     { Filter } = require('./filter'),
     { Content } = require('./content'),
-    { Result } = require('./result'),
+    { Result, MAX_ERRORS, MAX_RUNS } = require('./result'),
     { Message } = require('./message'),
     { Template } = require('./template'),
     dbext = require('./db');
@@ -28,9 +28,12 @@ module.exports = {
 
     Filter,
     Content,
-    Result,
     Info,
 
+    Result,
+    MAX_ERRORS,
+    MAX_RUNS,
+
     Message,
 
     Template,

plugins/push/api/send/data/result.js

@@ -1,7 +1,9 @@
 'use strict';
 
 const { PushError, ERROR } = require('./error'),
-    { toDate, Validatable } = require('./const');
+    { toDate, Validatable } = require('./const'),
+    MAX_RUNS = 10,
+    MAX_ERRORS = 10;
 
 /**
  * Message sending result
@@ -247,7 +249,7 @@ class Result extends Validatable {
         if (!this._data.lastErrors) {
             this._data.lastErrors = [];
         }
-        while (this._data.lastErrors.length >= 10) {
+        while (this._data.lastErrors.length >= MAX_ERRORS) {
             this._data.lastErrors.shift();
         }
         this._data.lastErrors.push(error);
@@ -280,6 +282,15 @@ class Result extends Validatable {
         return this._data.errors[code] = (this._data.errors[code] || 0) + count;
     }
 
+    /**
+     * Getter for last of lastRuns
+     * 
+     * @returns {object|undefined} last run object with start, processed, errored & end? keys
+     */
+    get lastRun() {
+        return this._data.lastRuns && this._data.lastRuns[this._data.lastRuns.length - 1] || undefined;
+    }
+
     /**
      * Getter for lastRuns
      * 
@@ -292,16 +303,19 @@ class Result extends Validatable {
     /**
      * Add another run to lastRuns
      * 
-     * @param {object} run the run to push
+     * @param {Date} date date of run start
+     * @returns {object} run object with start, processed & errored props
      */
-    pushRun(run) {
+    startRun(date) {
         if (!this._data.lastRuns) {
             this._data.lastRuns = [];
         }
-        while (this._data.lastRuns.length >= 10) {
+        while (this._data.lastRuns.length >= MAX_RUNS) {
             this._data.lastRuns.shift();
         }
+        let run = {start: date, processed: 0, errored: 0};
         this._data.lastRuns.push(run);
+        return run;
     }
 
     /**
@@ -437,5 +451,5 @@ class Result extends Validatable {
 
 
 
-module.exports = { Result };
+module.exports = { Result, MAX_RUNS, MAX_ERRORS };
 

plugins/push/api/send/platforms/i.js

@@ -559,8 +559,8 @@ class APN extends Base {
         }
         if (this.creds instanceof CREDS.apn_token) {
             this.headersFirst.authorization = this.headersSecond.authorization = this.creds.bearer;
-            this.headersFirst['apns-topic'] = this.headersSecond['apns-topic'] = this.creds._data.bundle;
         }
+        this.headersFirst['apns-topic'] = this.headersSecond['apns-topic'] = this.creds._data.bundle;
     }
 
     /**

plugins/push/frontend/public/localization/push.properties

@@ -397,6 +397,7 @@ push-notification.error-code.NoApp.desc = App not found when sending the notific
 push-notification.error-code.NoMessage.desc = The message was not found when sending the notification.
 push-notification.error-code.NoCredentials.desc = Push notification credentials were not found when sending the notification.
 push-notification.error-code.NoConnection.desc = Failed to connect to push notifications provider, please check APN / FCM / HPK are available from Countly host. In case you use proxy server, please check it's up and running.
+push-notification.error-code.TooLateToSend.desc = Countly was unable to send these notifications in time (1 hour from scheduled date).
 push-notification.error-code.ExpiredCreds.desc = Push Notification credentials have probably expired. Please upload new credentials.
 push-notification.error-code.BadDeviceToken.desc = The push token received from your app by Countly Server was rejected by APNS as invalid. Please make sure you set `pushTestMode` property on the SDK's initial configuration correctly. Also please make sure provisioning profile (Development or Distribution) is valid, bundle ID is correct and entitlements are properly set.
 push-notification.error-code.MissingTopic.desc = The server failed to parse the certificate, please ensure you use universal certificate and contact support if you do.