Merge pull request #520 from topcoder-platform/dev

[PROD] Next Release

Author: sachin-maheshwari

.circleci/config.yml

@@ -68,8 +68,7 @@ workflows:
             branches:
               only:
                 - dev
-                - change-validatations-in-job-jc
-                - feature/enriching-skills-data-with-api-2
+                - feature/shapeup4-cqrs-update
 
       # Production builds are exectuted only on tagged commits to the
       # master branch.

config/default.js

@@ -95,6 +95,9 @@ module.exports = {
   KAFKA_ERROR_TOPIC: process.env.KAFKA_ERROR_TOPIC || 'common.error.reporting',
   // The originator value for the kafka messages
   KAFKA_MESSAGE_ORIGINATOR: process.env.KAFKA_MESSAGE_ORIGINATOR || 'taas-api',
+
+  // topics for error
+  TAAS_ERROR_TOPIC: process.env.TAAS_ERROR_TOPIC || 'taas.action.error',
   // topics for job service
   // the create job entity Kafka message topic
   TAAS_JOB_CREATE_TOPIC: process.env.TAAS_JOB_CREATE_TOPIC || 'taas.job.create',

data/demo-data.json

@@ -19,7 +19,7 @@
                 "a2b4bc11-c641-4a19-9eb7-33980378f82e"
             ],
             "status": "in-review",
-            "isApplicationPageActive": false,
+            "isApplicationPageActive": true,
             "minSalary": 100,
             "maxSalary": 200,
             "hoursPerWeek": 20,
@@ -51,7 +51,7 @@
                 "a2b4bc11-c641-4a19-9eb7-33980378f82e"
             ],
             "status": "in-review",
-            "isApplicationPageActive": false,
+            "isApplicationPageActive": true,
             "minSalary": 100,
             "maxSalary": 200,
             "hoursPerWeek": 80,
@@ -83,7 +83,7 @@
                 "a2b4bc11-c641-4a19-9eb7-33980378f82e"
             ],
             "status": "in-review",
-            "isApplicationPageActive": false,
+            "isApplicationPageActive": true,
             "minSalary": 100,
             "maxSalary": 200,
             "hoursPerWeek": 90,
@@ -115,7 +115,7 @@
                 "a2b4bc11-c641-4a19-9eb7-33980378f82e"
             ],
             "status": "in-review",
-            "isApplicationPageActive": false,
+            "isApplicationPageActive": true,
             "minSalary": 100,
             "maxSalary": 200,
             "hoursPerWeek": 20,
@@ -148,7 +148,7 @@
                 "0b104b7c-0792-4118-8bc7-a274e9ee19e3"
             ],
             "status": "closed",
-            "isApplicationPageActive": false,
+            "isApplicationPageActive": true,
             "minSalary": null,
             "maxSalary": null,
             "hoursPerWeek": null,

docs/swagger.yaml

@@ -118,6 +118,12 @@ paths:
           schema:
             type: integer
           description: The project id.
+        - in: query
+          name: isApplicationPageActive 
+          required: false
+          schema:
+            type: boolean 
+          description: Is application page active.
         - in: query
           name: projectIds
           required: false

docs/taas-ER-diagram.png

@@ -31,22 +31,22 @@ async function resetNotificationRecords () {
   // reset completed interview records
   localLogger.info('reset completed interview records')
   const pastTime = moment.duration(config.INTERVIEW_COMPLETED_PAST_TIME)
-  const endTimestamp = moment().subtract(pastTime).add(config.INTERVIEW_COMPLETED_MATCH_WINDOW).toDate()
+  const completedStartTimestamp = moment().subtract(pastTime).add(config.INTERVIEW_COMPLETED_MATCH_WINDOW).toDate()
   const completedInterview = await Interview.findById('9efd72c3-1dc7-4ce2-9869-8cca81d0adeb')
   const duration = 30
-  const completedStartTimestamp = moment().subtract(pastTime).subtract(30, 'm').toDate()
-  await completedInterview.update({ startTimestamp: completedStartTimestamp, duration, endTimestamp, status: Interviews.Status.Scheduled, guestNames: ['guest1', 'guest2'], hostName: 'hostName' })
+  const completedEndTimestamp = moment(completedStartTimestamp).clone().add(30, 'm').toDate()
+  await completedInterview.update({ startTimestamp: completedStartTimestamp, duration, endTimeStamp: completedEndTimestamp, status: Interviews.Status.Scheduled, guestNames: ['guest1', 'guest2'], hostName: 'hostName' })
+  const completedInterview2 = await Interview.findById('3144fa65-ea1a-4bec-81b0-7cb1c8845826')
+  await completedInterview2.update({ startTimestamp: completedStartTimestamp, duration, endTimeStamp: completedEndTimestamp, status: Interviews.Status.Scheduled, guestNames: ['guest1', 'guest2'], hostName: 'hostName' })
 
   // reset post interview candidate action reminder records
   localLogger.info('reset post interview candidate action reminder records')
-  const jobCandidate = await JobCandidate.findById('881a19de-2b0c-4bb9-b36a-4cb5e223bdb5')
+  const jobCandidate = await JobCandidate.findById('827ee401-df04-42e1-abbe-7b97ce7937ff')
   await jobCandidate.update({ status: 'interview' })
-  const c2Interview = await Interview.findById('077aa2ca-5b60-4ad9-a965-1b37e08a5046')
-  await c2Interview.update({ startTimestamp: moment().subtract(moment.duration(config.POST_INTERVIEW_ACTION_MATCH_WINDOW)).subtract(30, 'm').toDate(), duration, endTimestamp, guestNames: ['guest1', 'guest2'], hostName: 'hostName' })
-  const jobCandidateWithinOneDay = await JobCandidate.findById('827ee401-df04-42e1-abbe-7b97ce7937ff')
-  await jobCandidateWithinOneDay.update({ status: 'interview' })
-  const interviewWithinOneDay = await Interview.findById('3144fa65-ea1a-4bec-81b0-7cb1c8845826')
-  await interviewWithinOneDay.update({ startTimestamp: completedStartTimestamp, duration, endTimestamp, guestNames: ['guest1', 'guest2'], hostName: 'hostName' })
+  const c2Interview = await Interview.findById('3144fa65-ea1a-4bec-81b0-7cb1c8845826')
+  await c2Interview.update({ startTimestamp: moment().subtract(moment.duration(config.POST_INTERVIEW_ACTION_MATCH_WINDOW)).subtract(30, 'm').toDate(), duration, endTimeStamp: completedEndTimestamp, guestNames: ['guest1', 'guest2'], hostName: 'hostName' })
+  const c2InterviewR2 = await Interview.findById('b1f7ba76-640f-47e2-9463-59e51b51ec60')
+  await c2InterviewR2.update({ status: 'Scheduled', startTimestamp: moment().subtract(moment.duration(config.POST_INTERVIEW_ACTION_MATCH_WINDOW)).subtract(30, 'm').toDate(), duration, endTimeStamp: completedEndTimestamp, guestNames: ['guest1', 'guest2'], hostName: 'hostName' })
 
   // reset upcoming resource booking expiration records
   localLogger.info('reset upcoming resource booking expiration records')

scripts/demo-email-notifications/index.js

@@ -999,6 +999,26 @@ async function postEvent (topic, payload, options = {}) {
   await eventDispatcher.handleEvent(topic, { value: payload, options })
 }
 
+/**
+ * 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 postErrorEvent (topic, payload, action) {
+  _.set(payload, 'apiAction', action)
+  const client = getBusApiClient()
+  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 client.postEvent(message)
+}
+
 /**
  * Test if an error is document missing exception
  *
@@ -2094,6 +2114,7 @@ module.exports = {
   getM2MToken,
   getM2MUbahnToken,
   postEvent,
+  postErrorEvent,
   getBusApiClient,
   isDocumentMissingException,
   getProjects,

src/common/helper.js

@@ -0,0 +1,127 @@
+/**
+ * Interview Processor
+ */
+
+const _ = require('lodash')
+const helper = require('../common/helper')
+const config = require('config')
+
+const esClient = helper.getESClient()
+
+/**
+ * Updates jobCandidate via a painless script
+ *
+ * @param {String} jobCandidateId job candidate id
+ * @param {String} script script definition
+ */
+async function updateJobCandidateViaScript (jobCandidateId, script) {
+  await esClient.update({
+    index: config.get('esConfig.ES_INDEX_JOB_CANDIDATE'),
+    id: jobCandidateId,
+    body: { script },
+    refresh: 'wait_for'
+  })
+}
+
+/**
+ * Process request interview entity.
+ * Creates an interview record under jobCandidate.
+ *
+ * @param {Object} interview interview object
+ */
+async function processRequestInterview (interview) {
+  // add interview in collection if there's already an existing collection
+  // or initiate a new one with this interview
+  const script = {
+    source: `
+      ctx._source.containsKey("interviews")
+        ? ctx._source.interviews.add(params.interview)
+        : ctx._source.interviews = [params.interview]
+    `,
+    params: { interview }
+  }
+  await updateJobCandidateViaScript(interview.jobCandidateId, script)
+}
+
+/**
+ * Process update interview entity
+ * Updates the interview record under jobCandidate.
+ *
+ * @param {Object} interview interview object
+ */
+async function processUpdateInterview (interview) {
+  // if there's an interview with this id,
+  // update it
+  const script = {
+    source: `
+      if (ctx._source.containsKey("interviews")) {
+        def target = ctx._source.interviews.find(i -> i.id == params.interview.id);
+        if (target != null) {
+          for (prop in params.interview.entrySet()) {
+            target[prop.getKey()] = prop.getValue()
+          }
+        }
+      }
+    `,
+    params: { interview }
+  }
+  await updateJobCandidateViaScript(interview.jobCandidateId, script)
+}
+
+/**
+ * Process bulk (partially) update interviews entity.
+ * Currently supports status, updatedAt and updatedBy fields.
+ * Update Joi schema to allow more fields.
+ * (implementation should already handle new fields - just updating Joi schema should be enough)
+ *
+ * payload format:
+ * {
+ *   "jobCandidateId": {
+ *     "interviewId": { ...fields },
+ *     "interviewId2": { ...fields },
+ *     ...
+ *   },
+ *   "jobCandidateId2": { // like above... },
+ *   ...
+ * }
+ *
+ * @param {Object} jobCandidates job candidates
+ */
+async function processBulkUpdateInterviews (jobCandidates) {
+  // script to update & params
+  const script = {
+    source: `
+      def completedInterviews = params.jobCandidates[ctx._id];
+      for (interview in completedInterviews.entrySet()) {
+        def interviewId = interview.getKey();
+        def affectedFields = interview.getValue();
+        def target = ctx._source.interviews.find(i -> i.id == interviewId);
+        if (target != null) {
+          for (field in affectedFields.entrySet()) {
+            target[field.getKey()] = field.getValue();
+          }
+        }
+      }
+    `,
+    params: { jobCandidates }
+  }
+  // update interviews
+  await esClient.updateByQuery({
+    index: config.get('esConfig.ES_INDEX_JOB_CANDIDATE'),
+    body: {
+      script,
+      query: {
+        ids: {
+          values: _.keys(jobCandidates)
+        }
+      }
+    },
+    refresh: true
+  })
+}
+
+module.exports = {
+  processRequestInterview,
+  processUpdateInterview,
+  processBulkUpdateInterviews
+}

src/esProcessors/InterviewProcessor.js

@@ -0,0 +1,54 @@
+/**
+ * Jobcandidate Processor
+ */
+
+const config = require('config')
+const helper = require('../common/helper')
+
+const esClient = helper.getESClient()
+
+/**
+  * Process create entity
+  * @param {Object} entity entity object
+  */
+async function processCreate (entity) {
+  await esClient.create({
+    index: config.get('esConfig.ES_INDEX_JOB_CANDIDATE'),
+    id: entity.id,
+    body: entity,
+    refresh: 'wait_for'
+  })
+}
+
+/**
+  * Process update entity
+  * @param {Object} entity entity object
+  */
+async function processUpdate (entity) {
+  await esClient.update({
+    index: config.get('esConfig.ES_INDEX_JOB_CANDIDATE'),
+    id: entity.id,
+    body: {
+      doc: entity
+    },
+    refresh: 'wait_for'
+  })
+}
+
+/**
+  * Process delete entity
+  * @param {Object} entity entity object
+  */
+async function processDelete (entity) {
+  await esClient.delete({
+    index: config.get('esConfig.ES_INDEX_JOB_CANDIDATE'),
+    id: entity.id,
+    refresh: 'wait_for'
+  })
+}
+
+module.exports = {
+  processCreate,
+  processUpdate,
+  processDelete
+}

src/esProcessors/JobCandidateProcessor.js

@@ -0,0 +1,54 @@
+/**
+ * Job Processor
+ */
+
+const helper = require('../common/helper')
+const config = require('config')
+
+const esClient = helper.getESClient()
+
+/**
+ * Process create entity
+ * @param {Object} entity entity object
+ */
+async function processCreate (entity) {
+  await esClient.create({
+    index: config.get('esConfig.ES_INDEX_JOB'),
+    id: entity.id,
+    body: entity,
+    refresh: 'wait_for'
+  })
+}
+
+/**
+ * Process update entity
+ * @param {Object} entity entity object
+ */
+async function processUpdate (entity) {
+  await esClient.update({
+    index: config.get('esConfig.ES_INDEX_JOB'),
+    id: entity.id,
+    body: {
+      doc: entity
+    },
+    refresh: 'wait_for'
+  })
+}
+
+/**
+ * Process delete entity
+ * @param {Object} entity entity object
+ */
+async function processDelete (entity) {
+  await esClient.delete({
+    index: config.get('esConfig.ES_INDEX_JOB'),
+    id: entity.id,
+    refresh: 'wait_for'
+  })
+}
+
+module.exports = {
+  processCreate,
+  processUpdate,
+  processDelete
+}