Implement CSV download

server/api.js

@@ -3,7 +3,7 @@
 const zlib = require("zlib");
 const Router = require("koa-router");
 const PassThrough = require("stream").PassThrough;
-
+const config = require("./config");
 const db = require("./db");
 const apiFunctions = require("./api-functions");
 const expression = require("../common/expression");
@@ -102,6 +102,88 @@ for (let [resource, flags] of Object.entries(resources)) {
     stream.end("]");
   });
 
+  // CSV download
+  router.get(`/${resource}.csv`, async (ctx, next) => {
+    let options = {};
+    let filter = true;
+    if (ctx.request.query.filter)
+      filter = expression.parse(ctx.request.query.filter);
+    if (ctx.request.query.limit) options.limit = +ctx.request.query.limit;
+    if (ctx.request.query.skip) options.skip = +ctx.request.query.skip;
+    if (ctx.request.query.sort)
+      options.sort = JSON.parse(ctx.request.query.sort);
+
+    if (!ctx.state.authorizer.hasAccess(resource, 2)) return next();
+
+    const columns = JSON.parse(ctx.request.query.columns);
+    const now = Date.now();
+    for (let [k, v] of Object.entries(columns))
+      columns[k] = expression.evaluate(
+        expression.parse(v),
+        null,
+        now
+      );
+
+    // Exclude temporary tasks and faults
+    if (resource === "tasks" || resource === "faults")
+      filter = expression.and(filter, [
+        "NOT",
+        ["<", ["PARAM", "expiry"], Date.now() + 60000]
+      ]);
+
+    let stream;
+    switch (ctx.acceptsEncodings("gzip", "deflate", "identity")) {
+      case "gzip":
+        stream = zlib.createGzip();
+        ctx.set("Content-Encoding", "gzip");
+        break;
+      case "deflate":
+        stream = zlib.createDeflate();
+        ctx.set("Content-Encoding", "deflate");
+        break;
+      default:
+        stream = new PassThrough();
+        break;
+    }
+
+    ctx.body = stream;
+    ctx.type = "text/csv";
+
+    stream.write(
+      Object.keys(columns).map(k => `"${k.replace(/"/, '""')}"`) + "\n"
+    );
+    await db.query(resource, filter, options, obj => {
+      let arr = Object.values(columns).map(exp => {
+        let v = expression.evaluate(exp, obj, null, e => {
+          if (Array.isArray(e))
+            if (e[0] === "PARAM") {
+              if (resource === "devices") {
+                if (e[1] === "Tags") {
+                  let tags = [];
+                  for (let p in obj)
+                    if (p.startsWith("Tags.")) tags.push(p.slice(5));
+
+                  return tags.join(", ");
+                }
+              }
+            } else if (e[0] === "FUNC") {
+              if (e[1] === "DATE_STRING")
+                if (e[2] && !Array.isArray(e[2]))
+                  return new Date(e[2]).toISOString();
+            }
+
+          return e;
+        });
+
+        if (Array.isArray(v) || v == null) return "";
+        if (typeof v === "string") return `"${v.replace(/"/g, '""')}"`;
+        return v;
+      });
+      stream.write(arr.join(",") + "\n");
+    });
+    stream.end();
+  });
+
   router.head(`/${resource}/:id`, async (ctx, next) => {
     let filter = ["=", ["PARAM", RESOURCE_IDS[resource]], ctx.params.id];
     if (!ctx.state.authorizer.hasAccess(resource, 2)) return next();

server/db.js

@@ -35,7 +35,7 @@ function getClient(db) {
 function query(resource, filter, options, callback) {
   options = options || {};
   let q;
-  filter = expression.evaluate(filter, null, null, Date.now());
+  filter = expression.evaluate(filter, null, Date.now());
 
   if (Array.isArray(filter)) {
     if (resource === "devices")
@@ -79,6 +79,10 @@ function query(resource, filter, options, callback) {
           if (err) return reject(err);
           if (resource === "devices")
             docs = docs.map(d => mongodbFunctions.flattenDevice(d));
+          else if (resource === "faults")
+            docs = docs.map(d => mongodbFunctions.flattenFault(d));
+          else if (resource === "tasks")
+            docs = docs.map(d => mongodbFunctions.flattenTask(d));
           return resolve(docs);
         });
       else
@@ -86,6 +90,10 @@ function query(resource, filter, options, callback) {
           doc => {
             if (resource === "devices")
               doc = mongodbFunctions.flattenDevice(doc);
+            else if (resource === "faults")
+              doc = mongodbFunctions.flattenFault(doc);
+            else if (resource === "tasks")
+              doc = mongodbFunctions.flattenTask(doc);
             callback(doc);
           },
           err => {
@@ -99,7 +107,7 @@ function query(resource, filter, options, callback) {
 
 function count(resource, filter) {
   let q;
-  filter = expression.evaluate(filter, null, null, Date.now());
+  filter = expression.evaluate(filter, null, Date.now());
 
   if (Array.isArray(filter)) {
     if (resource === "devices")

server/mongodb-functions.js

@@ -337,6 +337,25 @@ function flattenDevice(device) {
   return newDevice;
 }
 
+function flattenFault(fault) {
+  const f = Object.assign({}, fault);
+  if (f.timestamp) f.timestamp = +f.timestamp;
+  if (f.expiry) f.expiry = +f.expiry;
+  if (f.detail) {
+    for (let [k, v] of Object.entries(f.detail)) fault[`detail.${k}`] = v;
+    delete f.detail;
+  }
+  return f;
+}
+
+function flattenTask(task) {
+  let t = Object.assign({}, task);
+  t._id = "" + t._id;
+  if (t.timestamp) t.timestamp = +t.timestamp;
+  if (t.expiry) t.expiry = +t.expiry;
+  return t;
+}
+
 exports.processDeviceFilter = processDeviceFilter;
 exports.processTasksFilter = processTasksFilter;
 exports.processFaultsFilter = processFaultsFilter;
@@ -344,3 +363,5 @@ exports.filterToMongoQuery = filterToMongoQuery;
 exports.processDeviceProjection = processDeviceProjection;
 exports.processDeviceSort = processDeviceSort;
 exports.flattenDevice = flattenDevice;
+exports.flattenFault = flattenFault;
+exports.flattenTask = flattenTask;