cleaning...

Author: Sachin Maheshwari

src/producer.js

@@ -21,19 +21,20 @@ async function setupPgClient() {
       await pgClient.query(`LISTEN ${triggerFunction}`)
     }
     pgClient.on('notification', async (message) => {
+      const pl_randonseq = 'err-' + (new Date()).getTime().toString(36) + Math.random().toString(36).slice(2)
       try {
         const payload = JSON.parse(message.payload)
         const validTopicAndOriginator = (pgOptions.triggerTopics.includes(payload.topic)) && (pgOptions.triggerOriginators.includes(payload.originator)) // Check if valid topic and originator
         if (validTopicAndOriginator) {
           if (!isFailover) {
             logger.info('trying to push on kafka topic')
             await pushToKafka(payload)
-            audit(message)
             logger.info('pushed to kafka and added for audit trail')
           } else {
             logger.info('taking backup on dynamodb for reconciliation')
             await pushToDynamoDb(payload)
           }
+          audit(message)
         } else {
           logger.debug('Ignoring message with incorrect topic or originator')
           // push to slack - alertIt("slack message")
@@ -43,7 +44,7 @@ async function setupPgClient() {
         logger.debug(`error-sync: producer parse message : "${error.message}"`)
         logger.logFullError(error)
         if (!isFailover) {
-          await auditTrail([pl_randonseq, 1111, 'pl_table', 'pl_uniquecolumn', 'pl_operation', "error-producer", "", "", error.message, 'pl_payload', new Date(), 'pl_topic'], 'producer')
+          await auditTrail([pl_randonseq, 1111, "", "", "", "error-producer", "", "", error.message, "", new Date(), ""], 'producer')
         }
         // push to slack - alertIt("slack message"
       }
@@ -52,6 +53,7 @@ async function setupPgClient() {
   } catch (err) {
     logger.error('Error in setting up postgres client: ', err.message)
     logger.logFullError(err)
+    // push to slack - alertIt("slack message")
     terminate()
   }
 }
@@ -70,16 +72,27 @@ async function audit(message) {
   const pl_processid = message.processId
   const payload = JSON.parse(message.payload)
   const pl_seqid = payload.payload.payloadseqid
-  const pl_topic = payload.topic
+  const pl_topic = payload.topic // TODO can move in config ? 
   const pl_table = payload.payload.table
   const pl_uniquecolumn = payload.payload.Uniquecolumn
   const pl_operation = payload.payload.operation
   const pl_timestamp = payload.timestamp
   const pl_payload = JSON.stringify(payload.payload)
-  logger.debug(`producer : ${pl_seqid} ${pl_processid} ${pl_table} ${pl_uniquecolumn} ${pl_operation} ${payload.timestamp}`);
+  const logMessage = `${pl_seqid} ${pl_processid} ${pl_table} ${pl_uniquecolumn} ${pl_operation} ${payload.timestamp}`
+  if (!isFailover) {
+    logger.debug(`producer : ${logMessage}`);
+  } else {
+    logger.debug(`Producer DynamoDb : ${logMessage}`);
+  }
   auditTrail([pl_seqid, pl_processid, pl_table, pl_uniquecolumn, pl_operation, "push-to-kafka", "", "", "", pl_payload, pl_timestamp, pl_topic], 'producer')
 }
 
+function alertIt(message) {
+  /**
+      // call slack 
+   */
+}
+
 app.get('/health', (req, res) => {
   res.send('health ok')
 })