Feature/replay poison checks (#1254)

* ideas

* [functions-app] add poison queue replay function (no bindings)

* [functions-app] remove method that was not used

* [functions-app] set to disabled and a valid cron expression (once per year: 12 Dec 14:30)

functions-app/lib/azure-storage-helper.js

@@ -0,0 +1,99 @@
+'use strict'
+
+require('dotenv').config()
+const azureStorage = require('azure-storage')
+const bluebird = require('bluebird')
+
+let azureTableService
+let azureQueueService
+let azureBlobService
+
+const azureStorageHelper = {
+  /**
+   * Promisify and cache the azureTableService library as it still lacks Promise support
+   */
+  getPromisifiedAzureTableService: function getPromisifiedAzureTableService () {
+    if (azureTableService) {
+      return azureTableService
+    }
+    azureTableService = azureStorage.createTableService()
+    bluebird.promisifyAll(azureTableService, {
+      promisifier: (originalFunction) => function (...args) {
+        return new Promise((resolve, reject) => {
+          try {
+            originalFunction.call(this, ...args, (error, result, response) => {
+              if (error) {
+                return reject(error)
+              }
+              resolve({ result, response })
+            })
+          } catch (error) {
+            reject(error)
+          }
+        })
+      }
+    })
+
+    return azureTableService
+  },
+
+  /**
+   * Promisify the azureQueueService
+   * @return {*}
+   */
+  getPromisifiedAzureQueueService: function getPromisifiedAzureQueueService () {
+    if (azureQueueService) {
+      return azureQueueService
+    }
+    azureQueueService = azureStorage.createQueueService()
+    bluebird.promisifyAll(azureQueueService, {
+      promisifier: (originalFunction) => function (...args) {
+        return new Promise((resolve, reject) => {
+          try {
+            originalFunction.call(this, ...args, (error, result, response) => {
+              if (error) {
+                return reject(error)
+              }
+              resolve({ result, response })
+            })
+          } catch (error) {
+            reject(error)
+          }
+        })
+      }
+    })
+
+    return azureQueueService
+  },
+
+  /**
+   * Promisify the azureBlobService
+   * @return {*}
+   */
+  getPromisifiedAzureBlobService: function getPromisifiedAzureBlobService () {
+    if (azureBlobService) {
+      return azureBlobService
+    }
+    azureBlobService = azureStorage.createBlobService()
+    bluebird.promisifyAll(azureBlobService, {
+      promisifier: (originalFunction) => function (...args) {
+        return new Promise((resolve, reject) => {
+          try {
+            originalFunction.call(this, ...args, (error, result, response) => {
+              if (error) {
+                return reject(error)
+              }
+              resolve({ result, response })
+            })
+          } catch (error) {
+            reject(error)
+          }
+        })
+      }
+    })
+
+    return azureBlobService
+  }
+}
+
+module.exports = azureStorageHelper

functions-app/package.json

@@ -14,6 +14,7 @@
   "dependencies": {
     "archiver": "^3.0.0",
     "azure-storage": "^2.10.3",
+    "bluebird": "^3.5.5",
     "device": "^0.3.9",
     "dotenv": "^8.0.0",
     "fast-csv": "^2.5.0",

functions-app/poison-replay-completed-checks/function.json

@@ -0,0 +1,11 @@
+{
+  "disabled": true,
+  "bindings": [
+    {
+      "schedule": "0 30 14 20 12 *",
+      "name": "poisonReplayCompletedChecks",
+      "type": "timerTrigger",
+      "direction": "in"
+    }
+  ]
+}

functions-app/poison-replay-completed-checks/ideas.js

@@ -0,0 +1,33 @@
+'use strict'
+
+const fs = require('fs')
+const uuid = require('uuid/v4')
+const path = require('path')
+const compressionService = require('../../functions/lib/compression.service')
+
+module.exports = async function (context, check) {
+  let version
+  let message
+  if (check.version && check.version === '2') {
+    version = 2
+    context.log('decompressing v2 message')
+    message = compressionService.decompress(check.archive)
+  } else {
+    version = 1
+    message = check
+  }
+  const id = uuid()
+  const file = path.join('./', `v${version}`, `${id}.json`)
+  context.log(`saving to ${file}`)
+  try {
+    const jsonString = JSON.stringify(message)
+    fs.writeFileSync(file, jsonString)
+  } catch (error) {
+    context.log(error)
+  }
+}
+
+/*
+1. understand which schools are affected
+2.
+*/

functions-app/poison-replay-completed-checks/index.js

@@ -0,0 +1,23 @@
+'use strict'
+
+const { performance } = require('perf_hooks')
+
+const v1 = require('./v1')
+const name = 'poison-replay-completed-checks'
+
+module.exports = async function (context) {
+  const start = performance.now()
+
+  let meta
+  try {
+    meta = await v1.process(context.log)
+  } catch (error) {
+    context.log.error(`${name}: ERROR: ${error.message}`)
+    throw error
+  }
+
+  const end = performance.now()
+  const durationInMilliseconds = end - start
+  const timeStamp = new Date().toISOString()
+  context.log(`${name}: ${timeStamp} processed ${meta.processCount} checks, errors: ${meta.errorCount};  run took ${durationInMilliseconds} ms`)
+}

functions-app/poison-replay-completed-checks/v1.js

@@ -0,0 +1,72 @@
+'use strict'
+const azureStorageHelper = require('../lib/azure-storage-helper')
+const R = require('ramda')
+
+const functionName = 'poison-replay-completed-checks'
+const completedCheckPoisonQueue = 'check-complete-poison'
+const completedCheckQueue = 'check-complete'
+let logger
+
+async function replayMessage (result) {
+  const azureQueueService = azureStorageHelper.getPromisifiedAzureQueueService()
+  const completedCheckMessage = JSON.parse(Buffer.from(result.messageText, 'base64').toString())
+  const checkCode = R.prop('checkCode', completedCheckMessage)
+
+  if (checkCode) {
+    logger.info(`${functionName}: replaying check [${checkCode}] onto queue [${completedCheckQueue}]`)
+  } else {
+    logger.info(`${functionName}: checkCode not found - replaying unlikely to work`)
+  }
+
+  // replay message
+  const insertResult = await azureQueueService.createMessageAsync(completedCheckQueue, result.messageText)
+  if (insertResult.response.isSuccessful === true) {
+    // delete the message from the poison queue
+    await azureQueueService.deleteMessageAsync(completedCheckPoisonQueue, result.messageId, result.popReceipt)
+  }
+}
+
+const v1 = {
+
+  /**
+   * Replay multiple messages from the poison queue to the processing queue
+   * See: https://azure.github.io/azure-storage-node/QueueService.html#getMessages__anchor
+   * @param loggerArg Context.log
+   * @return {Promise<{processCount: number, errorCount: number}>}
+   */
+  process: async function process (loggerArg) {
+    logger = loggerArg
+    let messages
+    let done = false
+    const meta = { processCount: 0, errorCount: 0 }
+    const azureQueueService = azureStorageHelper.getPromisifiedAzureQueueService()
+
+    while (!done) {
+      try {
+        const options = { numOfMessages: 32, visibilityTimeout: 120 }
+        messages = await azureQueueService.getMessagesAsync(completedCheckPoisonQueue, options)
+        logger.info(`${functionName}: Got ${messages.result.length} messages from the queue`)
+      } catch (error) {
+        logger.error(`${functionName}: Failed to fetch messages: ${error.message}`)
+        throw error
+      }
+
+      for (let message of messages.result) {
+        try {
+          await replayMessage(message)
+          meta.processCount += 1
+        } catch (error) {
+          logger.error(`${functionName}: Failed to replay message: ${error.message}`)
+          meta.errorCount += 1
+        }
+      }
+      if (messages.result.length === 0) {
+        done = true
+      }
+    }
+
+    return meta
+  }
+}
+
+module.exports = v1

functions-app/yarn.lock

@@ -340,6 +340,11 @@ bl@^2.0.1:
     readable-stream "^2.3.5"
     safe-buffer "^5.1.1"
 
+bluebird@^3.5.5:
+  version "3.5.5"
+  resolved "https://registry.yarnpkg.com/bluebird/-/bluebird-3.5.5.tgz#a8d0afd73251effbbd5fe384a77d73003c17a71f"
+  integrity sha512-5am6HnnfN+urzt4yfg7IgTbotDjIT/u8AJpEt0sIU9FtXfVeezXAPKswrG+xKUCOYAINpSdgZVDU6QFh+cuH3w==
+
 brace-expansion@^1.1.7:
   version "1.1.11"
   resolved "https://registry.yarnpkg.com/brace-expansion/-/brace-expansion-1.1.11.tgz#3c7fcbf529d87226f3d2f52b966ff5271eb441dd"