Merge pull request #69 from eisbilir/feature/new-action-topic

action topic and service added

Author: maxceem

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

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: {

package-lock.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",

package.json

@@ -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
 }

src/app.js

@@ -0,0 +1,40 @@
+/**
+ * This file defines application errors
+ */
+const util = require('util')
+
+/**
+  * Helper function to create generic error object with http status code
+  * @param {String} name the error name
+  * @param {Number} statusCode the http status code
+  * @returns {Function} the error constructor
+  * @private
+  */
+function createError (name, statusCode) {
+  /**
+    * The error constructor
+    * @param {String} message the error message
+    * @param {String} [cause] the error cause
+    * @constructor
+    */
+  function ErrorCtor (message, cause) {
+    Error.call(this)
+    Error.captureStackTrace(this)
+    this.message = message || name
+    this.cause = cause
+    this.httpStatus = statusCode
+  }
+
+  util.inherits(ErrorCtor, Error)
+  ErrorCtor.prototype.name = name
+  return ErrorCtor
+}
+
+module.exports = {
+  BadRequestError: createError('BadRequestError', 400),
+  UnauthorizedError: createError('UnauthorizedError', 401),
+  ForbiddenError: createError('ForbiddenError', 403),
+  NotFoundError: createError('NotFoundError', 404),
+  ConflictError: createError('ConflictError', 409),
+  InternalServerError: createError('InternalServerError', 500)
+}