Dead letter queue

packages/sqs/index.ts

@@ -1,12 +1,15 @@
 export type {
-  SQSConsumerDependencies,
-  SQSQueueLocatorType,
+  ExtraSQSCreationParams,
   SQSDependencies,
+  SQSCreationConfig,
+  SQSQueueLocatorType,
 } from './lib/sqs/AbstractSqsService'
 
-export * from './lib/sqs/AbstractSqsConsumer'
 export { SqsConsumerErrorResolver } from './lib/errors/SqsConsumerErrorResolver'
 
+export type { SQSConsumerDependencies, SQSConsumerOptions } from './lib/sqs/AbstractSqsConsumer'
+export { AbstractSqsConsumer } from './lib/sqs/AbstractSqsConsumer'
+
 export { AbstractSqsPublisher } from './lib/sqs/AbstractSqsPublisher'
 export type { SQSMessageOptions } from './lib/sqs/AbstractSqsPublisher'
 

packages/sqs/lib/sqs/AbstractSqsConsumer.ts

@@ -1,4 +1,4 @@
-import type { CreateQueueRequest } from '@aws-sdk/client-sqs'
+import { SendMessageCommand, SetQueueAttributesCommand } from '@aws-sdk/client-sqs'
 import type { Either, ErrorResolver } from '@lokalise/node-core'
 import type {
   QueueConsumer as QueueConsumer,
@@ -7,6 +7,7 @@ import type {
   PrehandlingOutputs,
   Prehandler,
   BarrierResult,
+  QueueConsumerDependencies,
 } from '@message-queue-toolkit/core'
 import {
   isMessageError,
@@ -18,24 +19,51 @@ import { Consumer } from 'sqs-consumer'
 import type { ConsumerOptions } from 'sqs-consumer/src/types'
 
 import type { SQSMessage } from '../types/MessageTypes'
+import { deleteSqs, initSqs } from '../utils/sqsInitter'
 import { readSqsMessage } from '../utils/sqsMessageReader'
 
-import type { SQSConsumerDependencies, SQSQueueLocatorType } from './AbstractSqsService'
+import type { SQSCreationConfig, SQSDependencies, SQSQueueLocatorType } from './AbstractSqsService'
 import { AbstractSqsService } from './AbstractSqsService'
 
 const ABORT_EARLY_EITHER: Either<'abort', never> = {
   error: 'abort',
 }
 
-export type ExtraSQSCreationParams = {
-  topicArnsWithPublishPermissionsPrefix?: string
-  updateAttributesIfExists?: boolean
+// TODO: should we include DLQ types on core?
+type DeadLetterQueueCreationConfig<CreationConfigType extends SQSCreationConfig> = Omit<
+  CreationConfigType,
+  'queue'
+> & {
+  queue: Omit<CreationConfigType['queue'], 'QueueName'> &
+    (
+      | {
+          queueNameSuffix?: string
+          QueueName?: never
+        }
+      | {
+          queueNameSuffix?: never
+          QueueName?: string
+        }
+    )
 }
 
-export type SQSCreationConfig = {
-  queue: CreateQueueRequest
-  updateAttributesIfExists?: boolean
-} & ExtraSQSCreationParams
+type DeadLetterQueueOptions<
+  CreationConfigType extends SQSCreationConfig,
+  QueueLocatorType extends SQSQueueLocatorType,
+> = {
+  redrivePolicy: { maxReceiveCount: number }
+} & (
+  | {
+      creationConfig?: DeadLetterQueueCreationConfig<CreationConfigType>
+      locatorConfig?: never
+    }
+  | {
+      creationConfig?: never
+      locatorConfig?: QueueLocatorType
+    }
+)
+
+export type SQSConsumerDependencies = SQSDependencies & QueueConsumerDependencies
 
 export type SQSConsumerOptions<
   MessagePayloadSchemas extends object,
@@ -51,6 +79,7 @@ export type SQSConsumerOptions<
   PrehandlerOutput
 > & {
   consumerOverrides?: Partial<ConsumerOptions>
+  deadLetterQueue?: DeadLetterQueueOptions<CreationConfigType, QueueLocatorType>
 }
 export abstract class AbstractSqsConsumer<
     MessagePayloadType extends object,
@@ -97,6 +126,13 @@ export abstract class AbstractSqsConsumer<
   protected readonly messageSchemaContainer: MessageSchemaContainer<MessagePayloadType>
   protected readonly executionContext: ExecutionContext
 
+  private readonly deadLetterQueueOptions?: DeadLetterQueueOptions<
+    CreationConfigType,
+    QueueLocatorType
+  >
+
+  protected deadLetterQueueUrl?: string
+
   protected constructor(
     dependencies: SQSConsumerDependencies,
     options: ConsumerOptionsType,
@@ -107,8 +143,9 @@ export abstract class AbstractSqsConsumer<
     this.errorResolver = dependencies.consumerErrorResolver
 
     this.consumerOptionsOverride = options.consumerOverrides ?? {}
-    const messageSchemas = options.handlers.map((entry) => entry.schema)
+    this.deadLetterQueueOptions = options.deadLetterQueue
 
+    const messageSchemas = options.handlers.map((entry) => entry.schema)
     this.messageSchemaContainer = new MessageSchemaContainer<MessagePayloadType>({
       messageSchemas,
       messageTypeField: options.messageTypeField,
@@ -124,6 +161,61 @@ export abstract class AbstractSqsConsumer<
     this.executionContext = executionContext
   }
 
+  override async init(): Promise<void> {
+    await super.init()
+    await this.initDeadLetterQueue()
+  }
+
+  private async initDeadLetterQueue() {
+    if (!this.deadLetterQueueOptions) return
+
+    const dlqCreationConfig = this.resolvedDlqCreationConfig(this.deadLetterQueueOptions)
+    if (this.deletionConfig && dlqCreationConfig) {
+      await deleteSqs(this.sqsClient, this.deletionConfig, dlqCreationConfig)
+    }
+
+    const result = await initSqs(
+      this.sqsClient,
+      this.deadLetterQueueOptions.locatorConfig,
+      dlqCreationConfig,
+    )
+    const updateAttrCommand = new SetQueueAttributesCommand({
+      QueueUrl: this.queueUrl,
+      Attributes: {
+        RedrivePolicy: JSON.stringify({
+          deadLetterTargetArn: result.queueArn,
+          maxReceiveCount: this.deadLetterQueueOptions?.redrivePolicy.maxReceiveCount,
+        }),
+      },
+    })
+    await this.sqsClient.send(updateAttrCommand)
+
+    this.deadLetterQueueUrl = result.queueUrl
+  }
+
+  private resolvedDlqCreationConfig(
+    deadLetterQueueOptions: DeadLetterQueueOptions<CreationConfigType, QueueLocatorType>,
+  ): SQSCreationConfig | undefined {
+    if (deadLetterQueueOptions.locatorConfig) return undefined
+
+    let dlqName
+    if (deadLetterQueueOptions.creationConfig?.queue.QueueName) {
+      dlqName = deadLetterQueueOptions.creationConfig.queue.QueueName
+    } else if (deadLetterQueueOptions.creationConfig?.queue.queueNameSuffix) {
+      dlqName = `${this.queueName}${deadLetterQueueOptions.creationConfig.queue.queueNameSuffix}`
+    } else {
+      dlqName = `${this.queueName}-dlq`
+    }
+
+    return {
+      ...deadLetterQueueOptions.creationConfig,
+      queue: {
+        ...deadLetterQueueOptions.creationConfig?.queue,
+        QueueName: dlqName,
+      },
+    }
+  }
+
   public async start() {
     await this.init()
 
@@ -133,7 +225,6 @@ export abstract class AbstractSqsConsumer<
     this.consumer = Consumer.create({
       queueUrl: this.queueUrl,
       handleMessage: async (message: SQSMessage) => {
-        /* c8 ignore next */
         if (message === null) return
 
         const deserializedMessage = this.deserializeMessage(message)
@@ -295,7 +386,6 @@ export abstract class AbstractSqsConsumer<
       return ABORT_EARLY_EITHER
     }
     // Empty content for whatever reason
-    /* c8 ignore next */
     if (!resolveMessageResult.result) return ABORT_EARLY_EITHER
 
     // @ts-ignore
@@ -352,7 +442,13 @@ export abstract class AbstractSqsConsumer<
     }
   }
 
-  private async failProcessing(_message: SQSMessage) {
-    // Not implemented yet - needs dead letter queue
+  private async failProcessing(message: SQSMessage) {
+    if (!this.deadLetterQueueUrl) return
+
+    const command = new SendMessageCommand({
+      QueueUrl: this.deadLetterQueueUrl,
+      MessageBody: message.Body,
+    })
+    await this.sqsClient.send(command)
   }
 }

packages/sqs/lib/sqs/AbstractSqsPublisher.ts

@@ -13,9 +13,8 @@ import type { ZodSchema } from 'zod'
 
 import type { SQSMessage } from '../types/MessageTypes'
 
-import type { SQSCreationConfig } from './AbstractSqsConsumer'
 import { AbstractSqsService } from './AbstractSqsService'
-import type { SQSDependencies, SQSQueueLocatorType } from './AbstractSqsService'
+import type { SQSDependencies, SQSQueueLocatorType, SQSCreationConfig } from './AbstractSqsService'
 
 export type SQSMessageOptions = {
   MessageGroupId?: string

packages/sqs/lib/sqs/AbstractSqsService.ts

@@ -1,21 +1,23 @@
-import type { SQSClient } from '@aws-sdk/client-sqs'
-import type {
-  QueueConsumerDependencies,
-  QueueDependencies,
-  QueueOptions,
-} from '@message-queue-toolkit/core'
+import type { CreateQueueRequest, SQSClient } from '@aws-sdk/client-sqs'
+import type { QueueDependencies, QueueOptions } from '@message-queue-toolkit/core'
 import { AbstractQueueService } from '@message-queue-toolkit/core'
 
 import type { SQSMessage } from '../types/MessageTypes'
 import { deleteSqs, initSqs } from '../utils/sqsInitter'
 
-import type { SQSCreationConfig } from './AbstractSqsConsumer'
-
 export type SQSDependencies = QueueDependencies & {
   sqsClient: SQSClient
 }
 
-export type SQSConsumerDependencies = SQSDependencies & QueueConsumerDependencies
+export type ExtraSQSCreationParams = {
+  topicArnsWithPublishPermissionsPrefix?: string
+  updateAttributesIfExists?: boolean
+}
+
+export type SQSCreationConfig = {
+  queue: CreateQueueRequest
+  updateAttributesIfExists?: boolean
+} & ExtraSQSCreationParams
 
 export type SQSQueueLocatorType = {
   queueUrl: string

packages/sqs/lib/utils/sqsInitter.ts

@@ -3,8 +3,7 @@ import type { SQSClient, QueueAttributeName } from '@aws-sdk/client-sqs'
 import type { DeletionConfig } from '@message-queue-toolkit/core'
 import { isProduction } from '@message-queue-toolkit/core'
 
-import type { SQSCreationConfig } from '../sqs/AbstractSqsConsumer'
-import type { SQSQueueLocatorType } from '../sqs/AbstractSqsService'
+import type { SQSCreationConfig, SQSQueueLocatorType } from '../sqs/AbstractSqsService'
 
 import { assertQueue, deleteQueue, getQueueAttributes } from './sqsUtils'
 

packages/sqs/lib/utils/sqsUtils.ts

@@ -10,8 +10,7 @@ import type { CreateQueueCommandInput, SQSClient, QueueAttributeName } from '@aw
 import type { Either } from '@lokalise/node-core'
 import { isShallowSubset, waitAndRetry } from '@message-queue-toolkit/core'
 
-import type { ExtraSQSCreationParams } from '../sqs/AbstractSqsConsumer'
-import type { SQSQueueLocatorType } from '../sqs/AbstractSqsService'
+import type { ExtraSQSCreationParams, SQSQueueLocatorType } from '../sqs/AbstractSqsService'
 
 import { generateQueuePublishForTopicPolicy } from './sqsAttributeUtils'
 import { updateQueueAttributes } from './sqsInitter'

packages/sqs/test/consumers/SqsPermissionConsumer.spec.ts

@@ -18,12 +18,15 @@ import { SqsPermissionConsumer } from './SqsPermissionConsumer'
 
 describe('SqsPermissionConsumer', () => {
   describe('init', () => {
+    const queueName = 'myTestQueue'
+
     let diContainer: AwilixContainer<Dependencies>
     let sqsClient: SQSClient
+
     beforeEach(async () => {
       diContainer = await registerDependencies()
       sqsClient = diContainer.cradle.sqsClient
-      await deleteQueue(sqsClient, 'existingQueue')
+      await deleteQueue(sqsClient, queueName)
     })
 
     afterEach(async () => {
@@ -34,7 +37,7 @@ describe('SqsPermissionConsumer', () => {
     it('throws an error when invalid queue locator is passed', async () => {
       const newConsumer = new SqsPermissionConsumer(diContainer.cradle, {
         locatorConfig: {
-          queueUrl: 'http://s3.localhost.localstack.cloud:4566/000000000000/existingQueue',
+          queueUrl: `http://s3.localhost.localstack.cloud:4566/000000000000/${queueName}`,
         },
       })
 
@@ -43,24 +46,24 @@ describe('SqsPermissionConsumer', () => {
 
     it('does not create a new queue when queue locator is passed', async () => {
       await assertQueue(sqsClient, {
-        QueueName: 'existingQueue',
+        QueueName: queueName,
       })
 
       const newConsumer = new SqsPermissionConsumer(diContainer.cradle, {
         locatorConfig: {
-          queueUrl: 'http://s3.localhost.localstack.cloud:4566/000000000000/existingQueue',
+          queueUrl: `http://s3.localhost.localstack.cloud:4566/000000000000/${queueName}`,
         },
       })
 
       await newConsumer.init()
       expect(newConsumer.queueProps.url).toBe(
-        'http://s3.localhost.localstack.cloud:4566/000000000000/existingQueue',
+        `http://s3.localhost.localstack.cloud:4566/000000000000/${queueName}`,
       )
     })
 
     it('updates existing queue when one with different attributes exist', async () => {
       await assertQueue(sqsClient, {
-        QueueName: 'existingQueue',
+        QueueName: queueName,
         Attributes: {
           KmsMasterKeyId: 'somevalue',
         },
@@ -69,7 +72,7 @@ describe('SqsPermissionConsumer', () => {
       const newConsumer = new SqsPermissionConsumer(diContainer.cradle, {
         creationConfig: {
           queue: {
-            QueueName: 'existingQueue',
+            QueueName: queueName,
             Attributes: {
               KmsMasterKeyId: 'othervalue',
             },
@@ -86,7 +89,7 @@ describe('SqsPermissionConsumer', () => {
 
       await newConsumer.init()
       expect(newConsumer.queueProps.url).toBe(
-        'http://sqs.eu-west-1.localstack:4566/000000000000/existingQueue',
+        `http://sqs.eu-west-1.localstack:4566/000000000000/${queueName}`,
       )
 
       const updateCall = sqsSpy.mock.calls.find((entry) => {
@@ -103,7 +106,7 @@ describe('SqsPermissionConsumer', () => {
 
     it('does not update existing queue when attributes did not change', async () => {
       await assertQueue(sqsClient, {
-        QueueName: 'existingQueue',
+        QueueName: queueName,
         Attributes: {
           KmsMasterKeyId: 'somevalue',
         },
@@ -112,7 +115,7 @@ describe('SqsPermissionConsumer', () => {
       const newConsumer = new SqsPermissionConsumer(diContainer.cradle, {
         creationConfig: {
           queue: {
-            QueueName: 'existingQueue',
+            QueueName: queueName,
             Attributes: {
               KmsMasterKeyId: 'somevalue',
             },
@@ -129,7 +132,7 @@ describe('SqsPermissionConsumer', () => {
 
       await newConsumer.init()
       expect(newConsumer.queueProps.url).toBe(
-        'http://sqs.eu-west-1.localstack:4566/000000000000/existingQueue',
+        `http://sqs.eu-west-1.localstack:4566/000000000000/${queueName}`,
       )
 
       const updateCall = sqsSpy.mock.calls.find((entry) => {