Function processSingleMessage

backend/src/bin/nodejs-worker.ts:145–223

(message: SqsMessage)

  handlerLogger.info('Listening for messages!')

  const processSingleMessage = async (message: SqsMessage): Promise<void> => {
    await tracer.startActiveSpan('ProcessMessage', async (span) => {
      const msg: NodeWorkerMessageBase = JSON.parse(message.Body)

      const messageLogger = getChildLogger('messageHandler', serviceLogger, {
        messageId: message.MessageId,
        type: msg.type,
      })

      try {
        if (
          msg.type === NodeWorkerMessageType.NODE_MICROSERVICE &&
          (msg as any).service === 'enrich_member_organizations'
        ) {
          messageLogger.warn(
            'Skipping enrich_member_organizations message! Purging the queue because they are not needed anymore!',
          )
          await removeFromQueue(message.ReceiptHandle)
          return
        }

        messageLogger.info(
          { messageType: msg.type, messagePayload: JSON.stringify(msg) },
          'Received a new queue message!',
        )

        let processFunction: (msg: NodeWorkerMessageBase, logger?: Logger) => Promise<void>

        switch (msg.type) {
          case NodeWorkerMessageType.INTEGRATION_PROCESS:
            processFunction = processIntegration
            break
          case NodeWorkerMessageType.NODE_MICROSERVICE:
            processFunction = processNodeMicroserviceMessage
            break
          case NodeWorkerMessageType.DB_OPERATIONS:
            processFunction = processDbOperationsMessage
            break
          case NodeWorkerMessageType.PROCESS_WEBHOOK:
            processFunction = processWebhook
            break

          default:
            messageLogger.error('Error while parsing queue message! Invalid type.')
        }

        if (processFunction) {
          await logExecutionTimeV2(
            async () => {
              // remove the message from the queue as it's about to be processed
              await removeFromQueue(message.ReceiptHandle)
              messagesInProgress.set(message.MessageId, msg)
              try {
                await processFunction(msg, messageLogger)
              } catch (err) {
                messageLogger.error(err, 'Error while processing queue message!')
              } finally {
                messagesInProgress.delete(message.MessageId)