initial commit

Author: Nkumar

Procfile

@@ -0,0 +1 @@
+web: node index.js

common/auth.js

@@ -0,0 +1,40 @@
+/**
+ * Authentication and authorization middleware.
+ */
+const _ = require('lodash')
+const createError = require('http-errors')
+const config = require('config')
+
+const helper = require('./helper')
+
+/**
+ * Check if the request is authenticated and authorized.
+ *
+ * @param {Object} req the request
+ * @param {Object} res the response
+ * @param {Function} next the next middleware
+ */
+function auth (req, res, next) {
+  // Parse the token from request header
+  let token
+  if (req.headers.authorization && req.headers.authorization.split(' ')[0] === 'Bearer') {
+    token = req.headers.authorization.split(' ')[1]
+  }
+
+  if (!token) {
+    throw createError.Unauthorized('Action is not allowed for anonymous')
+  }
+
+  const payload = helper.verifyJwtToken(token)
+
+  // Check if the service name is in the allowed list
+  if (!_.includes(config.ALLOWED_SERVICES, payload.name)) {
+    throw createError.Forbidden(`Action is not allowed for ${payload.name} service`)
+  }
+
+  // Set source service name to the request
+  req.sourceServiceName = payload.name
+  next()
+}
+
+module.exports = auth

common/helper.js

@@ -0,0 +1,213 @@
+'use strict';
+
+function _asyncToGenerator(fn) { return function () { var gen = fn.apply(this, arguments); return new Promise(function (resolve, reject) { function step(key, arg) { try { var info = gen[key](arg); var value = info.value; } catch (error) { reject(error); return; } if (info.done) { resolve(value); } else { return Promise.resolve(value).then(function (value) { step("next", value); }, function (err) { step("throw", err); }); } } return step("next"); }); }; }
+
+/**
+ * This file defines helper methods.
+ */
+var util = require('util');
+var _ = require('lodash');
+var Joi = require('joi');
+var getParams = require('get-parameter-names');
+var config = require('config');
+var jwt = require('jsonwebtoken');
+var createError = require('http-errors');
+
+var logger = require('./logger');
+
+/**
+ * Convert array with arguments to object.
+ *
+ * @param {Array} params the name of parameters
+ * @param {Array} arr the array with values
+ * @returns {Object} the combined object
+ * @private
+ */
+function combineObject(params, arr) {
+  var ret = {};
+  _.forEach(arr, function (arg, i) {
+    ret[params[i]] = arg;
+  });
+  return ret;
+}
+
+/**
+ * Decorate all functions of a service and log debug information if DEBUG is enabled.
+ *
+ * @param {Object} service the service
+ * @private
+ */
+function decorateWithLogging(service) {
+  if (config.LOG_LEVEL !== 'debug') {
+    return;
+  }
+  _.forEach(service, function (method, name) {
+    var params = method.params || getParams(method);
+    service[name] = function () {
+      var _ref = _asyncToGenerator( /*#__PURE__*/regeneratorRuntime.mark(function _callee() {
+        for (var _len = arguments.length, args = Array(_len), _key = 0; _key < _len; _key++) {
+          args[_key] = arguments[_key];
+        }
+
+        var result;
+        return regeneratorRuntime.wrap(function _callee$(_context) {
+          while (1) {
+            switch (_context.prev = _context.next) {
+              case 0:
+                logger.debug('ENTER ' + name);
+                logger.debug('input arguments');
+                logger.debug(util.inspect(combineObject(params, args)));
+                _context.next = 5;
+                return method.apply(this, args);
+
+              case 5:
+                result = _context.sent;
+
+                logger.debug('EXIT ' + name);
+                logger.debug('output arguments');
+                logger.debug(util.inspect(result));
+                return _context.abrupt('return', result);
+
+              case 10:
+              case 'end':
+                return _context.stop();
+            }
+          }
+        }, _callee, this);
+      }));
+
+      function serviceMethodWithLogging() {
+        return _ref.apply(this, arguments);
+      }
+
+      return serviceMethodWithLogging;
+    }();
+  });
+}
+
+/**
+ * Decorate all functions of a service and validate input values
+ * and replace input arguments with sanitized result = require('Joi.
+ * Service method must have a `schema` property with Joi schema.
+ *
+ * @param {Object} service the service
+ * @private
+ */
+function decorateWithValidator(service) {
+  _.forEach(service, function (method, name) {
+    if (!method.schema) {
+      return;
+    }
+    var params = getParams(method);
+    service[name] = function () {
+      var _ref2 = _asyncToGenerator( /*#__PURE__*/regeneratorRuntime.mark(function _callee2() {
+        for (var _len2 = arguments.length, args = Array(_len2), _key2 = 0; _key2 < _len2; _key2++) {
+          args[_key2] = arguments[_key2];
+        }
+
+        var value, normalized, newArgs;
+        return regeneratorRuntime.wrap(function _callee2$(_context2) {
+          while (1) {
+            switch (_context2.prev = _context2.next) {
+              case 0:
+                value = combineObject(params, args);
+                normalized = Joi.attempt(value, method.schema, { abortEarly: false });
+                // Joi will normalize values
+                // for example string number '1' to 1
+                // if schema type is number
+
+                newArgs = _.map(params, function (param) {
+                  return normalized[param];
+                });
+                return _context2.abrupt('return', method.apply(this, newArgs));
+
+              case 4:
+              case 'end':
+                return _context2.stop();
+            }
+          }
+        }, _callee2, this);
+      }));
+
+      function serviceMethodWithValidation() {
+        return _ref2.apply(this, arguments);
+      }
+
+      return serviceMethodWithValidation;
+    }();
+    service[name].params = params;
+  });
+}
+
+/**
+ * Apply logger and validation decorators to the service.
+ *
+ * @export helper/buildService
+ * @param {any} service the service to wrap
+ */
+function buildService(service) {
+  decorateWithValidator(service);
+  decorateWithLogging(service);
+}
+
+/**
+ * Verify the JWT token and get the payload.
+ *
+ * @param {String} token the JWT token to verify
+ * @returns {Object} the payload decoded from the token
+ */
+function verifyJwtToken(token) {
+  var payload = void 0;
+
+  try {
+    payload = jwt.verify(token, config.JWT_TOKEN_SECRET);
+  } catch (err) {
+    if (err.message === 'jwt expired') {
+      throw createError.Unauthorized('Token has been expired');
+    }
+
+    throw createError.Unauthorized('Failed to verify token');
+  }
+
+  if (!payload) {
+    throw createError.Unauthorized('Failed to decode token');
+  }
+
+  return payload;
+}
+
+/**
+ * Sign the payload and get the JWT token.
+ *
+ * @param {Object} payload the payload to be sign
+ * @returns {String} the token
+ */
+function signJwtToken(payload) {
+  return jwt.sign(payload, config.JWT_TOKEN_SECRET, { expiresIn: config.JWT_TOKEN_EXPIRES_IN });
+}
+
+/**
+ * Validate the event based on the source service, type, and message.
+ *
+ * @param {String} sourceServiceName the source service name
+ * @param {Object} event the event
+ */
+function validateEvent(sourceServiceName, event) {
+  // The message should be a JSON-formatted string
+  try {
+    JSON.parse(event.message);
+  } catch (err) {
+    logger.error(err);
+    throw createError.BadRequest('"message" is not a valid JSON-formatted string: ' + err.message);
+  }
+
+  // The message should match with the source service and type
+  // no-op for now
+}
+
+module.exports = {
+  buildService: buildService,
+  verifyJwtToken: verifyJwtToken,
+  signJwtToken: signJwtToken,
+  validateEvent: validateEvent
+};

common/logger.js

@@ -0,0 +1,16 @@
+/**
+ * Configure the logger.
+ */
+const config = require('config')
+const winston = require('winston')
+
+const logger = new winston.Logger({
+  level: config.LOG_LEVEL,
+  transports: [
+    new winston.transports.Console({
+      timestamp: () => new Date().toISOString()
+    })
+  ]
+})
+
+module.exports = logger

config/default.js

@@ -0,0 +1,14 @@
+/**
+ * The default configuration.
+ */
+module.exports = {
+    KAFKA_CLIENT_CERT: process.env.KAFKA_CLIENT_CERT ? process.env.KAFKA_CLIENT_CERT.replace('\\n', '\n') : null,
+  KAFKA_CLIENT_CERT_KEY: process.env.KAFKA_CLIENT_CERT_KEY ? process.env.KAFKA_CLIENT_CERT_KEY.replace('\\n', '\n') : null,
+  ALLOWED_SERVICES: ['project-service','message-service'],
+  LOG_LEVEL: process.env.LOG_LEVEL || 'debug',
+  CONTEXT_PATH: process.env.API_VERSION || '/eventbus/',
+  PORT: process.env.PORT || '3000',
+  JWT_TOKEN_SECRET: process.env.JWT_TOKEN_SECRET || 'JWT_TOKEN_SECRET',
+  JWT_TOKEN_EXPIRES_IN: process.env.JWT_TOKEN_EXPIRES_IN || '100 days',
+  KAFKA_TOPIC_PREFIX: process.env.KAFKA_TOPIC_PREFIX || 'joan-26673.notifications.'
+}

controllers/EventController.js

@@ -0,0 +1,45 @@
+'use strict';
+
+/**
+ * Create a new event.
+ *
+ * @param {Object} req the request
+ * @param {Object} res the response
+ * @param {Function} next the next middleware
+ */
+var create = function () {
+  var _ref = _asyncToGenerator( /*#__PURE__*/regeneratorRuntime.mark(function _callee(req, res, next) {
+    return regeneratorRuntime.wrap(function _callee$(_context) {
+      while (1) {
+        switch (_context.prev = _context.next) {
+          case 0:
+            _context.next = 2;
+            return MessageBusService.postEvent(req.sourceServiceName, req.body);
+
+          case 2:
+            res.status(204).end();
+            next();
+
+          case 4:
+          case 'end':
+            return _context.stop();
+        }
+      }
+    }, _callee, this);
+  }));
+
+  return function create(_x, _x2, _x3) {
+    return _ref.apply(this, arguments);
+  };
+}();
+
+function _asyncToGenerator(fn) { return function () { var gen = fn.apply(this, arguments); return new Promise(function (resolve, reject) { function step(key, arg) { try { var info = gen[key](arg); var value = info.value; } catch (error) { reject(error); return; } if (info.done) { resolve(value); } else { return Promise.resolve(value).then(function (value) { step("next", value); }, function (err) { step("throw", err); }); } } return step("next"); }); }; }
+
+/**
+ * The Event controller.
+ */
+var MessageBusService = require('../services/MessageBusService');
+
+module.exports = {
+  create: create
+};

controllers/HealthController.js

@@ -0,0 +1,41 @@
+/**
+ * Contains endpoints related to service health
+ */
+'use strict';
+
+/**
+ * Health Check.
+ * @param req the request
+ * @param res the response
+ */
+
+var health = function () {
+  var _ref = _asyncToGenerator( /*#__PURE__*/regeneratorRuntime.mark(function _callee(req, res, next) {
+    return regeneratorRuntime.wrap(function _callee$(_context) {
+      while (1) {
+        switch (_context.prev = _context.next) {
+          case 0:
+            res.json({ health: 'ok' });
+            next();
+
+          case 2:
+          case 'end':
+            return _context.stop();
+        }
+      }
+    }, _callee, this);
+  }));
+
+  return function health(_x, _x2, _x3) {
+    return _ref.apply(this, arguments);
+  };
+}();
+
+// Exports
+
+
+function _asyncToGenerator(fn) { return function () { var gen = fn.apply(this, arguments); return new Promise(function (resolve, reject) { function step(key, arg) { try { var info = gen[key](arg); var value = info.value; } catch (error) { reject(error); return; } if (info.done) { resolve(value); } else { return Promise.resolve(value).then(function (value) { step("next", value); }, function (err) { step("throw", err); }); } } return step("next"); }); }; }
+
+module.exports = {
+  health: health
+};