Merge pull request #38 from topcoder-platform/feature/shapeup-cqrs-update2

DEV: shapeup#4 CQRS Standards Update for all objects

Author: sachin-maheshwari

.circleci/config.yml

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

README.md

@@ -20,70 +20,22 @@ The following parameters can be set in config files or in env variables:
     if not provided, then SSL connection is not used, direct insecure connection is used;
     if provided, it can be either path to private key file or private key content
 - KAFKA_GROUP_ID: the Kafka group id, default value is 'ubahn-processor-es'
-- UBAHN_CREATE_TOPIC: the create ubahn entity Kafka message topic, default value is 'u-bahn.action.create'
-- UBAHN_UPDATE_TOPIC: the update ubahn entity Kafka message topic, default value is 'u-bahn.action.update'
-- UBAHN_DELETE_TOPIC: the delete ubahn entity Kafka message topic, default value is 'u-bahn.action.delete'
-- UBAHN_AGGREGATE_TOPIC: the ubahn entity aggregate topic, that contains create, update and delete topics. Default value is 'u-bahn.action.aggregate'
 - GROUPS_MEMBER_ADD_TOPIC: the add groups member Kafka message topic, default value is 'groups.notification.member.add'
 - GROUPS_MEMBER_DELETE_TOPIC: the delete groups member Kafka message topic, default value is 'groups.notification.member.delete'
 - GROUPS_MEMBERSHIP_TYPE: the groups membership type that should be processed, default value is 'user'
 - ES_HOST: Elasticsearch host, default value is 'localhost:9200'
 - ES.AWS_REGION: The Amazon region to use when using AWS Elasticsearch service, default value is 'us-east-1'
-- ES.ACHIEVEMENT_PROVIDER_INDEX: Elasticsearch index name for achievement provider, default value is 'achievement_provider'
-- ES.ACHIEVEMENT_PROVIDER_TYPE: Elasticsearch index type for achievement provider, default value is '_doc'
-- ES.ATTRIBUTE_INDEX: Elasticsearch index name for attribute, default value is 'attribute'
-- ES.ATTRIBUTE_TYPE: Elasticsearch index type for attribute, default value is '_doc'
-- ES.ATTRIBUTE_GROUP_INDEX: Elasticsearch index name for attribute group, default value is 'attribute_group'
-- ES.ATTRIBUTE_GROUP_TYPE: Elasticsearch index type for attribute group, default value is '_doc'
-- ES.ORGANIZATION_INDEX: Elasticsearch index name for organization, default value is 'organization'
-- ES.ORGANIZATION_TYPE: Elasticsearch index type for organization, default value is '_doc'
-- ES.ROLE_INDEX: Elasticsearch index name for role, default value is 'role'
-- ES.ROLE_TYPE: Elasticsearch index type for role, default value is '_doc'
-- ES.SKILL_INDEX: Elasticsearch index name for skill, default value is 'skill'
-- ES.SKILL_TYPE: Elasticsearch index type for skill, default value is '_doc'
-- ES.SKILL_PROVIDER_INDEX: Elasticsearch index name for skill provider, default value is 'skill_provider'
-- ES.SKILL_PROVIDER_TYPE: Elasticsearch index type for skill provider, default value is '_doc'
-- ES.USER_INDEX: Elasticsearch index name for user, default value is 'user'
-- ES.USER_TYPE: Elasticsearch index type for user, default value is '_doc'
-- ES.USER_ACHIEVEMENT_PROPERTY_NAME: the user property name of achievement, default value is 'achievements',
-- ES.USER_EXTERNALPROFILE_PROPERTY_NAME: the user property name of externalProfile, default value is 'externalProfiles',
-- ES.USER_ATTRIBUTE_PROPERTY_NAME: the user property name of attribute, default value is 'attributes',
-- ES.USER_ROLE_PROPERTY_NAME: the user property name of role, default value is 'roles',
-- ES.USER_SKILL_PROPERTY_NAME: the user property name of skill, default value is 'skills'
-- ES.ORGANIZATION_SKILLPROVIDER_PROPERTY_NAME: the org property name of org skill providers, default value is 'skillProviders'
 - ES.USER_GROUP_PROPERTY_NAME: the user property name of group, default value is 'groups'
-- ATTRIBUTE_GROUP_PIPELINE_ID: The pipeline id for enrichment with attribute group. Default is `attributegroup-pipeline`
-- SKILL_PROVIDER_PIPELINE_ID: The pipeline id for enrichment with skill provider. Default is `skillprovider-pipeline`
 - USER_PIPELINE_ID: The pipeline id for enrichment of user details. Default is `user-pipeline`
-- ATTRIBUTE_GROUP_ENRICH_POLICYNAME: The enrich policy for attribute group. Default is `attributegroup-policy`
-- SKILL_PROVIDER_ENRICH_POLICYNAME: The enrich policy for skill provider. Default is `skillprovider-policy`
-- ROLE_ENRICH_POLICYNAME: The enrich policy for role. Default is `role-policy`
-- ACHIEVEMENT_PROVIDER_ENRICH_POLICYNAME: The enrich policy for achievement provider. Default is `achievementprovider-policy`
-- SKILL_ENRICH_POLICYNAME: The enrich policy for skill. Default is `skill-policy`
-- ATTRIBUTE_ENRICH_POLICYNAME: The enrich policy for skill. Default is `attribute-policy`
 - ELASTICCLOUD_ID: The elastic cloud id, if your elasticsearch instance is hosted on elastic cloud. DO NOT provide a value for ES_HOST if you are using this
 - ELASTICCLOUD_USERNAME: The elastic cloud username for basic authentication. Provide this only if your elasticsearch instance is hosted on elastic cloud
 - ELASTICCLOUD_PASSWORD: The elastic cloud password for basic authentication. Provide this only if your elasticsearch instance is hosted on elastic cloud
-- AUTH0_URL: The auth0 url, Default is 'https://topcoder-dev.auth0.com/oauth/token'
-- AUTH0_AUDIENCE: The auth0 audience for accessing ubahn api(s), Default is 'https://m2m.topcoder-dev.com/'
-- AUTH0_CLIENT_ID: The auth0 client id
-- AUTH0_CLIENT_SECRET: The auth0 client secret
-- AUTH0_PROXY_SERVER_URL: The auth0 proxy server url
-- TOKEN_CACHE_TIME: The token cache time
-- TOPCODER_GROUP_API: The topcoder groups api, Default is 'https://api.topcoder-dev.com/v5/groups'
 
 There is a `/health` endpoint that checks for the health of the app. This sets up an expressjs server and listens on the environment variable `PORT`. It's not part of the configuration file and needs to be passed as an environment variable
 
 Configuration for the tests is at `config/test.js`, only add such new configurations different from `config/default.js`
 
 - WAIT_TIME: wait time used in test, default is 1500 or 1.5 second
-- ES.ACHIEVEMENT_PROVIDER_INDEX: Elasticsearch index name for achievement provider in testing environment
-- ES.ATTRIBUTE_INDEX: Elasticsearch index name for attribute in testing environment
-- ES.ATTRIBUTE_GROUP_INDEX: Elasticsearch index name for attribute group in testing environment
-- ES.ORGANIZATION_INDEX: Elasticsearch index name for organization in testing environment
-- ES.ROLE_INDEX: Elasticsearch index name for role in testing environment
-- ES.SKILL_INDEX: Elasticsearch index name for skill in testing environment
-- ES.SKILL_PROVIDER_INDEX: Elasticsearch index name for skill provider in testing environment
 - ES.USER_INDEX: Elasticsearch index name for user in testing environment
 
 ## Local Kafka and ElasticSearch setup
@@ -150,14 +102,3 @@ To run the UBahn ES Processor using docker, follow the below steps
     ```
 
 5. When you are running the application for the first time, It will take some time initially to download the image and install the dependencies
-
-## Unit Tests and E2E Tests
-
-- Run `npm run test` to execute unit tests.
-- Run `npm run test:cov` to execute unit tests and generate coverage report.
-- RUN `npm run e2e` to execute e2e tests.
-- RUN `npm run e2e:cov` to execute e2e tests and generate coverage report.
-
-## Verification
-
-see [VERIFICATION.md](VERIFICATION.md)

VERIFICATION.md

@@ -14,19 +14,6 @@ module.exports = {
   // Kafka group id
   KAFKA_GROUP_ID: process.env.KAFKA_GROUP_ID || 'ubahn-processor-es',
 
-  TOPCODER_GROUP_API: process.env.TOPCODER_GROUP_API || 'https://api.topcoder-dev.com/v5/groups',
-
-  AUTH0_URL: process.env.AUTH0_URL || 'https://topcoder-dev.auth0.com/oauth/token', // Auth0 credentials
-  AUTH0_AUDIENCE: process.env.AUTH0_AUDIENCE || 'https://m2m.topcoder-dev.com/',
-  TOKEN_CACHE_TIME: process.env.TOKEN_CACHE_TIME,
-  AUTH0_CLIENT_ID: process.env.AUTH0_CLIENT_ID,
-  AUTH0_CLIENT_SECRET: process.env.AUTH0_CLIENT_SECRET,
-  AUTH0_PROXY_SERVER_URL: process.env.AUTH0_PROXY_SERVER_URL,
-
-  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',
   GROUPS_MEMBER_ADD_TOPIC: process.env.GROUPS_MEMBER_ADD_TOPIC || 'groups.notification.member.add',
   GROUPS_MEMBER_DELETE_TOPIC: process.env.GROUPS_MEMBER_DELETE_TOPIC || 'groups.notification.member.delete',
   GROUPS_MEMBERSHIP_TYPE: process.env.GROUPS_MEMBERSHIP_TYPE || 'user',
@@ -41,78 +28,12 @@ module.exports = {
     },
 
     AWS_REGION: process.env.AWS_REGION || 'us-east-1', // AWS Region to be used if we use AWS ES
-    ACHIEVEMENT_PROVIDER_INDEX: process.env.ACHIEVEMENT_PROVIDER_INDEX || 'achievement_provider',
-    ACHIEVEMENT_PROVIDER_TYPE: process.env.ACHIEVEMENT_PROVIDER_TYPE || '_doc',
-    ATTRIBUTE_INDEX: process.env.ATTRIBUTE_INDEX || 'attribute',
-    ATTRIBUTE_TYPE: process.env.ATTRIBUTE_TYPE || '_doc',
-    ATTRIBUTE_GROUP_INDEX: process.env.ATTRIBUTE_GROUP_INDEX || 'attribute_group',
-    ATTRIBUTE_GROUP_TYPE: process.env.ATTRIBUTE_GROUP_TYPE || '_doc',
-    ORGANIZATION_INDEX: process.env.ORGANIZATION_INDEX || 'organization',
-    ORGANIZATION_TYPE: process.env.ORGANIZATION_TYPE || '_doc',
-    ROLE_INDEX: process.env.ROLE_INDEX || 'role',
-    ROLE_TYPE: process.env.ROLE_TYPE || '_doc',
-    SKILL_INDEX: process.env.SKILL_INDEX || 'skill',
-    SKILL_TYPE: process.env.SKILL_TYPE || '_doc',
-    SKILL_PROVIDER_INDEX: process.env.SKILL_PROVIDER_INDEX || 'skill_provider',
-    SKILL_PROVIDER_TYPE: process.env.SKILL_PROVIDER_TYPE || '_doc',
     USER_INDEX: process.env.USER_INDEX || 'user',
     USER_TYPE: process.env.USER_TYPE || '_doc',
 
-    USER_ACHIEVEMENT_PROPERTY_NAME: process.env.USER_ACHIEVEMENT_PROPERTY_NAME || 'achievements',
-    USER_EXTERNALPROFILE_PROPERTY_NAME: process.env.USER_EXTERNALPROFILE_PROPERTY_NAME || 'externalProfiles',
-    USER_ATTRIBUTE_PROPERTY_NAME: process.env.USER_ATTRIBUTE_PROPERTY_NAME || 'attributes',
-    USER_ROLE_PROPERTY_NAME: process.env.USER_ROLE_PROPERTY_NAME || 'roles',
-    USER_SKILL_PROPERTY_NAME: process.env.USER_SKILL_PROPERTY_NAME || 'skills',
-    USER_GROUP_PROPERTY_NAME: process.env.USER_GROUP_PROPERTY_NAME || 'groups',
-
-    ORGANIZATION_SKILLPROVIDER_PROPERTY_NAME: process.env.ORGANIZATION_SKILLPROVIDER_PROPERTY_NAME || 'skillProviders',
-
     ENRICHMENT: {
-      attributegroup: {
-        enrichPolicyName: process.env.ATTRIBUTE_GROUP_ENRICH_POLICYNAME || 'attributegroup-policy',
-        pipelineId: process.env.ATTRIBUTE_GROUP_PIPELINE_ID || 'attributegroup-pipeline'
-      },
-      skillprovider: {
-        enrichPolicyName: process.env.SKILL_PROVIDER_ENRICH_POLICYNAME || 'skillprovider-policy',
-        pipelineId: process.env.SKILL_PROVIDER_PIPELINE_ID || 'skillprovider-pipeline'
-      },
       user: {
         pipelineId: process.env.USER_PIPELINE_ID || 'user-pipeline'
-      },
-      role: {
-        enrichPolicyName: process.env.ROLE_ENRICH_POLICYNAME || 'role-policy'
-      },
-      achievementprovider: {
-        enrichPolicyName: process.env.ACHIEVEMENT_PROVIDER_ENRICH_POLICYNAME || 'achievementprovider-policy'
-      },
-      skill: {
-        enrichPolicyName: process.env.SKILL_ENRICH_POLICYNAME || 'skill-policy'
-      },
-      attribute: {
-        enrichPolicyName: process.env.ATTRIBUTE_ENRICH_POLICYNAME || 'attribute-policy'
-      },
-      organization: {
-        enrichPolicyName: process.env.ORGANIZATION_ENRICH_POLICYNAME || 'organization-policy'
-      },
-      // sub resources under user
-      achievement: {
-        userField: process.env.USER_ACHIEVEMENT_PROPERTY_NAME || 'achievements'
-      },
-      externalprofile: {
-        userField: process.env.USER_EXTERNALPROFILE_PROPERTY_NAME || 'externalProfiles'
-      },
-      userattribute: {
-        userField: process.env.USER_ATTRIBUTE_PROPERTY_NAME || 'attributes'
-      },
-      userrole: {
-        userField: process.env.USER_ROLE_PROPERTY_NAME || 'roles'
-      },
-      userskill: {
-        userField: process.env.USER_SKILL_PROPERTY_NAME || 'skills'
-      },
-      // sub resources under organization
-      organizationskillprovider: {
-        orgField: process.env.ORGANIZATION_SKILLPROVIDER_PROPERTY_NAME || 'skillProviders'
       }
     }
   }

config/default.js

@@ -12,7 +12,7 @@ services:
       - "9092:9092"
     environment:
       KAFKA_ADVERTISED_HOST_NAME: localhost
-      KAFKA_CREATE_TOPICS: "u-bahn.action.aggregate:1:1,groups.notification.member.add:1:1,groups.notification.member.delete:1:1"
+      KAFKA_CREATE_TOPICS: "groups.notification.member.add:1:1,groups.notification.member.delete:1:1"
       KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
   esearch:
     image: elasticsearch:7.7.1

config/test.js

@@ -0,0 +1 @@
+

docker-kafka-es/docker-compose.yml

@@ -6,23 +6,12 @@
   "scripts": {
     "start": "node src/app.js",
     "lint": "standard",
-    "lint:fix": "standard --fix",
-    "view-data": "node test/common/view-data.js",
-    "test": "mocha test/unit/test.js --require test/unit/prepare.js --timeout 20000 --exit",
-    "test:cov": "nyc --reporter=html --reporter=text mocha test/unit/test.js --require test/unit/prepare.js --timeout 20000 --exit",
-    "e2e": "mocha test/e2e/test.js --timeout 20000 --exit",
-    "e2e:cov": "nyc --reporter=html --reporter=text mocha test/e2e/test.js --timeout 20000 --exit"
+    "lint:fix": "standard --fix"
   },
   "author": "TCSCODER",
   "license": "none",
   "devDependencies": {
-    "mocha": "^7.1.2",
-    "mocha-prepare": "^0.1.0",
-    "nock": "^12.0.3",
-    "nyc": "^14.1.1",
-    "should": "^13.2.3",
-    "standard": "^12.0.1",
-    "superagent": "^5.1.0"
+    "standard": "^12.0.1"
   },
   "dependencies": {
     "@elastic/elasticsearch": "^7.9.1",
@@ -35,7 +24,6 @@
     "lodash": "^4.17.19",
     "no-kafka": "^3.4.3",
     "axios": "^0.19.2",
-    "tc-core-library-js": "appirio-tech/tc-core-library-js.git#v2.6",
     "topcoder-healthcheck-dropin": "^1.0.3",
     "winston": "^3.2.1"
   },

docker/api.env

@@ -9,7 +9,6 @@ const _ = require('lodash')
 const healthcheck = require('topcoder-healthcheck-dropin')
 const logger = require('./common/logger')
 const helper = require('./common/helper')
-const ProcessorService = require('./services/ProcessorService')
 const GroupsProcessorService = require('./services/GroupsProcessorService')
 const Mutex = require('async-mutex').Mutex
 
@@ -70,21 +69,7 @@ const dataHandler = (messageSet, topic, partition) => Promise.each(messageSet, a
   }
   const transactionId = _.uniqueId('transaction_')
   try {
-    if (messageJSON.payload.originalTopic) {
-      switch (messageJSON.payload.originalTopic) {
-        case config.UBAHN_CREATE_TOPIC:
-          await ProcessorService.processCreate(messageJSON, transactionId)
-          break
-        case config.UBAHN_UPDATE_TOPIC:
-          await ProcessorService.processUpdate(messageJSON, transactionId)
-          break
-        case config.UBAHN_DELETE_TOPIC:
-          await ProcessorService.processDelete(messageJSON, transactionId)
-          break
-        default:
-          throw new Error(`Unknown original topic: ${messageJSON.payload.originalTopic}`)
-      }
-    } else {
+    if (!messageJSON.payload.originalTopic) {
       switch (topic) {
         case config.GROUPS_MEMBER_ADD_TOPIC:
           await GroupsProcessorService.processMemberAdd(messageJSON, transactionId)
@@ -122,8 +107,7 @@ const check = () => {
   return connected
 }
 
-// const topics = [config.UBAHN_CREATE_TOPIC, config.UBAHN_UPDATE_TOPIC, config.UBAHN_DELETE_TOPIC]
-const topics = [config.UBAHN_AGGREGATE_TOPIC, config.GROUPS_MEMBER_ADD_TOPIC, config.GROUPS_MEMBER_DELETE_TOPIC]
+const topics = [config.GROUPS_MEMBER_ADD_TOPIC, config.GROUPS_MEMBER_DELETE_TOPIC]
 
 consumer
   .init([{

package-lock.json

@@ -8,10 +8,7 @@ const elasticsearch = require('@elastic/elasticsearch')
 const _ = require('lodash')
 const Joi = require('@hapi/joi')
 const { Mutex } = require('async-mutex')
-const axios = require('axios')
 const logger = require('./logger')
-const m2mAuth = require('tc-core-library-js').auth.m2m
-const topcoderM2M = m2mAuth(_.pick(config, ['AUTH0_URL', 'AUTH0_AUDIENCE', 'TOKEN_CACHE_TIME', 'AUTH0_PROXY_SERVER_URL']))
 
 AWS.config.region = config.ES.AWS_REGION
 
@@ -22,34 +19,6 @@ let transactionId
 const esClientMutex = new Mutex()
 const mutexReleaseMap = {}
 
-/* Function to get M2M token
- * (Topcoder APIs only)
- * @returns {Promise}
- */
-async function getTopcoderM2Mtoken () {
-  return topcoderM2M.getMachineToken(config.AUTH0_CLIENT_ID, config.AUTH0_CLIENT_SECRET)
-}
-
-/**
- * Returns the user in Topcoder identified by the email
- * @param {String} email The user email
- */
-async function getUserGroup (memberId) {
-  const url = config.TOPCODER_GROUP_API
-  const token = await getTopcoderM2Mtoken()
-  const params = { memberId, membershipType: 'user', page: 1 }
-
-  logger.debug(`request GET ${url} with params: ${JSON.stringify(params)}`)
-  let groups = []
-  let groupRes = await axios.get(url, { headers: { Authorization: `Bearer ${token}` }, params })
-  while (groupRes.data.length > 0) {
-    groups = _.concat(groups, _.map(groupRes.data, g => _.pick(g, 'id', 'name')))
-    params.page = params.page + 1
-    groupRes = await axios.get(url, { headers: { Authorization: `Bearer ${token}` }, params })
-  }
-  return groups
-}
-
 /**
  * Get Kafka options
  * @return {Object} the Kafka options
@@ -175,41 +144,6 @@ async function updateUser (userId, body, seqNo, primaryTerm, transactionId) {
   }
 }
 
-/**
- * Function to get org from es
- * @param {String} organizationId
- * @param {String} transactionId
- * @returns {Object} organization
- */
-async function getOrg (organizationId, transactionId) {
-  const client = await getESClient()
-  const { body: org } = await client.get({ index: config.get('ES.ORGANIZATION_INDEX'), type: config.get('ES.ORGANIZATION_TYPE'), id: organizationId, transactionId })
-  return { seqNo: org._seq_no, primaryTerm: org._primary_term, org: org._source }
-}
-
-/**
- * Function to update es organization
- * @param {String} organizationId
- * @param {Number} seqNo
- * @param {Number} primaryTerm
- * @param {String} transactionId
- * @param {Object} body
- */
-async function updateOrg (organizationId, body, seqNo, primaryTerm, transactionId) {
-  const client = await getESClient()
-  await client.index({
-    index: config.get('ES.ORGANIZATION_INDEX'),
-    type: config.get('ES.ORGANIZATION_TYPE'),
-    id: organizationId,
-    transactionId,
-    body,
-    if_seq_no: seqNo,
-    if_primary_term: primaryTerm,
-    refresh: 'wait_for'
-  })
-  await client.enrich.executePolicy({ name: config.get('ES.ENRICHMENT.organization.enrichPolicyName') })
-}
-
 /**
  * Fuction to get an Error with statusCode property
  * @param {String} message error message
@@ -242,10 +176,7 @@ module.exports = {
   getESClient,
   validProperties,
   getUser,
-  getUserGroup,
   updateUser,
-  getOrg,
-  updateOrg,
   getErrorWithStatus,
   checkEsMutexRelease
 }