feat(aggregation): Reads aggregations from secondary. (#1095)

Author: ryasmi

lib/models/statement.js

@@ -89,7 +89,10 @@ schema.plugin(filterByOrg);
 schema.plugin(addCRUDFunctions);
 
 const streamAggregation = ({ pipeline, skip, limit, batchSize, maxTimeMS, maxScan }) => {
- let query = Statement.aggregate(pipeline).allowDiskUse(ALLOW_AGGREGATION_DISK_USE);
+ let query = Statement
+ .aggregate(pipeline)
+ .read('secondaryPreferred')
+ .allowDiskUse(ALLOW_AGGREGATION_DISK_USE);
  if (skip !== -1) query = query.skip(skip);
  if (limit !== -1) query = query.limit(limit);
  if (!query.options) {
@@ -153,19 +156,19 @@ const setCachedAggregation = ({ client, dataKey, isRunningKey, stream }) =>
  * @return {Stream} stream
  */
 schema.statics.aggregateByAuth = function aggregateByAuth(
-  authInfo,
-  pipeline = [],
-  {
-  skip = 0,
-  limit = -1,
-  cache = false,
-  batchSize = 100,
-  getStream = false,
-  maxTimeMS = MAX_TIME_MS,
-  maxScan = MAX_SCAN,
-  },
-  cb = () => {}
- ) {
+ authInfo,
+ pipeline = [],
+ {
+ skip = 0,
+ limit = -1,
+ cache = false,
+ batchSize = 100,
+ getStream = false,
+ maxTimeMS = MAX_TIME_MS,
+ maxScan = MAX_SCAN,
+ },
+ cb = () => { }
+) {
  return parseQuery(pipeline, {
  organisation: getOrgFromAuthInfo(authInfo)
  }).then(async (parsedPipeline) => {
@@ -227,9 +230,9 @@ schema.statics.aggregateByAuth = function aggregateByAuth(
  }
  return dataKeyTTL >= 5 ? cachedStreamPromise : streamPromise;
  })
- .then((stringResult) => {
- cb(null, stringResult);
- });
+  .then((stringResult) => {
+  cb(null, stringResult);
+  });
  }).catch(cb);
 };