Extra logging for subscription failures

docker-compose.yml

@@ -8,7 +8,7 @@ services:
       - rabbit_data:/var/lib/rabbitmq
 
   localstack:
-    image: localstack/localstack:2.1.0
+    image: localstack/localstack:2.3.2
     network_mode: bridge
     hostname: localstack
     ports:
@@ -18,9 +18,9 @@ services:
       - SERVICES=sns,sqs
       - DEBUG=0
       - DATA_DIR=${DATA_DIR-}
-      - LAMBDA_EXECUTOR=local
       - DOCKER_HOST=unix:///var/run/docker.sock
       - HOSTNAME_EXTERNAL=localstack
+      - LOCALSTACK_HOST=localstack
     #      - LOCALSTACK_API_KEY=someDummyKey
     volumes:
       - '${TMPDIR:-/tmp}/localstack:/var/log/localstack'

packages/core/index.ts

@@ -5,6 +5,7 @@ export type {
   TransactionObservabilityManager,
   Logger,
   SchemaMap,
+  ExtraParams,
 } from './lib/types/MessageQueueTypes'
 
 export { AbstractQueueService } from './lib/queues/AbstractQueueService'

packages/core/lib/types/MessageQueueTypes.ts

@@ -27,6 +27,10 @@ export type LogFn = {
   (msg: string, ...args: any[]): void
 }
 
+export type ExtraParams = {
+  logger?: Logger
+}
+
 export type Logger = {
   error: LogFn
   info: LogFn

packages/sns/docker-compose.yml

@@ -10,9 +10,9 @@ services:
       - SERVICES=sns,sqs
       - DEBUG=0
       - DATA_DIR=${DATA_DIR-}
-      - LAMBDA_EXECUTOR=local
       - DOCKER_HOST=unix:///var/run/docker.sock
       - HOSTNAME_EXTERNAL=localstack
+      - LOCALSTACK_HOST=localstack
     #      - LOCALSTACK_API_KEY=someDummyKey
     volumes:
       - '${TMPDIR:-/tmp}/localstack:/var/log/localstack'

packages/sns/lib/sns/AbstractSnsSqsConsumerMonoSchema.ts

@@ -112,6 +112,9 @@ export abstract class AbstractSnsSqsConsumerMonoSchema<
         this.creationConfig.queue,
         this.creationConfig.topic,
         this.subscriptionConfig,
+        {
+          logger: this.logger,
+        },
       )
     } else if (this.deletionConfig && this.creationConfig) {
       await deleteSqs(this.sqsClient, this.deletionConfig, this.creationConfig)
@@ -123,6 +126,9 @@ export abstract class AbstractSnsSqsConsumerMonoSchema<
       this.locatorConfig,
       this.creationConfig,
       this.subscriptionConfig,
+      {
+        logger: this.logger,
+      },
     )
     this.queueUrl = initSnsSqsResult.queueUrl
     this.topicArn = initSnsSqsResult.topicArn

packages/sns/lib/sns/AbstractSnsSqsConsumerMultiSchema.ts

@@ -77,6 +77,9 @@ export abstract class AbstractSnsSqsConsumerMultiSchema<
       this.locatorConfig,
       this.creationConfig,
       this.subscriptionConfig,
+      {
+        logger: this.logger,
+      },
     )
     this.queueUrl = initSnsSqsResult.queueUrl
     this.topicArn = initSnsSqsResult.topicArn

packages/sns/lib/utils/snsInitter.ts

@@ -1,6 +1,6 @@
 import type { SNSClient, CreateTopicCommandInput } from '@aws-sdk/client-sns'
 import type { SQSClient, CreateQueueCommandInput } from '@aws-sdk/client-sqs'
-import type { DeletionConfig } from '@message-queue-toolkit/core'
+import type { DeletionConfig, ExtraParams } from '@message-queue-toolkit/core'
 import { isProduction } from '@message-queue-toolkit/core'
 import type { SQSCreationConfig } from '@message-queue-toolkit/sqs'
 import { deleteQueue, getQueueAttributes } from '@message-queue-toolkit/sqs'
@@ -18,6 +18,7 @@ export async function initSnsSqs(
   locatorConfig?: SNSSQSQueueLocatorType,
   creationConfig?: SNSCreationConfig & SQSCreationConfig,
   subscriptionConfig?: SNSSubscriptionOptions,
+  extraParams?: ExtraParams,
 ) {
   if (!locatorConfig?.subscriptionArn) {
     if (!creationConfig?.topic) {
@@ -46,6 +47,7 @@ export async function initSnsSqs(
         queueUrlsWithSubscribePermissionsPrefix:
           creationConfig.queueUrlsWithSubscribePermissionsPrefix,
         topicArnsWithPublishPermissionsPrefix: creationConfig.topicArnsWithPublishPermissionsPrefix,
+        logger: extraParams?.logger,
       },
     )
     if (!subscriptionArn) {
@@ -86,6 +88,7 @@ export async function deleteSnsSqs(
   queueConfiguration: CreateQueueCommandInput,
   topicConfiguration: CreateTopicCommandInput,
   subscriptionConfiguration: SNSSubscriptionOptions,
+  extraParams?: ExtraParams,
 ) {
   if (!deletionConfig.deleteIfExists) {
     return
@@ -103,6 +106,7 @@ export async function deleteSnsSqs(
     queueConfiguration,
     topicConfiguration,
     subscriptionConfiguration,
+    extraParams,
   )
 
   if (!subscriptionArn) {

packages/sns/lib/utils/snsSubscriber.ts

@@ -2,6 +2,7 @@ import type { CreateTopicCommandInput, SNSClient } from '@aws-sdk/client-sns'
 import { SubscribeCommand } from '@aws-sdk/client-sns'
 import type { SubscribeCommandInput } from '@aws-sdk/client-sns/dist-types/commands/SubscribeCommand'
 import type { CreateQueueCommandInput, SQSClient } from '@aws-sdk/client-sqs'
+import type { ExtraParams } from '@message-queue-toolkit/core'
 import type { ExtraSQSCreationParams } from '@message-queue-toolkit/sqs'
 import { assertQueue } from '@message-queue-toolkit/sqs'
 
@@ -20,7 +21,7 @@ export async function subscribeToTopic(
   queueConfiguration: CreateQueueCommandInput,
   topicConfiguration: CreateTopicCommandInput,
   subscriptionConfiguration: SNSSubscriptionOptions,
-  extraParams?: ExtraSNSCreationParams & ExtraSQSCreationParams,
+  extraParams?: ExtraSNSCreationParams & ExtraSQSCreationParams & ExtraParams,
 ) {
   const topicArn = await assertTopic(snsClient, topicConfiguration, {
     queueUrlsWithSubscribePermissionsPrefix: extraParams?.queueUrlsWithSubscribePermissionsPrefix,
@@ -37,11 +38,22 @@ export async function subscribeToTopic(
     ...subscriptionConfiguration,
   })
 
-  const subscriptionResult = await snsClient.send(subscribeCommand)
-  return {
-    subscriptionArn: subscriptionResult.SubscriptionArn,
-    topicArn,
-    queueUrl,
-    queueArn,
+  try {
+    const subscriptionResult = await snsClient.send(subscribeCommand)
+    return {
+      subscriptionArn: subscriptionResult.SubscriptionArn,
+      topicArn,
+      queueUrl,
+      queueArn,
+    }
+  } catch (err) {
+    const logger = extraParams?.logger ?? console
+    // @ts-ignore
+    logger.error(
+      `Error while creating subscription for queue "${queueConfiguration.QueueName}", topic "${
+        topicConfiguration.Name
+      }": ${(err as Error).message}`,
+    )
+    throw err
   }
 }

packages/sns/test/fakes/FakeLogger.ts

@@ -0,0 +1,26 @@
+import type { Logger } from '@message-queue-toolkit/core'
+
+export class FakeLogger implements Logger {
+  public readonly loggedMessages: unknown[] = []
+  public readonly loggedWarnings: unknown[] = []
+  public readonly loggedErrors: unknown[] = []
+
+  debug(obj: unknown) {
+    this.loggedMessages.push(obj)
+  }
+  error(obj: unknown) {
+    this.loggedErrors.push(obj)
+  }
+  fatal(obj: unknown) {
+    this.loggedErrors.push(obj)
+  }
+  info(obj: unknown) {
+    this.loggedMessages.push(obj)
+  }
+  trace(obj: unknown) {
+    this.loggedMessages.push(obj)
+  }
+  warn(obj: unknown) {
+    this.loggedWarnings.push(obj)
+  }
+}

packages/sns/test/utils/snsSubscriber.spec.ts

@@ -0,0 +1,86 @@
+import type { SNSClient } from '@aws-sdk/client-sns'
+import type { SQSClient } from '@aws-sdk/client-sqs'
+import { deleteQueue } from '@message-queue-toolkit/sqs'
+import type { AwilixContainer } from 'awilix'
+import { afterEach, describe, expect } from 'vitest'
+
+import { subscribeToTopic } from '../../lib/utils/snsSubscriber'
+import { deleteTopic } from '../../lib/utils/snsUtils'
+import { FakeLogger } from '../fakes/FakeLogger'
+
+import type { Dependencies } from './testContext'
+import { registerDependencies } from './testContext'
+
+const TOPIC_NAME = 'topic'
+const QUEUE_NAME = 'queue'
+
+describe('snsSubscriber', () => {
+  let diContainer: AwilixContainer<Dependencies>
+  let snsClient: SNSClient
+  let sqsClient: SQSClient
+  beforeEach(async () => {
+    diContainer = await registerDependencies({}, false)
+    snsClient = diContainer.cradle.snsClient
+    sqsClient = diContainer.cradle.sqsClient
+  })
+
+  afterEach(async () => {
+    const { awilixManager } = diContainer.cradle
+    await awilixManager.executeDispose()
+    await diContainer.dispose()
+
+    await deleteTopic(snsClient, TOPIC_NAME)
+    await deleteQueue(sqsClient, QUEUE_NAME)
+  })
+
+  describe('subscribeToTopic', () => {
+    it('logs queue in subscription error', async () => {
+      const logger = new FakeLogger()
+      await subscribeToTopic(
+        sqsClient,
+        snsClient,
+        {
+          QueueName: QUEUE_NAME,
+        },
+        {
+          Name: TOPIC_NAME,
+        },
+        {
+          Attributes: {
+            FilterPolicy: `{"type":["remove"]}`,
+            FilterPolicyScope: 'MessageAttributes',
+          },
+        },
+      )
+
+      await expect(
+        subscribeToTopic(
+          sqsClient,
+          snsClient,
+          {
+            QueueName: QUEUE_NAME,
+          },
+          {
+            Name: TOPIC_NAME,
+          },
+          {
+            Attributes: {
+              FilterPolicy: `{"type":["add"]}`,
+              FilterPolicyScope: 'MessageBody',
+            },
+          },
+          {
+            logger,
+          },
+        ),
+      ).rejects.toThrow(
+        /Invalid parameter: Attributes Reason: Subscription already exists with different attributes/,
+      )
+
+      expect(logger.loggedErrors).toHaveLength(1)
+      expect(logger.loggedErrors[0]).toBe(
+        'Error while creating subscription for queue "queue", topic "topic": Invalid parameter: Attributes Reason: Subscription already exists with different attributes',
+      )
+    })
+  })
+})

packages/sqs/docker-compose.yml

@@ -10,9 +10,9 @@ services:
       - SERVICES=sns
       - DEBUG=0
       - DATA_DIR=${DATA_DIR-}
-      - LAMBDA_EXECUTOR=local
       - DOCKER_HOST=unix:///var/run/docker.sock
       - HOSTNAME_EXTERNAL=localstack
+      - LOCALSTACK_HOST=localstack
     #      - LOCALSTACK_API_KEY=someDummyKey
     volumes:
       - '${TMPDIR:-/tmp}/localstack:/var/log/localstack'