Nodejs worker fixes (#1854)

Author: Uros Marolt

backend/src/bin/jobs/checkSqsQueues.ts

@@ -2,7 +2,6 @@ import { CrowdJob } from '../../types/jobTypes'
 import integrationTicks from './integrationTicks'
 import weeklyAnalyticsEmailsCoordinator from './weeklyAnalyticsEmailsCoordinator'
 import memberScoreCoordinator from './memberScoreCoordinator'
-import checkSqsQueues from './checkSqsQueues'
 import refreshMaterializedViews from './refreshMaterializedViews'
 import refreshMaterializedViewsForCube from './refreshMaterializedViewsForCube'
 import downgradeExpiredPlans from './downgradeExpiredPlans'
@@ -20,7 +19,6 @@ const EMAILS_ENABLED = WEEKLY_EMAILS_CONFIG.enabled === 'true'
 const jobs: CrowdJob[] = [
   integrationTicks,
   memberScoreCoordinator,
-  checkSqsQueues,
   refreshMaterializedViews,
   refreshMaterializedViewsForCube,
   downgradeExpiredPlans,

backend/src/bin/jobs/index.ts

@@ -1,16 +1,23 @@
 import { timeout } from '@crowd/common'
 import { Logger, getChildLogger, getServiceLogger, logExecutionTimeV2 } from '@crowd/logging'
+import {
+  SqsDeleteMessageRequest,
+  SqsMessage,
+  SqsReceiveMessageRequest,
+  deleteMessage,
+  receiveMessage,
+  sendMessage,
+} from '@crowd/sqs'
 import { SpanStatusCode, getServiceTracer } from '@crowd/tracing'
-import { DeleteMessageRequest, Message, ReceiveMessageRequest } from 'aws-sdk/clients/sqs'
 import moment from 'moment'
 import { SQS_CONFIG } from '../conf'
 import { processDbOperationsMessage } from '../serverless/dbOperations/workDispatcher'
 import { processNodeMicroserviceMessage } from '../serverless/microservices/nodejs/workDispatcher'
 import { NodeWorkerMessageType } from '../serverless/types/workerTypes'
 import { sendNodeWorkerMessage } from '../serverless/utils/nodeWorkerSQS'
 import { NodeWorkerMessageBase } from '../types/mq/nodeWorkerMessageBase'
-import { deleteMessage, receiveMessage, sendMessage } from '../utils/sqs'
 import { processIntegration, processWebhook } from './worker/integrations'
+import { SQS_CLIENT } from '@/serverless/utils/serviceSQS'
 
 /* eslint-disable no-constant-condition */
 
@@ -26,24 +33,30 @@ process.on('SIGTERM', async () => {
   exiting = true
 })
 
-const receive = (delayed?: boolean): Promise<Message | undefined> => {
-  const params: ReceiveMessageRequest = {
+const receive = async (delayed?: boolean): Promise<SqsMessage | undefined> => {
+  const params: SqsReceiveMessageRequest = {
     QueueUrl: delayed ? SQS_CONFIG.nodejsWorkerDelayableQueue : SQS_CONFIG.nodejsWorkerQueue,
     MessageAttributeNames: !delayed
       ? undefined
       : ['remainingDelaySeconds', 'tenantId', 'targetQueueUrl'],
   }
 
-  return receiveMessage(params)
+  const messages = await receiveMessage(SQS_CLIENT(), params)
+
+  if (messages && messages.length === 1) {
+    return messages[0]
+  }
+
+  return undefined
 }
 
 const removeFromQueue = (receiptHandle: string, delayed?: boolean): Promise<void> => {
-  const params: DeleteMessageRequest = {
+  const params: SqsDeleteMessageRequest = {
     QueueUrl: delayed ? SQS_CONFIG.nodejsWorkerDelayableQueue : SQS_CONFIG.nodejsWorkerQueue,
     ReceiptHandle: receiptHandle,
   }
 
-  return deleteMessage(params)
+  return deleteMessage(SQS_CLIENT(), params)
 }
 
 async function handleDelayedMessages() {
@@ -81,7 +94,7 @@ async function handleDelayedMessages() {
             if (message.MessageAttributes.targetQueueUrl) {
               const targetQueueUrl = message.MessageAttributes.targetQueueUrl.StringValue
               messageLogger.debug({ tenantId, targetQueueUrl }, 'Successfully delayed a message!')
-              await sendMessage({
+              await sendMessage(SQS_CLIENT(), {
                 QueueUrl: targetQueueUrl,
                 MessageGroupId: tenantId,
                 MessageDeduplicationId: `${tenantId}-${moment().valueOf()}`,
@@ -129,7 +142,7 @@ async function handleMessages() {
   })
   handlerLogger.info('Listening for messages!')
 
-  const processSingleMessage = async (message: Message): Promise<void> => {
+  const processSingleMessage = async (message: SqsMessage): Promise<void> => {
     await tracer.startActiveSpan('ProcessMessage', async (span) => {
       const msg: NodeWorkerMessageBase = JSON.parse(message.Body)
 

backend/src/bin/nodejs-worker.ts

@@ -1,12 +1,12 @@
-import { MessageBodyAttributeMap } from 'aws-sdk/clients/sqs'
-import moment from 'moment'
 import { getServiceChildLogger } from '@crowd/logging'
+import { SqsMessageAttributes, sendMessage } from '@crowd/sqs'
 import { AutomationTrigger } from '@crowd/types'
-import { NodeWorkerMessageBase } from '../../types/mq/nodeWorkerMessageBase'
+import moment from 'moment'
 import { IS_TEST_ENV, SQS_CONFIG } from '../../conf'
-import { sendMessage } from '../../utils/sqs'
-import { NodeWorkerMessageType } from '../types/workerTypes'
+import { NodeWorkerMessageBase } from '../../types/mq/nodeWorkerMessageBase'
 import { ExportableEntity } from '../microservices/nodejs/messageTypes'
+import { NodeWorkerMessageType } from '../types/workerTypes'
+import { SQS_CLIENT } from './serviceSQS'
 
 const log = getServiceChildLogger('nodeWorkerSQS')
 
@@ -24,7 +24,7 @@ export const sendNodeWorkerMessage = async (
   }
 
   // we can only delay for 15 minutes then we have to re-delay message
-  let attributes: MessageBodyAttributeMap
+  let attributes: SqsMessageAttributes
   let delay: number
   let delayed = false
   if (delaySeconds) {
@@ -80,7 +80,7 @@ export const sendNodeWorkerMessage = async (
     },
     'Sending nodejs-worker sqs message!',
   )
-  await sendMessage(params)
+  await sendMessage(SQS_CLIENT(), params)
 }
 
 export const sendNewActivityNodeSQSMessage = async (

backend/src/bin/scripts/trigger-webhook.ts

@@ -1,7 +1,8 @@
+import { sendMessage } from '@crowd/sqs'
 import moment from 'moment'
-import { sqs } from '../../services/aws'
 import { IS_TEST_ENV, KUBE_MODE, SQS_CONFIG } from '../../conf'
 import { PythonWorkerMessage } from '../types/workerTypes'
+import { SQS_CLIENT } from './serviceSQS'
 
 export const sendPythonWorkerMessage = async (
   tenantId: string,
@@ -16,12 +17,10 @@ export const sendPythonWorkerMessage = async (
     throw new Error("Can't send python-worker SQS message when not in kube mode!")
   }
 
-  await sqs
-    .sendMessage({
-      QueueUrl: SQS_CONFIG.pythonWorkerQueue,
-      MessageGroupId: tenantId,
-      MessageDeduplicationId: `${tenantId}-${moment().valueOf()}`,
-      MessageBody: JSON.stringify(body),
-    })
-    .promise()
+  await sendMessage(SQS_CLIENT(), {
+    QueueUrl: SQS_CONFIG.pythonWorkerQueue,
+    MessageGroupId: tenantId,
+    MessageDeduplicationId: `${tenantId}-${moment().valueOf()}`,
+    MessageBody: JSON.stringify(body),
+  })
 }