Retry action

README.md

@@ -20,7 +20,9 @@ The following parameters can be set in config files or in env variables:
 - `KAFKA_CLIENT_CERT_KEY`: Kafka connection private key, optional;
     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_MESSAGE_ORIGINATOR`: The originator value for the kafka messages
 - `KAFKA_GROUP_ID`: the Kafka group id
+- `topics.KAFKA_ERROR_TOPIC`: the error topic at which bus api will publish any errors
 - `topics.TAAS_JOB_CREATE_TOPIC`: the create job entity Kafka message topic
 - `topics.TAAS_JOB_UPDATE_TOPIC`: the update job entity Kafka message topic
 - `topics.TAAS_JOB_DELETE_TOPIC`: the delete job entity Kafka message topic
@@ -41,6 +43,10 @@ The following parameters can be set in config files or in env variables:
 - `topics.TAAS_ROLE_CREATE_TOPIC`: the create role entity Kafka message topic
 - `topics.TAAS_ROLE_UPDATE_TOPIC`: the update role entity Kafka message topic
 - `topics.TAAS_ROLE_DELETE_TOPIC`: the delete role entity Kafka message topic
+- `topics.TAAS_ACTION_RETRY_TOPIC`: the retry process Kafka message topic
+- `MAX_RETRY`: maximum allowed retry count for failed operations for sending `taas.action.retry` message
+- `BASE_RETRY_DELAY`: base amount of retry delay (ms) for failed operations
+- `BUSAPI_URL`: Topcoder Bus API URL
 - `esConfig.HOST`: Elasticsearch host
 - `esConfig.AWS_REGION`: The Amazon region to use when using AWS Elasticsearch service
 - `esConfig.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
@@ -56,6 +62,7 @@ The following parameters can be set in config files or in env variables:
 - `auth0.AUTH0_CLIENT_ID`: Auth0 client id, used to get TC M2M token
 - `auth0.AUTH0_CLIENT_SECRET`: Auth0 client secret, used to get TC M2M token
 - `auth0.AUTH0_PROXY_SERVER_URL`: Proxy Auth0 URL, used to get TC M2M token
+- `auth0.TOKEN_CACHE_TIME`: Auth0 token cache time, used to get TC M2M token
 
 - `zapier.ZAPIER_COMPANYID_SLUG`: your company id in zapier; numeric value
 - `zapier.ZAPIER_CONTACTID_SLUG`: your contact id in zapier; numeric value
@@ -103,6 +110,22 @@ The following parameters can be set in config files or in env variables:
     npm run lint:fix
     ```
 
+4. Local config
+
+   In the `taas-es-processor` root directory create `.env` file with the next environment variables. Values for **Auth0 config** should be shared with you on the forum.<br>
+
+      ```bash
+      # Auth0 config
+      AUTH0_URL=
+      AUTH0_AUDIENCE=
+      AUTH0_AUDIENCE_UBAHN=
+      AUTH0_CLIENT_ID=
+      AUTH0_CLIENT_SECRET=
+      ```
+
+      - Values from this file would be automatically used by many `npm` commands.
+      - ⚠️ Never commit this file or its copy to the repository!
+
 5. Start the processor and health check dropin
 
     ```bash

config/default.js

@@ -1,7 +1,7 @@
 /**
  * The default configuration file.
  */
-
+require('dotenv').config()
 module.exports = {
   PORT: process.env.PORT || 3001,
   LOG_LEVEL: process.env.LOG_LEVEL || 'debug',
@@ -14,8 +14,12 @@ module.exports = {
 
   // Kafka group id
   KAFKA_GROUP_ID: process.env.KAFKA_GROUP_ID || 'taas-es-processor',
+  // The originator value for the kafka messages
+  KAFKA_MESSAGE_ORIGINATOR: process.env.KAFKA_MESSAGE_ORIGINATOR || 'taas-es-processor',
 
   topics: {
+    // The error topic at which bus api will publish any errors
+    KAFKA_ERROR_TOPIC: process.env.KAFKA_ERROR_TOPIC || 'common.error.reporting',
     // topics for job service
     TAAS_JOB_CREATE_TOPIC: process.env.TAAS_JOB_CREATE_TOPIC || 'taas.job.create',
     TAAS_JOB_UPDATE_TOPIC: process.env.TAAS_JOB_UPDATE_TOPIC || 'taas.job.update',
@@ -42,8 +46,17 @@ module.exports = {
     // topics for role service
     TAAS_ROLE_CREATE_TOPIC: process.env.TAAS_ROLE_CREATE_TOPIC || 'taas.role.requested',
     TAAS_ROLE_UPDATE_TOPIC: process.env.TAAS_ROLE_UPDATE_TOPIC || 'taas.role.update',
-    TAAS_ROLE_DELETE_TOPIC: process.env.TAAS_ROLE_DELETE_TOPIC || 'taas.role.delete'
+    TAAS_ROLE_DELETE_TOPIC: process.env.TAAS_ROLE_DELETE_TOPIC || 'taas.role.delete',
+    // special kafka topics
+    TAAS_ACTION_RETRY_TOPIC: process.env.TAAS_ACTION_RETRY_TOPIC || 'taas.action.retry'
+
   },
+  // maximum allowed retry count for failed operations for sending `action.retry` message
+  MAX_RETRY: process.env.MAX_RETRY || 3,
+  // base amount of retry delay for failed operations
+  BASE_RETRY_DELAY: process.env.BASE_RETRY_DELAY || 500,
+  // Topcoder Bus API URL
+  BUSAPI_URL: process.env.BUSAPI_URL || 'https://api.topcoder-dev.com/v5',
 
   esConfig: {
     HOST: process.env.ES_HOST || 'http://localhost:9200',
@@ -67,7 +80,8 @@ module.exports = {
     AUTH0_AUDIENCE: process.env.AUTH0_AUDIENCE,
     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
+    AUTH0_PROXY_SERVER_URL: process.env.AUTH0_PROXY_SERVER_URL,
+    TOKEN_CACHE_TIME: process.env.TOKEN_CACHE_TIME
   },
 
   zapier: {

config/test.js

@@ -6,5 +6,7 @@ module.exports = {
   zapier: {
     ZAPIER_SWITCH: process.env.ZAPIER_SWITCH || 'ON',
     ZAPIER_JOB_CANDIDATE_SWITCH: process.env.ZAPIER_JOB_CANDIDATE_SWITCH || 'ON'
-  }
+  },
+  // don't retry actions during tests because tests for now don't expect it and should be updated first
+  MAX_RETRY: 0,
 }

package.json

@@ -34,10 +34,12 @@
   "dependencies": {
     "@elastic/elasticsearch": "^7.9.1",
     "@hapi/joi": "^15.1.0",
+    "@topcoder-platform/topcoder-bus-api-wrapper": "github:topcoder-platform/tc-bus-api-wrapper",
     "async-mutex": "^0.2.4",
     "aws-sdk": "^2.476.0",
     "bluebird": "^3.5.5",
     "config": "^3.1.0",
+    "dotenv": "^10.0.0",
     "get-parameter-names": "^0.3.0",
     "lodash": "^4.17.20",
     "no-kafka": "^3.4.3",

src/app.js

@@ -16,14 +16,14 @@ const WorkPeriodProcessorService = require('./services/WorkPeriodProcessorServic
 const InterviewProcessorService = require('./services/InterviewProcessorService')
 const WorkPeriodPaymentProcessorService = require('./services/WorkPeriodPaymentProcessorService')
 const RoleProcessorService = require('./services/RoleProcessorService')
+const ActionProcessorService = require('./services/ActionProcessorService')
 const Mutex = require('async-mutex').Mutex
 const events = require('events')
 
 const eventEmitter = new events.EventEmitter()
 
 // healthcheck listening port
 process.env.PORT = config.PORT
-
 const localLogger = {
   info: (message) => logger.info({ component: 'app', message }),
   debug: (message) => logger.debug({ component: 'app', message }),
@@ -57,7 +57,9 @@ const topicServiceMapping = {
   // role
   [config.topics.TAAS_ROLE_CREATE_TOPIC]: RoleProcessorService.processCreate,
   [config.topics.TAAS_ROLE_UPDATE_TOPIC]: RoleProcessorService.processUpdate,
-  [config.topics.TAAS_ROLE_DELETE_TOPIC]: RoleProcessorService.processDelete
+  [config.topics.TAAS_ROLE_DELETE_TOPIC]: RoleProcessorService.processDelete,
+  // action
+  [config.topics.TAAS_ACTION_RETRY_TOPIC]: ActionProcessorService.processRetry
 }
 
 // Start kafka consumer
@@ -179,5 +181,6 @@ if (!module.parent) {
 
 module.exports = {
   initConsumer,
-  eventEmitter
+  eventEmitter,
+  topicServiceMapping
 }