Shapeup4 : CQRS standards update, part 1

.circleci/config.yml

@@ -70,6 +70,7 @@ workflows:
             branches:
               only:
                 - develop
+                - feature/shapeup4-cqrs-update
 
       # Production builds are exectuted only on tagged commits to the
       # master branch.

config/default.js

@@ -35,6 +35,8 @@ module.exports = {
   KAFKA_MESSAGE_ORIGINATOR: process.env.KAFKA_MESSAGE_ORIGINATOR || 'u-bahn-api',
 
   // topics
+  UBAHN_ERROR_TOPIC: process.env.UBAHN_ERROR_TOPIC || 'ubahn.action.error',
+
   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',
@@ -126,7 +128,7 @@ module.exports = {
         orgField: process.env.ORGANIZATION_SKILLPROVIDER_PROPERTY_NAME || 'skillProviders'
       }
     },
-    MAX_BATCH_SIZE: parseInt(process.env.MAX_RESULT_SIZE, 10) || 10000,
+    MAX_BATCH_SIZE: parseInt(process.env.MAX_BATCH_SIZE, 10) || 10000,
     MAX_RESULT_SIZE: parseInt(process.env.MAX_RESULT_SIZE, 10) || 1000,
     MAX_BULK_SIZE: parseInt(process.env.MAX_BULK_SIZE, 10) || 100
   }

docker-pgsql-es/docker-compose.yml

@@ -4,14 +4,14 @@ services:
     image: "postgres:12.4"
     volumes:
       - database-data:/var/lib/postgresql/data/
-    ports: 
+    ports:
       - "5432:5432"
     environment:
       POSTGRES_PASSWORD: ${DB_PASSWORD}
       POSTGRES_USER: ${DB_USERNAME}
       POSTGRES_DB: ${DB_NAME}
   esearch:
-    image: elasticsearch:7.7.1
+    image: elasticsearch:7.13.4
     container_name: ubahn-data-processor-es_es
     ports:
       - "9200:9200"

scripts/db/dumpDbToEs.js

@@ -144,30 +144,40 @@ async function insertIntoES (modelName, dataset) {
     const chunked = _.chunk(dataset, config.get('ES.MAX_BULK_SIZE'))
     for (const ds of chunked) {
       const body = _.flatMap(ds, doc => [{ index: { _id: doc.id } }, doc])
-      await client.bulk({
-        index: topResources[esResourceName].index,
-        type: topResources[esResourceName].type,
-        body,
-        pipeline: topResources[esResourceName].ingest ? topResources[esResourceName].ingest.pipeline.id : undefined,
-        refresh: 'wait_for'
-      })
+      try {
+        await client.bulk({
+          index: topResources[esResourceName].index,
+          type: topResources[esResourceName].type,
+          body,
+          pipeline: topResources[esResourceName].ingest ? topResources[esResourceName].ingest.pipeline.id : undefined,
+          refresh: 'wait_for'
+        })
+      } catch (e) {
+        logger.error('ES, create mapping error.')
+        logger.error(JSON.stringify(e))
+      }
     }
   } else if (_.includes(_.keys(userResources), esResourceName)) {
     const userResource = userResources[esResourceName]
 
     if (userResource.nested === true && userResource.mappingCreated !== true) {
-      await client.indices.putMapping({
-        index: topResources.user.index,
-        type: topResources.user.type,
-        include_type_name: true,
-        body: {
-          properties: {
-            [userResource.propertyName]: {
-              type: 'nested'
+      try {
+        await client.indices.putMapping({
+          index: topResources.user.index,
+          type: topResources.user.type,
+          include_type_name: true,
+          body: {
+            properties: {
+              [userResource.propertyName]: {
+                type: 'nested'
+              }
             }
           }
-        }
-      })
+        })
+      } catch (e) {
+        logger.error('ES, nexted mapping error.')
+        logger.error(JSON.stringify(e))
+      }
       userResource.mappingCreated = true
     }
 
@@ -391,14 +401,12 @@ async function main () {
 
   for (let i = 0; i < keys.length; i++) {
     const key = keys[i]
+    const queryPage = { perPage: parseInt(config.get('ES.MAX_BATCH_SIZE'), 10), page: 1 }
     try {
-      const allData = await dbHelper.find(models[key], {})
-      let j = 0
-      const dataset = _.chunk(allData, config.get('ES.MAX_BATCH_SIZE'))
-      for (const data of dataset) {
+      while (true) {
+        const data = await dbHelper.find(models[key], { ...queryPage })
         for (let i = 0; i < data.length; i++) {
-          j++
-          logger.info(`Inserting data ${j} of ${allData.length}`)
+          logger.info(`Inserting data ${(i + 1) + (queryPage.perPage * (queryPage.page - 1))}`)
           logger.info(JSON.stringify(data[i]))
           if (!_.isString(data[i].created)) {
             data[i].created = new Date()
@@ -414,27 +422,33 @@ async function main () {
           }
         }
         await insertIntoES(key, data)
+        if (data.length < queryPage.perPage) {
+          logger.info('import data for ' + key + ' done')
+          break
+        } else {
+          queryPage.page = queryPage.page + 1
+        }
       }
-      logger.info('import data for ' + key + ' done')
     } catch (e) {
-      logger.error(e)
+      logger.error(JSON.stringify(_.get(e, 'meta.body', ''), null, 4))
+      logger.error(_.get(e, 'meta.meta.request.params.method', ''))
+      logger.error(_.get(e, 'meta.meta.request.params.path', ''))
       logger.warn('import data for ' + key + ' failed')
       continue
     }
-
     try {
       await createAndExecuteEnrichPolicy(key)
       logger.info('create and execute enrich policy for ' + key + ' done')
     } catch (e) {
-      logger.error(e)
+      logger.error(JSON.stringify(_.get(e, 'meta.body', ''), null, 4))
       logger.warn('create and execute enrich policy for ' + key + ' failed')
     }
 
     try {
       await createEnrichProcessor(key)
       logger.info('create enrich processor (pipeline) for ' + key + ' done')
     } catch (e) {
-      logger.error(e)
+      logger.error(JSON.stringify(_.get(e, 'meta.body', ''), null, 4))
       logger.warn('create enrich processor (pipeline) for ' + key + ' failed')
     }
   }

src/common/db-helper.js

@@ -105,24 +105,27 @@ async function get (model, pk, params) {
  * @param model the sequelize model object
  * @param entity entity to create
  * @param auth the user auth object
+ * @param transaction the transaction object
  * @returns {Promise<void>}
  */
-async function create (model, entity, auth) {
+async function create (model, entity, auth, transaction) {
   if (auth) {
     entity.createdBy = helper.getAuthUser(auth)
   }
-  return model.create(entity)
+  return model.create(entity, { transaction })
 }
 
 /**
  * delete object by pk
  * @param model the sequelize model object
  * @param pk the primary key
+ * @param transaction the transaction object
  * @returns {Promise<void>}
  */
-async function remove (model, pk, params) {
+async function remove (model, pk, params, transaction) {
   const instance = await get(model, pk, params)
-  return instance.destroy()
+  const result = await instance.destroy({ transaction })
+  return result
 }
 
 /**
@@ -132,13 +135,14 @@ async function remove (model, pk, params) {
  * @param entity entity to create
  * @param auth the auth object
  * @param auth the path params
+ * @param transaction the transaction object
  * @returns {Promise<void>}
  */
-async function update (model, pk, entity, auth, params) {
+async function update (model, pk, entity, auth, params, transaction) {
   // insure that object exists
   const instance = await get(model, pk, params)
   entity.updatedBy = helper.getAuthUser(auth)
-  return instance.update(entity)
+  return instance.update(entity, { transaction })
 }
 
 /**

src/common/es-helper.js

@@ -2,6 +2,7 @@ const config = require('config')
 const _ = require('lodash')
 const querystring = require('querystring')
 const logger = require('../common/logger')
+const helper = require('../common/helper')
 const appConst = require('../consts')
 const esClient = require('./es-client').getESClient()
 
@@ -282,6 +283,38 @@ function escapeRegex (str) {
   /* eslint-enable no-useless-escape */
 }
 
+/**
+ * Process create entity
+ * @param {String} resource resource name
+ * @param {Object} entity entity object
+ */
+async function processCreate (resource, entity) {
+  helper.validProperties(entity, ['id'])
+  await esClient.index({
+    index: DOCUMENTS[resource].index,
+    type: DOCUMENTS[resource].type,
+    id: entity.id,
+    body: entity,
+    refresh: 'wait_for'
+  })
+  logger.info(`Insert in Elasticsearch resource ${resource} entity, , ${JSON.stringify(entity, null, 2)}`)
+}
+
+/**
+ * Process delete entity
+ * @param {String} resource resource name
+ * @param {Object} entity entity object
+ */
+async function processDelete (resource, entity) {
+  helper.validProperties(entity, ['id'])
+  await esClient.delete({
+    index: DOCUMENTS[resource].index,
+    type: DOCUMENTS[resource].type,
+    id: entity.id,
+    refresh: 'wait_for'
+  })
+}
+
 async function getOrganizationId (handle) {
   const dBHelper = require('../common/db-helper')
   const sequelize = require('../models/index')
@@ -1453,6 +1486,9 @@ async function searchAchievementValues ({ organizationId, keyword }) {
 }
 
 module.exports = {
+  processCreate,
+  processUpdate: processCreate,
+  processDelete,
   searchElasticSearch,
   getFromElasticSearch,
   searchUsers,

src/common/helper.js

@@ -1,4 +1,5 @@
 const config = require('config')
+const Joi = require('@hapi/joi')
 const querystring = require('querystring')
 const errors = require('./errors')
 const appConst = require('../consts')
@@ -9,6 +10,20 @@ const busApi = require('tc-bus-api-wrapper')
 const busApiClient = busApi(_.pick(config, ['AUTH0_URL', 'AUTH0_AUDIENCE', 'TOKEN_CACHE_TIME', 'AUTH0_CLIENT_ID',
   'AUTH0_CLIENT_SECRET', 'BUSAPI_URL', 'KAFKA_ERROR_TOPIC', 'AUTH0_PROXY_SERVER_URL']))
 
+/**
+ * Function to valid require keys
+ * @param {Object} payload validated object
+ * @param {Array} keys required keys
+ * @throws {Error} if required key absent
+ */
+function validProperties (payload, keys) {
+  const schema = Joi.object(_.fromPairs(_.map(keys, key => [key, Joi.string().uuid().required()]))).unknown(true)
+  const error = schema.validate(payload).error
+  if (error) {
+    throw error
+  }
+}
+
 /**
  * get auth user handle or id
  * @param authUser the user
@@ -145,12 +160,33 @@ async function postEvent (topic, payload) {
   await busApiClient.postEvent(message)
 }
 
+/**
+ * Send error event to Kafka
+ * @params {String} topic the topic name
+ * @params {Object} payload the payload
+ * @params {String} action for which operation error occurred
+ */
+ async function publishError (topic, payload, action) {
+  _.set(payload, 'apiAction', action)
+  const message = {
+    topic,
+    originator: config.KAFKA_MESSAGE_ORIGINATOR,
+    timestamp: new Date().toISOString(),
+    'mime-type': 'application/json',
+    payload
+  }
+  logger.debug(`Publish error to Kafka topic ${topic}, ${JSON.stringify(message, null, 2)}`)
+  await busApiClient.postEvent(message)
+}
+
 module.exports = {
+  validProperties,
   getAuthUser,
   permissionCheck,
   checkIfExists,
   injectSearchMeta,
   getControllerMethods,
   getSubControllerMethods,
-  postEvent
+  postEvent,
+  publishError
 }