Fix issue with double payments

Author: ThomasKranitsas

src/app.js

@@ -60,6 +60,9 @@ const dataHandler = (messageSet, topic, partition) => Promise.each(messageSet, a
   }
 
   try {
+    // delay for a random amount of time between 5-20 sec
+    // to minimize the chance of having two processes doing the same at the same time
+    await helper.delay(helper.getRandomInt(5 * 1000, 20 * 1000))
     await processorService.processUpdate(messageJSON)
     logger.debug('Successfully processed message')
   } catch (err) {

src/common/helper.js

@@ -146,6 +146,29 @@ async function getCopilotId (challengeId) {
   return _.get(_.head(res.body), 'memberId')
 }
 
+/**
+ * Creates a delay promise
+ * @param {Number} ms the milliseconds to delay
+ * @returns a promise
+ */
+function delay (ms) {
+  return new Promise(resolve => {
+      setTimeout(resolve, ms)
+  })
+}
+
+/**
+ * Generate a random integer
+ * @param {Number} min the min
+ * @param {Number} max the max
+ * @returns the random integer
+ */
+function getRandomInt(min, max) {
+  min = Math.ceil(min)
+  max = Math.floor(max)
+  return Math.floor(Math.random() * (max - min + 1)) + min
+}
+
 module.exports = {
   getInformixConnection,
   getKafkaOptions,
@@ -154,5 +177,7 @@ module.exports = {
   putRequest,
   postRequest,
   getUserId,
-  getCopilotId
+  getCopilotId,
+  delay,
+  getRandomInt
 }

src/services/paymentService.js

@@ -46,14 +46,14 @@ const INSERT_PAYMENT_STATUS_REASON_XREF = 'INSERT INTO payment_detail_status_rea
  * @param {String} sql the sql
  * @return {Object} Informix statement
  */
-async function prepare (connection, sql) {
+async function prepare(connection, sql) {
   logger.debug(`Preparing SQL ${sql}`)
   const stmt = await connection.prepareAsync(sql)
   return Promise.promisifyAll(stmt)
 }
 
-async function paymentExists(payment) {
-  const connection = await helper.getInformixConnection()
+async function paymentExists(payment, connection) {
+  if (!connection) connection = await helper.getInformixConnection()
   try {
     const query = util.format(QUERY_PAYMENT, payment.memberId, payment.v5ChallengeId, payment.typeId)
     logger.debug(`Checking if paymentExists - ${query}`)
@@ -62,30 +62,38 @@ async function paymentExists(payment) {
     logger.error(`Error in 'paymentExists' ${e}`)
     throw e
   } finally {
-    await connection.closeAsync()
+    if (!existingConnection) await connection.closeAsync()
   }
 }
 
 /**
  * Create payment and save it to db
  * @param {Object} payment the payment info
  */
-async function createPayment (payment) {
-  const connection = await helper.getInformixConnection()
-  const paymentDetailId = await paymentDetailIdGen.getNextId()
-  const paymentId = await paymentIdGen.getNextId()
+async function createPayment(payment) {
   try {
+    const connection = await helper.getInformixConnection()
     await connection.beginTransactionAsync()
-    const insertDetail = await prepare(connection, INSERT_PAYMENT_DETAIL)
-    await insertDetail.executeAsync([paymentDetailId, payment.amount, payment.amount, payment.statusId, payment.modificationRationaleId, payment.desc, payment.typeId, payment.methodId, payment.projectId, payment.charityInd, payment.amount, payment.installmentNumber, payment.createUser, payment.v5ChallengeId])
-    const insertPayment = await prepare(connection, INSERT_PAYMENT)
-    await insertPayment.executeAsync([paymentId, payment.memberId, paymentDetailId])
-    const insertDetailXref = await prepare(connection, INSERT_PAYMENT_DETAIL_XREF)
-    await insertDetailXref.executeAsync([paymentId, paymentDetailId])
-    const insertStatusXref = await prepare(connection, INSERT_PAYMENT_STATUS_REASON_XREF)
-    await insertStatusXref.executeAsync([paymentDetailId, config.V5_PAYMENT_DETAIL_STATUS_REASON_ID])
-    logger.info(`Payment ${paymentId} with detail ${paymentDetailId} has been inserted`)
-    await connection.commitTransactionAsync()
+
+    const paymentExists = await paymentExists(payment, connection)
+    logger.debug(`Payment Exists Response: ${JSON.stringify(paymentExists)}`)
+    if (!paymentExists || paymentExists.length === 0) {
+      const paymentDetailId = await paymentDetailIdGen.getNextId()
+      const paymentId = await paymentIdGen.getNextId()
+      const insertDetail = await prepare(connection, INSERT_PAYMENT_DETAIL)
+      await insertDetail.executeAsync([paymentDetailId, payment.amount, payment.amount, payment.statusId, payment.modificationRationaleId, payment.desc, payment.typeId, payment.methodId, payment.projectId, payment.charityInd, payment.amount, payment.installmentNumber, payment.createUser, payment.v5ChallengeId])
+      const insertPayment = await prepare(connection, INSERT_PAYMENT)
+      await insertPayment.executeAsync([paymentId, payment.memberId, paymentDetailId])
+      const insertDetailXref = await prepare(connection, INSERT_PAYMENT_DETAIL_XREF)
+      await insertDetailXref.executeAsync([paymentId, paymentDetailId])
+      const insertStatusXref = await prepare(connection, INSERT_PAYMENT_STATUS_REASON_XREF)
+      await insertStatusXref.executeAsync([paymentDetailId, config.V5_PAYMENT_DETAIL_STATUS_REASON_ID])
+      logger.info(`Payment ${paymentId} with detail ${paymentDetailId} has been inserted`)
+      await connection.commitTransactionAsync()
+    } else {
+      logger.error(`Payment Exists for ${payment.v5ChallengeId}, skipping - ${JSON.stringify(paymentExists)}`)
+      await connection.commitTransactionAsync()
+    }
   } catch (e) {
     logger.error(`Error in 'createPayment' ${e}, rolling back transaction`)
     await connection.rollbackTransactionAsync()

src/services/processorService.js

@@ -59,13 +59,7 @@ async function processUpdate(message) {
             typeId: config.WINNER_PAYMENT_TYPE_ID
           }, basePayment)
 
-          const paymentExists = await paymentService.paymentExists(payment)
-          logger.debug(`Payment Exists Response: ${JSON.stringify(paymentExists)}`)
-          if(!paymentExists || paymentExists.length === 0) {
-            await paymentService.createPayment(payment)
-          } else {
-            logger.error(`Payment Exists for ${v5ChallengeId}, skipping - ${JSON.stringify(paymentExists)}`)
-          }
+          await paymentService.createPayment(payment)
         }
       } catch (error) {
         logger.error(`For challenge ${v5ChallengeId}, add winner payments error: ${error}`)
@@ -86,13 +80,7 @@ async function processUpdate(message) {
             typeId: config.CHECKPOINT_WINNER_PAYMENT_TYPE_ID
           }, basePayment)
 
-          const paymentExists = await paymentService.paymentExists(payment)
-          logger.debug(`Payment Exists Response: ${JSON.stringify(paymentExists)}`)
-          if(!paymentExists || paymentExists.length === 0) {
-            await paymentService.createPayment(payment)
-          } else {
-            logger.error(`Payment Exists for ${v5ChallengeId}, skipping - ${JSON.stringify(paymentExists)}`)
-          }
+          await paymentService.createPayment(payment)
         }
       } catch (error) {
         logger.error(`For challenge ${v5ChallengeId}, add checkpoint winner payments error: ${error}`)
@@ -116,13 +104,7 @@ async function processUpdate(message) {
           desc: (copilotPaymentDesc ? copilotPaymentDesc : `${message.payload.name} - Copilot`),
           typeId: config.COPILOT_PAYMENT_TYPE_ID
         }, basePayment)
-        const paymentExists = await paymentService.paymentExists(copilotPayment)
-        logger.debug(`Copilot Payment Exists Response: ${JSON.stringify(paymentExists)}`)
-        if(!paymentExists || paymentExists.length === 0) {
-          await paymentService.createPayment(copilotPayment)
-        } else {
-          logger.error(`Copilot Payment Exists for ${v5ChallengeId}, skipping - ${JSON.stringify(paymentExists)}`)
-        }
+        await paymentService.createPayment(copilotPayment)
       } catch (error) {
         logger.error(`For challenge ${v5ChallengeId}, add copilot payments error: ${error}`)
       }