Merge winning submission from enrich contest

Author: callmekatootie

README.md

@@ -29,7 +29,6 @@ The following parameters can be set in config files or in env variables:
 - 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.API_VERSION: Elasticsearch API version, default value is '7.4'
 - 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'
@@ -53,6 +52,7 @@ The following parameters can be set in config files or in env variables:
 - 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'
+- ENRICH_USER_PIPELINE_NAME: the user enrich pipeline name, default value is 'enrich_user'
 
 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
 

config/default.js

@@ -23,9 +23,8 @@ module.exports = {
   GROUPS_MEMBERSHIP_TYPE: process.env.GROUPS_MEMBERSHIP_TYPE || 'user',
 
   ES: {
-    HOST: process.env.ES_HOST || 'localhost:9200',
+    HOST: process.env.ES_HOST || 'http://localhost:9200',
     AWS_REGION: process.env.AWS_REGION || 'us-east-1', // AWS Region to be used if we use AWS ES
-    API_VERSION: process.env.ES_API_VERSION || '7.4',
     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',
@@ -50,6 +49,8 @@ module.exports = {
     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'
+    ORGANIZATION_SKILLPROVIDER_PROPERTY_NAME: process.env.ORGANIZATION_SKILLPROVIDER_PROPERTY_NAME || 'skillProviders',
+
+    ENRICH_USER_PIPELINE_NAME: process.env.ENRICH_USER_PIPELINE_NAME || 'enrich_user'
   }
 }

docker-kafka-es/docker-compose.yml

@@ -15,7 +15,7 @@ services:
       KAFKA_CREATE_TOPICS: "u-bahn.action.aggregate:1:1,groups.notification.member.add:1:1,groups.notification.member.delete:1:1"
       KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
   esearch:
-    image: docker.elastic.co/elasticsearch/elasticsearch:7.4.2
+    image: elasticsearch:7.7.1
     container_name: ubahn-data-processor-es_es
     ports:
       - "9200:9200"

package-lock.json

@@ -26,14 +26,14 @@
     "superagent": "^5.1.0"
   },
   "dependencies": {
+    "@elastic/elasticsearch": "^7.9.1",
     "@hapi/joi": "^15.1.0",
     "async-mutex": "^0.2.4",
+    "aws-elasticsearch-connector": "^9.0.0",
     "aws-sdk": "^2.476.0",
     "bluebird": "^3.5.5",
     "config": "^3.1.0",
-    "elasticsearch": "^16.7.1",
     "get-parameter-names": "^0.3.0",
-    "http-aws-es": "^6.0.0",
     "lodash": "^4.17.19",
     "no-kafka": "^3.4.3",
     "topcoder-healthcheck-dropin": "^1.0.3",

package.json

@@ -8,10 +8,14 @@ const { validProperties } = require('./helper')
 const topResources = {
   achievementprovider: {
     index: config.get('ES.ACHIEVEMENT_PROVIDER_INDEX'),
+    mappingFields: ['id', 'achievementsProviderId', 'name', 'uri', 'certifierId', 'certifiedDate', 'created', 'updated', 'createdBy', 'updatedBy'],
+    enrichPolicy: 'achievementprovider-policy',
     type: config.get('ES.ACHIEVEMENT_PROVIDER_TYPE')
   },
   attribute: {
     index: config.get('ES.ATTRIBUTE_INDEX'),
+    enrichPolicy: 'attribute-policy',
+    mappingFields: ['id', 'name', 'attributeGroupId', 'created', 'updated', 'createdBy', 'updatedBy'],
     type: config.get('ES.ATTRIBUTE_TYPE')
   },
   attributegroup: {
@@ -21,14 +25,20 @@ const topResources = {
 
   organization: {
     index: config.get('ES.ORGANIZATION_INDEX'),
+    enrichPolicy: 'organization-policy',
+    mappingFields: ['id', 'name', 'created', 'updated', 'createdBy', 'updatedBy'],
     type: config.get('ES.ORGANIZATION_TYPE')
   },
   role: {
     index: config.get('ES.ROLE_INDEX'),
+    enrichPolicy: 'role-policy',
+    mappingFields: ['id', 'name', 'created', 'updated', 'createdBy', 'updatedBy'],
     type: config.get('ES.ROLE_TYPE')
   },
   skill: {
     index: config.get('ES.SKILL_INDEX'),
+    enrichPolicy: 'skill-policy',
+    mappingFields: ['id', 'skillProviderId', 'name', 'externalId', 'uri', 'created', 'updated', 'createdBy', 'updatedBy'],
     type: config.get('ES.SKILL_TYPE')
   },
   skillprovider: {
@@ -44,27 +54,32 @@ const topResources = {
 const userResources = {
   achievement: {
     propertyName: config.get('ES.USER_ACHIEVEMENT_PROPERTY_NAME'),
+    relateTopResource: 'achievementprovider',
     relateKey: 'achievementsProviderId',
     validate: payload => validProperties(payload, ['userId', 'achievementsProviderId'])
   },
   externalprofile: {
     propertyName: config.get('ES.USER_EXTERNALPROFILE_PROPERTY_NAME'),
+    relateTopResource: 'organization',
     relateKey: 'organizationId',
     validate: payload => validProperties(payload, ['userId', 'organizationId'])
   },
   userattribute: {
     propertyName: config.get('ES.USER_ATTRIBUTE_PROPERTY_NAME'),
+    relateTopResource: 'attribute',
     relateKey: 'attributeId',
     validate: payload => validProperties(payload, ['userId', 'attributeId']),
     isNested: true // For ES index creation
   },
   userrole: {
     propertyName: config.get('ES.USER_ROLE_PROPERTY_NAME'),
+    relateTopResource: 'role',
     relateKey: 'roleId',
     validate: payload => validProperties(payload, ['userId', 'roleId'])
   },
   userskill: {
     propertyName: config.get('ES.USER_SKILL_PROPERTY_NAME'),
+    relateTopResource: 'skill',
     relateKey: 'skillId',
     validate: payload => validProperties(payload, ['userId', 'skillId'])
   }