Merge pull request #43 from topcoder-platform/develop

Sync master with develop

Author: callmekatootie

README.md

@@ -35,6 +35,7 @@ Configuration for the application is at config/default.js and config/production.
 - UBAHN_CREATE_TOPIC: Kafka topic for create message
 - UBAHN_UPDATE_TOPIC: Kafka topic for update message
 - UBAHN_DELETE_TOPIC: Kafka topic for delete message
+- UBAHN_AGGREGATE_TOPIC: Kafka topic that is used to combine all create, update and delete message(s)
 - ES.HOST: Elasticsearch host
 - ES.API_VERSION: Elasticsearch API version
 - ES.DOCUMENTS: Elasticsearch index, type and id mapping for resources.

config/default.js

@@ -36,6 +36,7 @@ module.exports = {
   UBAHN_CREATE_TOPIC: process.env.UBAHN_CREATE_TOPIC || 'u-bahn.action.create',
   UBAHN_UPDATE_TOPIC: process.env.UBAHN_UPDATE_TOPIC || 'u-bahn.action.update',
   UBAHN_DELETE_TOPIC: process.env.UBAHN_DELETE_TOPIC || 'u-bahn.action.delete',
+  UBAHN_AGGREGATE_TOPIC: process.env.UBAHN_AGGREGATE_TOPIC || 'u-bahn.action.aggregate',
 
   EMSI: {
     CLIENT_ID: process.env.EMSI_CLIENT_ID,

scripts/db/dumpDbToEs.js

@@ -11,15 +11,12 @@ const {
 
 async function cleanupES () {
   const client = getESClient()
-  client.indices.delete({
+
+  await client.indices.delete({
     index: '_all'
-  }, function (err, res) {
-    if (err) {
-      console.error(err.message)
-    } else {
-      console.log('Existing indices have been deleted!')
-    }
   })
+
+  console.log('Existing indices have been deleted!')
 }
 
 async function insertIntoES (modelName, body) {

src/common/helper.js

@@ -246,6 +246,12 @@ async function postEvent (topic, payload) {
     payload
   }
   await busApiClient.postEvent(message)
+
+  // Post to the aggregate topic
+  message.payload.originalTopic = topic
+  message.topic = config.UBAHN_AGGREGATE_TOPIC
+  logger.debug(`Posting event to aggregate topic ${message.topic}`)
+  await busApiClient.postEvent(message)
 }
 
 module.exports = {

src/modules/achievement/service.js

@@ -12,15 +12,15 @@ const methods = helper.getServiceMethods(
     achievementsProviderId: joi.string().required(),
     name: joi.string().required(),
     uri: joi.string(),
-    certifierId: joi.string(),
+    certifierId: joi.string().allow(''),
     certifiedDate: joi.date().format('iso')
   },
   { // patch request body joi schema
     userId: joi.string().required(),
     achievementsProviderId: joi.string().required(),
     name: joi.string(),
     uri: joi.string(),
-    certifierId: joi.string(),
+    certifierId: joi.string().allow(''),
     certifiedDate: joi.date().format('iso')
   },
   { // search request query joi schema