Merge pull request #34 from rtyler/streaming-export-33

Switch the export controller to stream results directly from PostgreSQL to the response buffer

package.json

@@ -55,8 +55,10 @@
     "feathers-memory": "^2.2.0",
     "feathers-sequelize": "^3.1.2",
     "helmet": "^3.13.0",
+    "jsonstream": "^1.0.3",
     "passport-github": "^1.1.0",
     "pg": "^7.4.3",
+    "pg-query-stream": "^1.1.2",
     "pug": "^2.0.3",
     "sequelize": "^4.38.0",
     "serve-favicon": "^2.5.0",

src/controllers/export.ts

@@ -5,27 +5,32 @@
 
 import authentication from '@feathersjs/authentication';
 import cookieParser from 'cookie-parser';
+import db from '../models';
+
+import QueryStream from 'pg-query-stream';
+import JSONStream from 'jsonstream'
 
 export default (app) => {
   app.post('/export',
     cookieParser(),
     authentication.express.authenticate('jwt'),
     (req, res, next) => {
-      app.service('/events/bulk')
-        .find({
-          query: {
-            type: req.body.type,
-            startDate: req.body.startDate,
-            endDate: req.body.endDate,
-          },
-          user: (req as any).user,
-        })
-        .then((result) => {
-          res.setHeader('Content-Disposition', `attachment; filename=${req.body.type}-${req.body.startDate}.json`);
-          res.setHeader('Content-Type', 'application/json');
-          res.send(result);
+      db.sequelize.connectionManager.getConnection().then((pgConn) => {
+        const query = new QueryStream('SELECT * FROM events WHERE type = $1 AND "createdAt" > $2 AND "createdAt" <= $3', [
+          req.body.type,
+          req.body.startDate,
+          req.body.endDate
+        ]);
+        const stream = pgConn.query(query);
+        res.writeHead(200, {
+          'Content-Disposition' : `attachment; filename=${req.body.type}-${req.body.startDate}.json`,
+          'Content-Type': 'application/json',
+        });
+        stream.pipe(JSONStream.stringify(false)).pipe(res);
+        stream.on('end', () => {
           res.end();
-        })
-        .catch(next);
+        });
+      })
+      .catch((err) => { console.log(err.stack); next(err); });
     });
 };

src/services/bulk.ts

@@ -1,68 +0,0 @@
-/**
- * The /events/bulk service is largely responsible for streaming responses in
- * the form of files to clients
- */
-
-import { Application, HooksObject, Params, SKIP } from '@feathersjs/feathers';
-import { BadRequest, NotFound } from '@feathersjs/errors';
-import { QueryTypes } from 'sequelize';
-
-import authorize from '../hooks/authorize';
-import logger from '../logger';
-import db from '../models';
-import Event from '../models/event';
-
-export const bulkHooks : HooksObject = {
-  before: {
-    all: [
-      authorize(),
-      (context) => {
-        context.params.grants = context.data.grants
-        return context;
-      },
-    ],
-  },
-  after: {},
-  error: {},
-};
-
-/**
- * The Bulk service class intentionally only implements the find method
- */
-export class Bulk {
-  protected readonly app : Application;
-
-  constructor(app : Application) {
-    this.app = app;
-  }
-
-  public async find(params : Params) : Promise<any> {
-    if (!params.query.type) {
-      return Promise.reject(new BadRequest('Request must have a `type` in the URL'));
-    }
-
-    const grantedTypes = params.grants.filter(g => (g == '*') || (g == params.query.type));
-    if (grantedTypes.length == 0) {
-      return Promise.reject(new NotFound('No data found'));
-    }
-
-    /*
-     * This is clearly stupid. I have no idea how we'll query very large
-     * datasets from PostgreSQL but this at least gets us _everything_
-     */
-    return db.sequelize.query("SELECT * FROM events WHERE type = :type AND \"createdAt\" > :startDate AND \"createdAt\" < :endDate", {
-      replacements: {
-        type: params.query.type,
-        startDate: params.query.startDate,
-        endDate: params.query.endDate,
-      },
-      type: QueryTypes.SELECT,
-    });
-  }
-}
-
-
-export default (app : Application) => {
-  app.use('/events/bulk', new Bulk(app));
-  app.service('events/bulk').hooks(bulkHooks);
-};

src/services/index.ts

@@ -1,11 +1,9 @@
-import bulk from './bulk';
 import events from './events';
 import grants from './grants';
 import types from './types';
 import users from './users';
 
 export default (app) => {
-  app.configure(bulk);
   app.configure(events);
   app.configure(grants);
   app.configure(types);