Push overhaul

package.json

@@ -1,6 +1,6 @@
 {
   "name": "parse-server",
-  "version": "2.6.5",
+  "version": "3.0.0-alpha.1",
   "description": "An express module providing a Parse-compatible API server",
   "main": "lib/index.js",
   "repository": {

spec/PushController.spec.js

@@ -477,17 +477,17 @@ describe('PushController', () => {
           return spy.calls.all()[callIndex].args[0].object;
         }
         expect(spy).toHaveBeenCalled();
-        expect(spy.calls.count()).toBe(4);
+        expect(spy.calls.count()).toBe(3);
         const allCalls = spy.calls.all();
         allCalls.forEach((call) => {
           expect(call.args.length).toBe(2);
           const object = call.args[0].object;
           expect(object instanceof Parse.Object).toBe(true);
         });
         expect(getPushStatus(0).get('status')).toBe('pending');
-        expect(getPushStatus(1).get('status')).toBe('running');
+        expect(getPushStatus(1).get('status')).toBe('succeeded');
         expect(getPushStatus(1).get('numSent')).toBe(0);
-        expect(getPushStatus(2).get('status')).toBe('running');
+        expect(getPushStatus(2).get('status')).toBe('succeeded');
         expect(getPushStatus(2).get('numSent')).toBe(10);
         expect(getPushStatus(2).get('numFailed')).toBe(5);
         // Those are updated from a nested . operation, this would
@@ -498,7 +498,6 @@ describe('PushController', () => {
         expect(getPushStatus(2).get('sentPerType')).toEqual({
           ios: 10
         });
-        expect(getPushStatus(3).get('status')).toBe('succeeded');
       })
       .then(done).catch(done.fail);
   });
@@ -805,7 +804,7 @@ describe('PushController', () => {
       return query.find({useMasterKey: true}).then((results) => {
         expect(results.length).toBe(1);
         const pushStatus = results[0];
-        expect(pushStatus.get('status')).not.toBe('scheduled');
+        expect(pushStatus.get('status')).toBe('succeeded');
         done();
       });
     }).catch((err) => {
@@ -871,7 +870,7 @@ describe('PushController', () => {
       return query.find({useMasterKey: true}).then((results) => {
         expect(results.length).toBe(1);
         const pushStatus = results[0];
-        expect(pushStatus.get('status')).toBe('scheduled');
+        expect(pushStatus.get('status')).toBe('succeeded');
       });
     }).then(done).catch(done.err);
   });
@@ -1248,7 +1247,7 @@ describe('PushController', () => {
           return q.get(pushStatusId, {useMasterKey: true});
         })
         .then((pushStatus) => {
-          expect(pushStatus.get('status')).toBe('scheduled');
+          expect(pushStatus.get('status')).toBe('pending');
           expect(pushStatus.get('pushTime')).toBe('2017-09-06T17:14:01.048');
         })
         .then(done, done.fail);

spec/PushWorker.spec.js

@@ -157,6 +157,138 @@ describe('PushWorker', () => {
       expect(PushUtils.bodiesPerLocales({where: {}})).toEqual({default: {where: {}}});
       expect(PushUtils.groupByLocaleIdentifier([])).toEqual({default: []});
     });
+
+    it('should propely apply translations strings', () => {
+      const bodies = PushUtils.bodiesPerLocales({
+        data: {
+          alert: 'Yo!',
+        },
+        translation: {
+          'fr': 'frenchy!',
+          'en': 'english',
+        }
+      }, ['fr', 'en']);
+      expect(bodies).toEqual({
+        fr: {
+          data: {
+            alert: 'frenchy!',
+          }
+        },
+        en: {
+          data: {
+            alert: 'english',
+          }
+        },
+        default: {
+          data: {
+            alert: 'Yo!'
+          }
+        }
+      });
+    });
+
+    it('should propely apply translations objects', () => {
+      const bodies = PushUtils.bodiesPerLocales({
+        data: {
+          alert: 'Yo!',
+          badge: 'Increment',
+        },
+        translation: {
+          'fr': { alert: 'frenchy!', title: 'yolo' },
+          'en': { alert: 'english', badge: 2, other: 'value' },
+        }
+      }, ['fr', 'en']);
+      expect(bodies).toEqual({
+        fr: {
+          data: {
+            alert: 'frenchy!',
+            title: 'yolo',
+          }
+        },
+        en: {
+          data: {
+            alert: 'english',
+            badge: 2,
+            other: 'value'
+          }
+        },
+        default: {
+          data: {
+            alert: 'Yo!',
+            badge: 'Increment',
+          }
+        }
+      });
+    });
+
+    it('should propely override alert-lang with translations', () => {
+      const bodies = PushUtils.bodiesPerLocales({
+        data: {
+          alert: 'Yo!',
+          badge: 'Increment',
+          'alert-fr': 'Yolo!',
+        },
+        translation: {
+          'fr': { alert: 'frenchy!', title: 'yolo' },
+          'en': { alert: 'english', badge: 2, other: 'value' },
+        }
+      }, ['fr', 'en']);
+      expect(bodies).toEqual({
+        fr: {
+          data: {
+            alert: 'frenchy!',
+            title: 'yolo',
+          }
+        },
+        en: {
+          data: {
+            alert: 'english',
+            badge: 2,
+            other: 'value'
+          }
+        },
+        default: {
+          data: {
+            alert: 'Yo!',
+            badge: 'Increment',
+          }
+        }
+      });
+    });
+
+    it('should propely override alert-lang with translations strings', () => {
+      const bodies = PushUtils.bodiesPerLocales({
+        data: {
+          alert: 'Yo!',
+          badge: 'Increment',
+          'alert-fr': 'Yolo!',
+          'alert-en': 'Yolo!'
+        },
+        translation: {
+          'fr': 'frenchy',
+        }
+      }, ['fr', 'en']);
+      expect(bodies).toEqual({
+        fr: {
+          data: {
+            alert: 'frenchy',
+            badge: 'Increment',
+          }
+        },
+        en: {
+          data: {
+            alert: 'Yolo!',
+            badge: 'Increment'
+          }
+        },
+        default: {
+          data: {
+            alert: 'Yo!',
+            badge: 'Increment',
+          }
+        }
+      });
+    });
   });
 
   describe('pushStatus', () => {
@@ -276,7 +408,6 @@ describe('PushWorker', () => {
           'failedPerType.ios': { __op: 'Increment', amount: 1 },
           [`sentPerUTCOffset.${UTCOffset}`]: { __op: 'Increment', amount: 1 },
           [`failedPerUTCOffset.${UTCOffset}`]: { __op: 'Increment', amount: 1 },
-          count: { __op: 'Increment', amount: -1 }
         });
         const query = new Parse.Query('_PushStatus');
         return query.get(handler.objectId, { useMasterKey: true });
@@ -354,7 +485,6 @@ describe('PushWorker', () => {
           'failedPerType.ios': { __op: 'Increment', amount: 1 },
           [`sentPerUTCOffset.${UTCOffset}`]: { __op: 'Increment', amount: 1 },
           [`failedPerUTCOffset.${UTCOffset}`]: { __op: 'Increment', amount: 1 },
-          count: { __op: 'Increment', amount: -1 }
         });
         done();
       });

spec/rest.spec.js

@@ -626,6 +626,31 @@ describe('rest update', () => {
   });
 });
 
+describe('rest each', () => {
+  it('should iterate through all results', () => {
+    const className = 'Yolo';
+    const config = Config.get('test');
+    const master = auth.master(config);
+    let promise = Promise.resolve();
+    let done = 0;
+    while (done != 10) {
+      done++;
+      promise = promise.then(() => {
+        return rest.create(config, auth, className, {});
+      });
+    }
+    return promise.then(() => {
+      let seen = 0;
+      // use limit 1 to get them 1 by one
+      return rest.each(config, master, className, {}, {limit: 1}, () => {
+        seen++;
+      }).then(() => {
+        expect(seen).toBe(10);
+      })
+    }).then(done, done.fail);
+  });
+})
+
 describe('read-only masterKey', () => {
   it('properly throws on rest.create, rest.update and rest.del', () => {
     const config = Config.get('test');

src/Controllers/PushController.js

@@ -87,6 +87,8 @@ export class PushController {
         return Promise.resolve();
       }
       return config.pushControllerQueue.enqueue(body, where, config, auth, pushStatus);
+    }).then(() => {
+      return pushStatus.complete();
     }).catch((err) => {
       return pushStatus.fail(err).then(() => {
         throw err;

src/Push/PushQueue.js

@@ -2,9 +2,11 @@ import { ParseMessageQueue }      from '../ParseMessageQueue';
 import rest                       from '../rest';
 import { applyDeviceTokenExists } from './utils';
 import Parse from 'parse/node';
+import logger from '../logger';
 
 const PUSH_CHANNEL = 'parse-server-push';
 const DEFAULT_BATCH_SIZE = 100;
+const DEFAULT_QUERY_BATCH_SIZE = 10000;
 
 export class PushQueue {
   parsePublisher: Object;
@@ -16,6 +18,7 @@ export class PushQueue {
   constructor(config: any = {}) {
     this.channel = config.channel || PushQueue.defaultPushChannel();
     this.batchSize = config.batchSize || DEFAULT_BATCH_SIZE;
+    this.installationsQueryBatchSize = config.installationsQueryBatchSize || DEFAULT_QUERY_BATCH_SIZE;
     this.parsePublisher = ParseMessageQueue.createPublisher(config);
   }
 
@@ -24,39 +27,44 @@ export class PushQueue {
   }
 
   enqueue(body, where, config, auth, pushStatus) {
-    const limit = this.batchSize;
-
     where = applyDeviceTokenExists(where);
-
-    // Order by objectId so no impact on the DB
-    const order = 'objectId';
     return Promise.resolve().then(() => {
-      return rest.find(config,
-        auth,
-        '_Installation',
-        where,
-        {limit: 0, count: true});
-    }).then(({results, count}) => {
-      if (!results || count == 0) {
+      const batches = [];
+      let currentBatch = [];
+      let total = 0;
+      const options = {
+        limit: this.installationsQueryBatchSize,
+        keys: 'objectId'
+      }
+      return rest.each(config, auth, '_Installation', where, options, (result) => {
+        total++;
+        currentBatch.push(result.objectId);
+        if (currentBatch.length == this.batchSize) {
+          batches.push(currentBatch);
+          currentBatch = [];
+        }
+      }, { useMasterKey: true, batchSize: 10000 }).then(() => {
+        if (currentBatch.length > 0) {
+          batches.push(currentBatch);
+        }
+        return Promise.resolve({ batches, total });
+      });
+    }).then(({ batches, total }) => {
+      if (total == 0) {
         return Promise.reject({error: 'PushController: no results in query'})
       }
-      pushStatus.setRunning(Math.ceil(count / limit));
-      let skip = 0;
-      while (skip < count) {
-        const query = { where,
-          limit,
-          skip,
-          order };
-
+      logger.verbose(`_PushStatus ${pushStatus.objectId}: sending push to installations with %d batches`, total);
+      batches.forEach((batch) => {
         const pushWorkItem = {
           body,
-          query,
+          query: {
+            where: { objectId: { '$in': batch }},
+          },
           pushStatus: { objectId: pushStatus.objectId },
           applicationId: config.applicationId
-        }
+        };
         this.parsePublisher.publish(this.channel, JSON.stringify(pushWorkItem));
-        skip += limit;
-      }
+      });
     });
   }
 }

src/Push/utils.js

@@ -52,14 +52,29 @@ export function stripLocalesFromBody(body) {
   return body;
 }
 
+export function translateBody(body, locale) {
+  body = deepcopy(body);
+  if (body.translation && body.translation[locale] && locale) {
+    const translation = body.translation[locale];
+    if (typeof translation === 'string') { // use translation
+      body.data = body.data || {};
+      body.data.alert = translation;
+    } else if (typeof translation === 'object') { // override tranlsation
+      body.data = translation;
+    }
+  }
+  delete body.translation;
+  return body;
+}
+
 export function bodiesPerLocales(body, locales = []) {
   // Get all tranformed bodies for each locale
   const result = locales.reduce((memo, locale) => {
-    memo[locale] = transformPushBodyForLocale(body, locale);
+    memo[locale] = translateBody(transformPushBodyForLocale(body, locale), locale);
     return memo;
   }, {});
   // Set the default locale, with the stripped body
-  result.default = stripLocalesFromBody(body);
+  result.default = translateBody(stripLocalesFromBody(body));
   return result;
 }