Allow aggregating/counting via REST api

lib/aggregate.js

@@ -0,0 +1,30 @@
+
+
+function create (conf, api) {
+
+  var template = function ( ) {
+    return [
+        {
+          $group: {
+            _id: null
+          , count: { $sum: 1 }
+          }
+        }
+      ];
+  }
+  // var collection = api( );
+  function aggregate (opts, done) {
+    var query = api.query_for(opts);
+
+    var pipeline = (conf.pipeline || [ ]).concat(opts.pipeline || [ ]);
+    var groupBy = [ {$match: query } ].concat(pipeline).concat(template( ));
+    console.log("AGGREGATE", groupBy);
+    api( ).aggregate(groupBy, done);
+  }
+
+  return aggregate;
+
+}
+
+module.exports = create;
+

lib/api/entries/index.js

@@ -341,6 +341,22 @@ function configure (app, wares, ctx) {
       });
     }
 
+    function count_records (req, res, next) {
+      var query = req.query;
+      var storage = req.storage || ctx.entries;
+      storage.aggregate(query, function payload (err, entries) {
+        // assign payload
+        res.entries = entries;
+        res.entries_err = err;
+        return next(err);
+      });
+    }
+
+    function format_results (req, res, next) {
+      res.json(res.entries);
+      next( );
+    }
+
     /**
       * @function delete_records
       * Delete entries.  The query logic works the same way as find/list.  This
@@ -555,6 +571,8 @@ curl -s -g 'http://localhost:1337/api/v1/times/20{14..15}/T{13..18}:{00..15}'.js
       */
     api.get('/times/:prefix?/:regex?', prep_storage, prep_pattern_field, prep_patterns, prep_patterns, query_models, format_entries);
 
+    api.get('/count/:storage/where', prep_storage, count_records, format_results);
+
     /**
       * @module get#/slice/:storage/:field/:type/:prefix/:regex
       * @routed

lib/api/index.js

@@ -50,6 +50,7 @@ function create (env, ctx) {
   app.all('/echo/*', entriesRouter);
   app.all('/times/*', entriesRouter);
   app.all('/slice/*', entriesRouter);
+  app.all('/count/*', entriesRouter);
 
   app.all('/treatments*', require('./treatments/')(app, wares, ctx));
   app.all('/profile*', require('./profile/')(app, wares, ctx));

lib/data/endpoints.js

@@ -35,6 +35,13 @@ function get_ddata (req, res, next) {
   });
 }
 
+function ensure_at (req, res, next) {
+  if (!req.at) {
+    req.at = new Date( );
+  }
+  next( );
+}
+
 function format_result (req, res, next) {
   res.json(res.data);
   next( );
@@ -45,7 +52,6 @@ function format_result (req, res, next) {
   * Configure the ddata endpoints module, given an existing express app, common
   * middlewares, and the global app's `ctx`.
   * @param Express app  The express app we'll mount onto.
-  * @param Object wares Common middleware used by lots of apps.
   * @param Object ctx The global ctx with all modules, storage, and event buses
   * configured.
   */
@@ -57,20 +63,21 @@ function configure (app, ctx) {
       api = express.Router( )
     ;
 
-    api.param('at', function (req, res, next, at) {
-      req.at = get_time_spec(at);
-      next( );
-    });
+  api.param('at', function (req, res, next, at) {
+    req.at = get_time_spec(at);
+    next( );
+  });
 
-    api.use(function (req, res, next) {
-      req.ctx = ctx;
-      next( );
-    });
+  api.use(function (req, res, next) {
+    req.ctx = ctx;
+    next( );
+  });
 
-    // api.get('/ctx/:at', get_context, format_result);
-    api.get('/at/:at', get_ddata, format_result);
+  // api.get('/ctx/:at', get_context, format_result);
+  api.get('/at/:at?', ensure_at, get_ddata, format_result);
+  // '/count/:store/
 
-    return api;
+  return api;
 }
 
 // expose module

lib/devicestatus.js

@@ -81,6 +81,7 @@ function storage (collection, ctx) {
   api.create = create;
   api.last = last;
   api.remove = remove;
+  api.aggregate = ctx.store.collector({ }, api);
   api.indexedFields = ['created_at'];
   return api;
 }

lib/entries.js

@@ -143,6 +143,7 @@ function storage(env, ctx) {
   api.persist = persist;
   api.query_for = query_for;
   api.getEntry = getEntry;
+  api.aggregate = ctx.store.collector({ }, api);
   api.indexedFields = [ 'date', 'type', 'sgv', 'mbg', 'sysTime', 'dateString' ];
   return api;
 }

lib/profile.js

@@ -47,6 +47,7 @@ function storage (collection, ctx) {
   api.remove = remove;
   api.last = last;
   api.indexedFields = ['startDate'];
+  api.aggregate = ctx.store.collector({ }, api);
   return api;
 }
 

lib/storage.js

@@ -1,12 +1,16 @@
 'use strict';
 
 var mongodb = require('mongodb');
+var collector = require('./aggregate');
+var find_options = require('./query');
 
 var connection = null;
 
 function init (env, cb, forceNewConnection) {
   var MongoClient = mongodb.MongoClient;
   var mongo = {};
+  mongo.collector = collector;
+  mongo.find_options = find_options;
 
   function maybe_connect (cb) {
 

lib/treatments.js

@@ -117,6 +117,7 @@ function storage (env, ctx) {
 
   api.remove = remove;
   api.save = save;
+  api.aggregate = ctx.store.collector({ }, api);
 
   return api;
 }