Push: Cleanup invalid device tokens (#4149)

* Adds collecting invalid / expired device tokens from GCM / APNS

* Cleanup invalid installations based on responses from the adapters

* Adds test for cleanup

* Adds tests for removal

spec/PushWorker.spec.js

@@ -1,6 +1,7 @@
 var PushWorker = require('../src').PushWorker;
 var PushUtils = require('../src/Push/utils');
 var Config = require('../src/Config');
+var { pushStatusHandler } = require('../src/StatusHandler');
 
 describe('PushWorker', () => {
   it('should run with small batch', (done) => {
@@ -156,4 +157,89 @@ describe('PushWorker', () => {
       expect(PushUtils.groupByLocaleIdentifier([])).toEqual({default: []});
     });
   });
+
+  describe('pushStatus', () => {
+    it('should remove invalid installations', (done) => {
+      const config = new Config('test');
+      const handler = pushStatusHandler(config);
+      const spy = spyOn(config.database, "update").and.callFake(() => {
+        return Promise.resolve();
+      });
+      handler.trackSent([
+        {
+          transmitted: false,
+          device: {
+            deviceToken: 1,
+            deviceType: 'ios',
+          },
+          response: { error: 'Unregistered' }
+        },
+        {
+          transmitted: true,
+          device: {
+            deviceToken: 10,
+            deviceType: 'ios',
+          },
+        },
+        {
+          transmitted: false,
+          device: {
+            deviceToken: 2,
+            deviceType: 'ios',
+          },
+          response: { error: 'NotRegistered' }
+        },
+        {
+          transmitted: false,
+          device: {
+            deviceToken: 3,
+            deviceType: 'ios',
+          },
+          response: { error: 'InvalidRegistration' }
+        },
+        {
+          transmitted: true,
+          device: {
+            deviceToken: 11,
+            deviceType: 'ios',
+          },
+        },
+        {
+          transmitted: false,
+          device: {
+            deviceToken: 4,
+            deviceType: 'ios',
+          },
+          response: { error: 'InvalidRegistration' }
+        },
+        {
+          transmitted: false,
+          device: {
+            deviceToken: 5,
+            deviceType: 'ios',
+          },
+          response: { error: 'InvalidRegistration' }
+        },
+        { // should not be deleted
+          transmitted: false,
+          device: {
+            deviceToken: 101,
+            deviceType: 'ios',
+          },
+          response: { error: 'invalid error...' }
+        }
+      ], true);
+      expect(spy).toHaveBeenCalled();
+      expect(spy.calls.count()).toBe(1);
+      const lastCall = spy.calls.mostRecent();
+      expect(lastCall.args[0]).toBe('_Installation');
+      expect(lastCall.args[1]).toEqual({
+        deviceToken: { '$in': [1,2,3,4,5] }
+      });
+      expect(lastCall.args[2]).toEqual({
+        deviceToken: { '__op': "Delete" }
+      });
+      done();
+    });
+  });
 });

src/Push/PushWorker.js

@@ -9,6 +9,7 @@ import { pushStatusHandler }  from '../StatusHandler';
 import * as utils             from './utils';
 import { ParseMessageQueue }  from '../ParseMessageQueue';
 import { PushQueue }          from './PushQueue';
+import logger                 from '../logger';
 
 function groupByBadge(installations) {
   return installations.reduce((map, installation) => {
@@ -80,6 +81,7 @@ export class PushWorker {
     }
 
     if (!utils.isPushIncrementing(body)) {
+      logger.verbose(`Sending push to ${installations.length}`);
       return this.adapter.send(body, installations, pushStatus.objectId).then((results) => {
         return pushStatus.trackSent(results);
       });

src/StatusHandler.js

@@ -162,12 +162,13 @@ export function pushStatusHandler(config, objectId = newObjectId(config.objectId
       {status: "running", updatedAt: new Date(), count });
   }
 
-  const trackSent = function(results) {
+  const trackSent = function(results, cleanupInstallations = process.env.PARSE_SERVER_CLEANUP_INVALID_INSTALLATIONS) {
     const update = {
       updatedAt: new Date(),
       numSent: 0,
       numFailed: 0
     };
+    const devicesToRemove = [];
     if (Array.isArray(results)) {
       results = flatten(results);
       results.reduce((memo, result) => {
@@ -181,6 +182,18 @@ export function pushStatusHandler(config, objectId = newObjectId(config.objectId
         if (result.transmitted) {
           memo.numSent++;
         } else {
+          if (result && result.response && result.response.error && result.device && result.device.deviceToken) {
+            const token = result.device.deviceToken;
+            const error = result.response.error;
+            // GCM errors
+            if (error === 'NotRegistered' || error === 'InvalidRegistration') {
+              devicesToRemove.push(token);
+            }
+            // APNS errors
+            if (error === 'Unregistered') {
+              devicesToRemove.push(token);
+            }
+          }
           memo.numFailed++;
         }
         return memo;
@@ -189,7 +202,7 @@ export function pushStatusHandler(config, objectId = newObjectId(config.objectId
     }
 
     logger.verbose(`_PushStatus ${objectId}: sent push! %d success, %d failures`, update.numSent, update.numFailed);
-
+    logger.verbose(`_PushStatus ${objectId}: needs cleanup`, { devicesToRemove });
     ['numSent', 'numFailed'].forEach((key) => {
       if (update[key] > 0) {
         update[key] = {
@@ -201,6 +214,14 @@ export function pushStatusHandler(config, objectId = newObjectId(config.objectId
       }
     });
 
+    if (devicesToRemove.length > 0 && cleanupInstallations) {
+      logger.info(`Removing device tokens on ${devicesToRemove.length} _Installations`);
+      database.update('_Installation', { deviceToken: { '$in': devicesToRemove }}, { deviceToken: {"__op": "Delete"} }, {
+        acl: undefined,
+        many: true
+      });
+    }
+
     return handler.update({ objectId }, update).then((res) => {
       if (res && res.count === 0) {
         return this.complete();