#21 - Listen on aggregate submission topic events

Author: callmekatootie

README.md

@@ -20,7 +20,7 @@ The following parameters can be set in config files or in env variables:
     if provided, it can be either path to private key file or private key content
 - KAFKA_GROUP_ID: consumer group id; default value: 'scorecard-processor'
 - REVIEW_TOPIC : Review topic, default value is 'submission.notification.score'
-- CREATE_SUBMISSION_TOPIC : create submission topic, default value is 'submission.notification.create'
+- AGGREGATE_SUBMISSION_TOPIC : aggregate submission topic, default value is 'submission.notification.aggregate'
 - SUBMISSION_API_URL: submission api url, default is 'https://api.topcoder-dev.com/v5'
 - CHALLENGE_API_URL: challenge API URL, default is 'https://api.topcoder-dev.com/v4/challenges'
 - SCORECARD_API_URL: scorecard API URL, default is 'http://localhost:4000/scorecards'
@@ -53,7 +53,7 @@ Configuration for the tests is at `config/test.js`, only add such new configurat
 - note that the zookeeper server is at localhost:2181, and Kafka server is at localhost:9092
 - use another terminal, go to same directory, create the needed topics:
   `bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic submission.notification.score`
-  `bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic submission.notification.create`
+  `bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic submission.notification.aggregate`
   `bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic avscan.action.scan`
   `bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic or.action.review`
 - verify that the topics are created:

Verification.md

@@ -37,10 +37,10 @@ curl --request POST --url https://topcoder-dev.auth0.com/oauth/token --header 'c
   `bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic avscan.action.scan --from-beginning`
 - start kafka-console-consumer to listen to topic `or.action.review`:
   `bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic or.action.review --from-beginning`
-- start kafka-console-producer to write messages to `submission.notification.create` topic:
-  `bin/kafka-console-producer.sh --broker-list localhost:9092 --topic submission.notification.create`
+- start kafka-console-producer to write messages to `submission.notification.aggregate` topic:
+  `bin/kafka-console-producer.sh --broker-list localhost:9092 --topic submission.notification.aggregate`
 - write message:
-  `{ "topic": "submission.notification.create", "originator": "or-app", "timestamp": "2019-02-25T00:00:00", "mime-type": "application/json", "payload": { "resource": "submission", "id": "104366f8-f46b-45db-a971-11bc69e6c8ff", "type": "Contest Submission", "url": "https://s3.amazonaws.com/topcoder-dev-submissions-dmz/30054740-8547899-SUBMISSION_ZIP-1554188341581.zip", "memberId": 8547899, "challengeId": 30049360, "created": "2019-04-02T06:59:29.785Z", "updated": "2019-04-02T06:59:29.785Z", "createdBy": "TonyJ", "updatedBy": "TonyJ", "submissionPhaseId": 764644, "fileType": "zip", "isFileSubmission": false } }`
+  `{ "topic": "submission.notification.aggregate", "originator": "or-app", "timestamp": "2019-02-25T00:00:00", "mime-type": "application/json", "payload": { "originalTopic": "submission.notification.create", "resource": "submission", "id": "104366f8-f46b-45db-a971-11bc69e6c8ff", "type": "Contest Submission", "url": "https://s3.amazonaws.com/topcoder-dev-submissions-dmz/30054740-8547899-SUBMISSION_ZIP-1554188341581.zip", "memberId": 8547899, "challengeId": 30049360, "created": "2019-04-02T06:59:29.785Z", "updated": "2019-04-02T06:59:29.785Z", "createdBy": "TonyJ", "updatedBy": "TonyJ", "submissionPhaseId": 764644, "fileType": "zip", "isFileSubmission": false } }`
 - watch the app console, it should show logging of processing the message:
 ```
 debug: Get M2M token

config/default.js

@@ -18,6 +18,8 @@ module.exports = {
   REVIEW_TOPIC: process.env.REVIEW_TOPIC || 'submission.notification.score',
   // Kafka topic related to create submission
   CREATE_SUBMISSION_TOPIC: process.env.CREATE_SUBMISSION_TOPIC || 'submission.notification.create',
+  // Kafka topic related to all events related to submission
+  AGGREGATE_SUBMISSION_TOPIC: process.env.AGGREGATE_SUBMISSION_TOPIC || 'submission.notification.aggregate',
 
   SUBMISSION_API_URL: process.env.SUBMISSION_API_URL || 'https://api.topcoder-dev.com/v5',
 

src/app.js

@@ -46,7 +46,8 @@ const dataHandler = (messageSet, topic, partition) => Promise.each(messageSet, a
   return (async () => {
     if (topic === config.REVIEW_TOPIC) {
       await ReviewProcessorService.processReview(messageJSON)
-    } else if (topic === config.CREATE_SUBMISSION_TOPIC) {
+    } else if (topic === config.AGGREGATE_SUBMISSION_TOPIC &&
+      messageJSON.payload.originalTopic === config.CREATE_SUBMISSION_TOPIC) {
       await SubmissionProcessorService.processSubmission(messageJSON)
     } else {
       throw new Error(`Invalid topic: ${topic}`)
@@ -76,7 +77,7 @@ function check () {
 }
 
 if (consumer) {
-  const topics = [config.REVIEW_TOPIC, config.CREATE_SUBMISSION_TOPIC]
+  const topics = [config.REVIEW_TOPIC, config.AGGREGATE_SUBMISSION_TOPIC]
 
   consumer
     .init([{

src/services/SubmissionProcessorService.js

@@ -62,7 +62,8 @@ function generateScoreSystemMessage (message, scoreSystem) {
   }
   if (scoreSystem.topic === 'or.action.review') {
     const payload = message.payload
-    payload.eventType = message.topic === config.CREATE_SUBMISSION_TOPIC ? 'CREATE' : 'UPDATE'
+    payload.eventType = 'CREATE'
+    delete payload.originalTopic
     return {
       topic: scoreSystem.topic,
       originator: 'tc-scorecard-processor',
@@ -164,6 +165,7 @@ processSubmission.schema = {
     timestamp: joi.date().required(),
     'mime-type': joi.string().required(),
     payload: joi.object().keys({
+      originalTopic: joi.string().valid(config.CREATE_SUBMISSION_TOPIC).required(),
       resource: joi.string().valid('submission').required(),
       id: joi.string().guid().required(),
       type: joi.string().valid('Contest Submission').required(),

test/common/testData.js

@@ -54,7 +54,7 @@ const testTopics = {
       }
     }
   },
-  'submission.notification.create': {
+  'submission.notification.aggregate': {
     operation: 'create',
     requiredFields: ['topic', 'originator', 'timestamp', 'mime-type',
       'payload.resource', 'payload.id', 'payload.url',
@@ -65,11 +65,12 @@ const testTopics = {
     dateFields: ['timestamp'],
     booleanFields: [],
     testMessage: {
-      topic: 'submission.notification.create',
+      topic: 'submission.notification.aggregate',
       originator: 'or-app',
       timestamp: '2019-02-25T00:00:00',
       'mime-type': 'application/json',
       payload: {
+        originalTopic: 'submission.notification.create',
         resource: 'submission',
         id: '104366f8-f46b-45db-a971-11bc69e6c8ff',
         type: 'Contest Submission',

test/common/testHelper.js

@@ -203,7 +203,7 @@ async function consumeMessages () {
   // consume and commit all not processed messages
   const consumer = new Kafka.GroupConsumer(helper.getKafkaOptions())
   await consumer.init([{
-    subscriptions: [config.REVIEW_TOPIC, config.CREATE_SUBMISSION_TOPIC,
+    subscriptions: [config.REVIEW_TOPIC, config.AGGREGATE_SUBMISSION_TOPIC,
       testData.avScanTopic, testData.reviewActionTopic],
     handler: (messageSet, topic, partition) => Promise.each(messageSet,
       (m) => consumer.commitOffset({ topic, partition, offset: m.offset }))

test/e2e/review.processor.test.js

@@ -83,7 +83,7 @@ describe('Topcoder - Scorecard Review Processor E2E Test', () => {
     testHelper.assertErrorMessage('Invalid or not supported eventType: INVALID_TYPE')
   })
 
-  for (const op of ['create', 'update']) {
+  for (const op of ['create']) {
     let { requiredFields, integerFields, stringFields, testMessage } = testTopics[op]
 
     it('test invalid parameters, field submissionId incorrect', async () => {

test/e2e/submission.processor.test.js

@@ -26,7 +26,7 @@ describe('Topcoder - Scorecard Submission Processor E2E Test', () => {
     testHelper.clearInterceptedLogging()
   })
 
-  for (const topic of [config.CREATE_SUBMISSION_TOPIC]) {
+  for (const topic of [config.AGGREGATE_SUBMISSION_TOPIC]) {
     const { operation, requiredFields, integerFields, stringFields, dateFields,
       booleanFields, testMessage } = testTopics[topic]
 

test/unit/review.processor.test.js

@@ -79,7 +79,7 @@ describe('Topcoder - Scorecard Review Processor Unit Test', () => {
     }
   })
 
-  for (const op of ['create', 'update']) {
+  for (const op of ['create']) {
     let { requiredFields, integerFields, stringFields, testMessage } = testTopics[op]
 
     it('test invalid parameters, field submissionId incorrect', async () => {

test/unit/submission.processor.test.js

@@ -24,7 +24,7 @@ describe('Topcoder - Scorecard Submission Processor Unit Test', () => {
     testHelper.clearInterceptedLogging()
   })
 
-  for (const topic of [config.CREATE_SUBMISSION_TOPIC]) {
+  for (const topic of [config.AGGREGATE_SUBMISSION_TOPIC]) {
     const { operation, requiredFields, integerFields, stringFields, dateFields,
       booleanFields, testMessage } = testTopics[topic]