Add module to write results to a postgres database

This commit adds a `write-to-db-module`. This module takes the result of fetching and processing a report, and adds the data to a postgres database. The plan is for the reporter to insert this data in such a way that another application can come in behind it and serve the data via an API.

To enable this feature, the reporter is run with the `--write-to-database` option. The reporter needs the database to be configured in the config under the `postgres` prop.

Ref #194

.travis.yml

@@ -4,6 +4,12 @@ node_js:
 sudo: false
 os:
 - linux
+services:
+  - postgresql
+addons:
+  postgresql: "9.4"
+before_script:
+- "createdb travis_ci_test"
 deploy:
   edge: true
   provider: cloudfoundry

index.js

@@ -6,6 +6,8 @@ var Analytics = require("./src/analytics"),
     csv = require("fast-csv"),
     zlib = require('zlib');
 
+const writeResultsToDatabase = require("./src/write-results-to-db")
+
 
 // AWS credentials are looked for in env vars or in ~/.aws/config.
 // AWS bucket and path need to be set in env vars mentioned in config.js.
@@ -78,9 +80,16 @@ var run = function(options) {
         // JSON
         else {
           // some reports can be slimmed down for direct rendering
-          if (options.slim && report.slim) delete data.data;
-
-          writeReport(name, JSON.stringify(data, null, 2), ".json", done);
+          if (options.slim && report.slim) {
+            delete data.data;
+            writeReport(name, JSON.stringify(data, null, 2), ".json", done);
+          } else if (options["write-to-database"]) {
+            writeResultsToDatabase(data, { realtime: report.realtime }).then(() => {
+              writeReport(name, JSON.stringify(data, null, 2), ".json", done);
+            }).catch(err => done(err))
+          } else {
+            writeReport(name, JSON.stringify(data, null, 2), ".json", done);
+          }
         }
     });
   };

package.json

@@ -69,6 +69,8 @@
     "proxyquire": "^1.7.11"
   },
   "optionalDependencies": {
-    "newrelic": "^1.36.1"
+    "knex": "^0.12.6",
+    "newrelic": "^1.36.1",
+    "pg": "^6.1.2"
   }
 }

src/config.js

@@ -33,6 +33,13 @@ module.exports = {
     hostname: process.env.ANALYTICS_HOSTNAME || "",
   },
 
+  postgres: {
+    host : process.env.POSTGRES_HOST,
+    user : process.env.POSTGRES_USER,
+    password : process.env.POSTGRES_PASSWORD,
+    database : process.env.POSTGRES_DATABASE || "analytics-reporter",
+  },
+
   static: {
     path: '../analytics.usa.gov/'
   },

src/write-results-to-db.js

@@ -0,0 +1,91 @@
+const knex = require("knex")
+const config = require("./config")
+
+const writeResultsToDatabase = (results, { realtime } = {}) => {
+  const db = knex({ client: "pg", connection: config.postgres })
+
+  if (realtime) {
+    return _writeRealtimeResults({ db, results }).then(() => db.destroy())
+  } else if (results.query.dimensions.match(/ga:date/)) {
+    return _writeRegularResults({ db, results }).then(() => db.destroy())
+  } else {
+    return Promise.resolve()
+  }
+}
+
+const _dataForDataPoint = (dataPoint, { realtime } = {}) => {
+  const data = Object.assign({}, dataPoint)
+  let dateTime
+  if (realtime) {
+    dateTime = new Date()
+  } else {
+    dateTime = _dateTimeForDataPoint(dataPoint)
+  }
+  delete data.date
+  delete data.hour
+
+  return {
+    date_time: dateTime,
+    data: data,
+  }
+}
+
+const _dateTimeForDataPoint = (dataPoint) => {
+  let dateString = dataPoint.date
+  if (dataPoint.hour) {
+    dateString = `${dateString}T${dataPoint.hour}:00:00`
+  }
+  if (!isNaN(Date.parse(dateString))) {
+    return new Date(dateString)
+  }
+}
+
+const _rowForDataPoint = ({ results, dataPoint, realtime }) => {
+  const row = _dataForDataPoint(dataPoint, { realtime })
+  row.report_name = results.name
+  row.report_agency = results.agency
+  return row
+}
+
+const _rowWasAlreadyInserted = ({ db, row }) => {
+  const query = Object.assign({}, row)
+  Object.keys(query).forEach(key => {
+    if (query[key] === undefined) {
+      delete query[key]
+    }
+  })
+  return db("analytics_data").where(query).count().then(result => {
+    const count = parseInt(result[0].count)
+    return count > 0
+  })
+}
+
+const _writeRealtimeResults = ({ db, results }) => {
+  const rows = results.data.map(dataPoint => {
+    return _rowForDataPoint({ results, dataPoint, realtime: true })
+  })
+  return db("analytics_data").insert(rows)
+}
+
+const _writeRegularResults = ({ db, results }) => {
+  const rows = results.data.map(dataPoint => {
+    return _rowForDataPoint({ results, dataPoint })
+  })
+
+  const rowPromises = rows.map(row => {
+    return _rowWasAlreadyInserted({ db, row }).then(inserted => {
+      if (!inserted) {
+        return row
+      }
+    })
+  })
+
+  return Promise.all(rowPromises).then(rows => {
+    rows = rows.filter(row => {
+      return row !== undefined && row.date_time !== undefined
+    })
+    return db("analytics_data").insert(rows)
+  })
+}
+
+module.exports = writeResultsToDatabase