add tax form processor (#2)

* add tax form processor

* fix tax form trait name

Author: xxcxy

README.md

@@ -24,6 +24,7 @@ The following parameters can be set in config files or in env variables:
 - `STANDARD_TERMS_ID`: The Topcoder standard terms id
 - `NDA_TERMS_ID`: The NDA terms id
 - `TERMS_USER_AGREEMENT_TOPIC`: The kafka topic on which the processor will listen to terms agreement events
+- `USER_TAXFORM_UPDATE_TOPIC`: The Kafka topic to which to listen to user tax form updated events
 - `auth0.AUTH0_URL`: Auth0 URL, used to get TC M2M token
 - `auth0.AUTH0_AUDIENCE`: Auth0 audience, used to get TC M2M token
 - `auth0.AUTH0_CLIENT_ID`: Auth0 client id, used to get TC M2M token

Verification.md

@@ -27,6 +27,13 @@ The following verification steps expect that the member-api is properly working
 - Malformed JSON message:
 `{"topic":"terms.notification.user.invalid","originator":"onboarding-api","timestamp":"2021-09-14T00:00:00.000Z","mime-type":"application/json","payload":{"userId":251280,"termsOfUseId":"0dedac8f-5a1a-4fe7-001f-e1d04dc65b7d","legacyId":123456,"created":"2021-09-14T00:00:`
 
+## Verify tax form processor service
+
+`docker exec -it onboarding-checklist-processor_kafka /opt/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic terms.notification.user.taxform.updated`
+
+- message for user tax form updated
+`{"topic":"terms.notification.user.taxform.updated","originator":"onboarding-api","timestamp":"2021-09-14T00:00:00.000Z","mime-type":"application/json","payload":{"userId":251280,"taxForm":"W-9(TopCoder)","Handle":"denis","created":"2021-09-14T00:00:00.000Z"}}`
+
 
 # Unit tests
 To run unit tests, execute the following command `npm run test`

config/default.js

@@ -22,7 +22,9 @@ module.exports = {
 
   topics: {
     // The Kafka topic to which to listen to user terms agreement events
-    TERMS_USER_AGREEMENT_TOPIC: process.env.TERMS_USER_AGREEMENT_TOPIC || 'terms.notification.user.agreed'
+    TERMS_USER_AGREEMENT_TOPIC: process.env.TERMS_USER_AGREEMENT_TOPIC || 'terms.notification.user.agreed',
+    // The Kafka topic to which to listen to user tax form updated events
+    USER_TAXFORM_UPDATE_TOPIC: process.env.USER_TAXFORM_UPDATE_TOPIC || 'terms.notification.user.taxform.updated'
   },
 
   auth0: {

src/app.js

@@ -9,6 +9,7 @@ const healthcheck = require('topcoder-healthcheck-dropin')
 const logger = require('./common/logger')
 const helper = require('./common/helper')
 const TermsAgreementProcessorService = require('./services/TermsAgreementProcessorService')
+const TaxFormProcessorService = require('./services/TaxFormProcessorService')
 const Mutex = require('async-mutex').Mutex
 const events = require('events')
 
@@ -23,7 +24,8 @@ const localLogger = {
 }
 
 const topicServiceMapping = {
-  [config.topics.TERMS_USER_AGREEMENT_TOPIC]: TermsAgreementProcessorService.processMessage
+  [config.topics.TERMS_USER_AGREEMENT_TOPIC]: TermsAgreementProcessorService.processMessage,
+  [config.topics.USER_TAXFORM_UPDATE_TOPIC]: TaxFormProcessorService.processMessage
 }
 
 // Start kafka consumer

src/common/constants.js

@@ -15,6 +15,10 @@ module.exports = {
   // The property name to use to store the nda terms in traits body
   NDA_TERMS_TRAIT_PROPERTY_NAME: 'nda_terms',
 
+  // The property name to use to store the tax form submitted in traits body
+  TAX_FORM_TRAIT_PROPERTY_NAME: 'tax_form_submitted',
+  TAX_FORM_TRAIT_PROPERTY_NAME_MAP: { 'w9_tax_form_submitted': 'W-9', 'w8ben_tax_form_submitted': 'W-8BEN' },
+
   CHECKLIST_STATUS: {
     COMPLETED: 'completed'
   },

src/services/TaxFormProcessorService.js

@@ -0,0 +1,78 @@
+/**
+ * This service processes the events fired when users agree to Standard Topcoder terms and NDA terms
+ */
+
+const Joi = require('@hapi/joi')
+const _ = require('lodash')
+const logger = require('../common/logger')
+const helper = require('../common/helper')
+const constants = require('../common/constants')
+
+/**
+  * Process the terms agreement event message
+  * @param {Object} message the kafka message
+  */
+async function processMessage (message) {
+  logger.info(`Process message of taxForm ${message.payload.taxForm}, userId ${message.payload.userId}, handle ${message.payload.Handle}`)
+  // Get the user handle from members api
+  const handle = _.get(message.payload, 'Handle')
+  const taxForm = _.get(message.payload, 'taxForm')
+
+  // Get the member Onboarding Checklist traits
+  const onboardingChecklistTraits = await helper.getMemberTraits(handle, constants.ONBOARDING_CHECKLIST_TRAIT_ID)
+
+  // Initialize the terms traits data object
+  const termsTraitsData = {
+    status: constants.CHECKLIST_STATUS.COMPLETED,
+    message: constants.CHECKLIST_MESSAGE.SUCCESS,
+    date: new Date().getTime()
+  }
+
+  const traitsBodyPropertyName = _.findKey(constants.TAX_FORM_TRAIT_PROPERTY_NAME_MAP, v => _.startsWith(taxForm, v)) || constants.TAX_FORM_TRAIT_PROPERTY_NAME
+  if (onboardingChecklistTraits.length === 0) {
+    const body = [{
+      categoryName: constants.ONBOARDING_CHECKLIST_CATEGORY_NAME,
+      traitId: constants.ONBOARDING_CHECKLIST_TRAIT_ID,
+      traits: {
+        data: [{
+          [traitsBodyPropertyName]: termsTraitsData
+        }]
+      }
+    }]
+    await helper.saveMemberTraits(handle, body, true)
+  } else {
+    // Onboarding checklist traits already exists for the member
+    // An update of the trait should be performed
+
+    // Update the currently processed terms property in the request body
+    onboardingChecklistTraits[0].traits.data[0][traitsBodyPropertyName] = termsTraitsData
+    await helper.saveMemberTraits(handle, onboardingChecklistTraits, false)
+  }
+
+  logger.info(`Successfully Processed message of taxForm ${taxForm}, userId ${message.payload.userId}, handle ${handle}`)
+}
+
+processMessage.schema = {
+  message: Joi.object()
+    .keys({
+      topic: Joi.string().required(),
+      originator: Joi.string().required(),
+      timestamp: Joi.date().required(),
+      'mime-type': Joi.string().required(),
+      payload: Joi.object()
+        .keys({
+          userId: Joi.alternatives(Joi.string(), Joi.positiveId()).required(),
+          taxForm: Joi.string().required(),
+          Handle: Joi.string().required(),
+          created: Joi.date()
+        })
+        .required()
+    })
+    .required()
+}
+
+module.exports = {
+  processMessage
+}
+
+logger.buildService(module.exports, 'TaxFormProcessorService')

test/common/taxFormTestData.js

@@ -0,0 +1,34 @@
+/**
+ * This file defines common data used in tests.
+ */
+
+const denisUserId = '251280'
+const upbeatUserHandle = 'upbeat'
+
+const createTaxFormTraitTestMessage = {
+  'topic': 'terms.notification.user.agreed',
+  'originator': 'onboarding-api',
+  'timestamp': '2021-09-14T00:00:00.000Z',
+  'mime-type': 'application/json',
+  'payload': {
+    'userId': denisUserId,
+    'taxForm': 'W-9(TopCoder)',
+    'Handle': 'denis',
+    'created': '2021-09-14T00:00:00.000Z'
+  }
+}
+
+const requiredFields = ['originator', 'timestamp', 'mime-type', 'topic', 'payload.userId', 'payload.taxForm', 'payload.Handle']
+
+const stringFields = ['topic', 'originator', 'mime-type', 'payload.taxForm', 'payload.Handle']
+
+const dateFields = ['timestamp', 'payload.created']
+
+module.exports = {
+  createTaxFormTraitTestMessage,
+  requiredFields,
+  stringFields,
+  dateFields,
+  denisUserId,
+  upbeatUserHandle
+}

test/unit/TaxFormProcessorService.test.js

@@ -0,0 +1,151 @@
+/**
+ * Unit tests for Onboarding Checklist Tax Form updated processor service
+ */
+
+process.env.NODE_ENV = 'test'
+
+require('../../src/bootstrap')
+const _ = require('lodash')
+const should = require('should')
+const logger = require('../../src/common/logger')
+const service = require('../../src/services/TaxFormProcessorService')
+const {
+  createTaxFormTraitTestMessage,
+  requiredFields,
+  stringFields,
+  dateFields,
+  upbeatUserHandle
+} = require('../common/taxFormTestData')
+
+describe('Topcoder Onboarding Checklist - Tax Form Processor Service Unit Tests', () => {
+  let infoLogs = []
+  let errorLogs = []
+  let debugLogs = []
+  const info = logger.info
+  const error = logger.error
+  const debug = logger.debug
+
+  /**
+   * Assert validation error
+   * @param err the error
+   * @param message the message
+   */
+  function assertValidationError (err, message) {
+    err.isJoi.should.be.true()
+    should.equal(err.name, 'ValidationError')
+    err.details.map(x => x.message).should.containEql(message)
+    errorLogs.should.not.be.empty()
+  }
+
+  /**
+   * Assert there is given info message
+   * @param message the message
+   */
+  function assertInfoMessage (message) {
+    infoLogs.should.not.be.empty()
+    infoLogs.some(x => String(x).includes(message)).should.be.true()
+  }
+
+  before(async () => {
+    // inject logger with log collector
+    logger.info = (message) => {
+      infoLogs.push(message)
+      info(message)
+    }
+    logger.error = (message) => {
+      errorLogs.push(message)
+      error(message)
+    }
+    logger.debug = (message) => {
+      debugLogs.push(message)
+      debug(message)
+    }
+  })
+
+  after(async () => {
+    // restore logger
+    logger.error = error
+    logger.info = info
+    logger.debug = debug
+  })
+
+  beforeEach(() => {
+    // clear logs
+    infoLogs = []
+    errorLogs = []
+    debugLogs = []
+  })
+
+  for (const requiredField of requiredFields) {
+    it(`test process message with invalid parameters, required field ${requiredField} is missing`, async () => {
+      let message = _.cloneDeep(createTaxFormTraitTestMessage)
+      message = _.omit(message, requiredField)
+      try {
+        await service.processMessage(message)
+      } catch (err) {
+        assertValidationError(err, `"${_.last(requiredField.split('.'))}" is required`)
+        return
+      }
+      throw new Error('should not throw error here')
+    })
+  }
+
+  for (const stringField of stringFields) {
+    it(`test process message with invalid parameters, invalid string type field ${stringField}`, async () => {
+      const message = _.cloneDeep(createTaxFormTraitTestMessage)
+      _.set(message, stringField, 123)
+      try {
+        await service.processMessage(message)
+      } catch (err) {
+        assertValidationError(err, `"${_.last(stringField.split('.'))}" must be a string`)
+        return
+      }
+      throw new Error('should not throw error here')
+    })
+
+    it(`test process message with invalid parameters, empty string field ${stringField}`, async () => {
+      const message = _.cloneDeep(createTaxFormTraitTestMessage)
+      _.set(message, stringField, '')
+      try {
+        await service.processMessage(message)
+      } catch (err) {
+        assertValidationError(err, `"${_.last(stringField.split('.'))}" is not allowed to be empty`)
+        return
+      }
+      throw new Error('should not throw error here')
+    })
+  }
+
+  for (const dateField of dateFields) {
+    it(`test process message with invalid parameters, invalid date type field ${dateField}`, async () => {
+      const message = _.cloneDeep(createTaxFormTraitTestMessage)
+      _.set(message, dateField, 'abc')
+      try {
+        await service.processMessage(message)
+      } catch (err) {
+        assertValidationError(err,
+          `"${_.last(dateField.split('.'))}" must be a number of milliseconds or valid date string`)
+        return
+      }
+      throw new Error('should not throw error here')
+    })
+  }
+
+  it(`test process message - onboarding checklist does not already exist for the member`, async () => {
+    const message = _.cloneDeep(createTaxFormTraitTestMessage)
+    await service.processMessage(message)
+
+    assertInfoMessage(`Process message of taxForm ${message.payload.taxForm}, userId ${message.payload.userId}, handle ${message.payload.Handle}`)
+    assertInfoMessage(`Successfully Processed message of taxForm ${message.payload.taxForm}, userId ${message.payload.userId}, handle ${message.payload.Handle}`)
+  })
+
+  it(`test process message - onboarding checklist already exist for the member`, async () => {
+    const message = _.cloneDeep(createTaxFormTraitTestMessage)
+    _.set(message, 'payload.Handle', upbeatUserHandle)
+
+    await service.processMessage(message)
+
+    assertInfoMessage(`Process message of taxForm ${message.payload.taxForm}, userId ${message.payload.userId}, handle ${message.payload.Handle}`)
+    assertInfoMessage(`Successfully Processed message of taxForm ${message.payload.taxForm}, userId ${message.payload.userId}, handle ${message.payload.Handle}`)
+  })
+})