Merge branch 'main' into test/node-rdkafka_perf_test

# Conflicts:
#	packages/kafka/lib/test.spec.ts

Author: CarlosGamero

.github/workflows/ci.yml

@@ -7,10 +7,10 @@ on:
   pull_request:
 
 jobs:
-  call-ci-flow:
+  general:
     strategy:
       matrix:
-        node-version: [18.x, 20.x, 22.x, 23.x]
+        node-version: [20.x, 22.x, 24.x]
         package-name: [
           '@message-queue-toolkit/amqp',
           '@message-queue-toolkit/core',
@@ -28,7 +28,7 @@ jobs:
       package_name: ${{ matrix.package-name }}
 
   automerge:
-    needs: [ call-ci-flow ]
+    needs: [ general ]
     runs-on: ubuntu-latest
     permissions:
       pull-requests: write

examples/sns-sqs/biome.json

@@ -0,0 +1,8 @@
+{
+    "$schema": "./node_modules/@biomejs/biome/configuration_schema.json",
+    "extends": [
+        "./node_modules/@lokalise/biome-config/configs/biome-base.jsonc",
+        "./node_modules/@lokalise/biome-config/configs/biome-esm.jsonc",
+        "./node_modules/@lokalise/biome-config/configs/biome-package.jsonc"
+    ]
+}

examples/sns-sqs/docker-compose.yml

@@ -0,0 +1,20 @@
+services:
+
+  localstack:
+    image: localstack/localstack:4.0.2
+    network_mode: bridge
+    hostname: localstack
+    ports:
+      - '127.0.0.1:4566:4566' # LocalStack Gateway
+      - '127.0.0.1:4510-4559:4510-4559' # external services port range
+    environment:
+      - SERVICES=sns,sqs,s3,sts
+      - DEBUG=0
+      - DATA_DIR=${DATA_DIR-}
+      - DOCKER_HOST=unix:///var/run/docker.sock
+      - LOCALSTACK_HOST=localstack
+    #      - LOCALSTACK_API_KEY=someDummyKey
+    volumes:
+      - '${TMPDIR:-/tmp}/localstack:/var/log/localstack'
+      - '/var/run/docker.sock:/var/run/docker.sock'
+    restart: on-failure

examples/sns-sqs/lib/01-publish-message.spec.ts

@@ -0,0 +1,47 @@
+import { afterEach, beforeEach, describe, expect, it } from 'vitest'
+import { userConsumer } from './common/Dependencies.ts'
+import { publisherManager } from './common/TestPublisherManager.ts'
+import { UserConsumer } from './common/UserConsumer.js'
+
+describe('Publish message', () => {
+  beforeEach(async () => {
+    await publisherManager.initRegisteredPublishers([UserConsumer.SUBSCRIBED_TOPIC_NAME])
+    await userConsumer.start()
+  })
+
+  afterEach(async () => {
+    await userConsumer.close()
+  })
+
+  it('Publishes a message', async () => {
+    await publisherManager.publish(UserConsumer.SUBSCRIBED_TOPIC_NAME, {
+      type: 'user.created',
+      payload: {
+        id: '456',
+        name: 'Jane Doe',
+      },
+    })
+
+    await publisherManager.publish(UserConsumer.SUBSCRIBED_TOPIC_NAME, {
+      type: 'user.created',
+      payload: {
+        id: '123',
+        name: 'John Doe',
+      },
+    })
+
+    const receivedMessage = await userConsumer.handlerSpy.waitForMessage({
+      type: 'user.created',
+      payload: {
+        name: 'John Doe',
+      },
+    })
+
+    expect(receivedMessage.message.payload).toMatchInlineSnapshot(`
+          {
+            "id": "123",
+            "name": "John Doe",
+          }
+        `)
+  })
+})

examples/sns-sqs/lib/02-configurable-init.spec.ts

@@ -0,0 +1,60 @@
+import { SnsConsumerErrorResolver } from '@message-queue-toolkit/sns'
+import { beforeAll, describe, it } from 'vitest'
+import {
+  errorReporter,
+  logger,
+  snsClient,
+  sqsClient,
+  stsClient,
+  transactionObservabilityManager,
+} from './common/Dependencies.ts'
+import { publisherManager } from './common/TestPublisherManager.ts'
+import { UserConsumer } from './common/UserConsumer.js'
+
+// This test suite illustrates the importance of only initting the consumers you need
+// to prevent test execution time from increasing with every new consumer added
+describe('Consumer init', () => {
+  beforeAll(async () => {
+    await publisherManager.initRegisteredPublishers([UserConsumer.SUBSCRIBED_TOPIC_NAME])
+  })
+
+  it('Inits one consumer', async () => {
+    const userConsumer = new UserConsumer({
+      errorReporter,
+      logger,
+      transactionObservabilityManager,
+      consumerErrorResolver: new SnsConsumerErrorResolver(),
+      sqsClient,
+      snsClient,
+      stsClient,
+    })
+
+    await userConsumer.start()
+    await userConsumer.close()
+  })
+
+  it('Inits twenty consumers', async () => {
+    const consumers: UserConsumer[] = []
+    for (let i = 0; i < 20; i++) {
+      consumers.push(
+        new UserConsumer({
+          errorReporter,
+          logger,
+          transactionObservabilityManager,
+          consumerErrorResolver: new SnsConsumerErrorResolver(),
+          sqsClient,
+          snsClient,
+          stsClient,
+        }),
+      )
+    }
+
+    for (const userConsumer of consumers) {
+      await userConsumer.start()
+    }
+
+    for (const userConsumer of consumers) {
+      await userConsumer.close()
+    }
+  })
+})

examples/sns-sqs/lib/common/Dependencies.ts

@@ -0,0 +1,38 @@
+import { SNSClient, type SNSClientConfig } from '@aws-sdk/client-sns'
+import { SQSClient } from '@aws-sdk/client-sqs'
+import { STSClient } from '@aws-sdk/client-sts'
+import { SnsConsumerErrorResolver } from '@message-queue-toolkit/sns'
+import { pino } from 'pino'
+import { UserConsumer } from './UserConsumer.ts'
+
+export const TEST_AWS_CONFIG: SNSClientConfig = {
+  endpoint: 'http://s3.localhost.localstack.cloud:4566',
+  region: 'eu-west-1',
+  credentials: {
+    accessKeyId: 'access',
+    secretAccessKey: 'secret',
+  },
+}
+
+export const sqsClient = new SQSClient(TEST_AWS_CONFIG)
+export const snsClient = new SNSClient(TEST_AWS_CONFIG)
+export const stsClient = new STSClient(TEST_AWS_CONFIG)
+
+export const errorReporter = { report: () => {} }
+export const logger = pino()
+export const transactionObservabilityManager = {
+  start: () => {},
+  startWithGroup: () => {},
+  stop: () => {},
+  addCustomAttributes: () => {},
+}
+
+export const userConsumer = new UserConsumer({
+  errorReporter,
+  logger,
+  transactionObservabilityManager,
+  consumerErrorResolver: new SnsConsumerErrorResolver(),
+  sqsClient,
+  snsClient,
+  stsClient,
+})

examples/sns-sqs/lib/common/TestMessages.ts

@@ -0,0 +1,41 @@
+import {
+  type SnsAwareEventDefinition,
+  enrichMessageSchemaWithBaseStrict,
+} from '@message-queue-toolkit/schemas'
+import type { CommonEventDefinition } from '@message-queue-toolkit/schemas'
+import { z } from 'zod/v3'
+
+type AllConsumerMessageSchemas<MessageDefinitionTypes extends CommonEventDefinition[]> = z.infer<
+  MessageDefinitionTypes[number]['consumerSchema']
+>
+
+export const USER_SCHEMA = z.object({
+  id: z.string(),
+  name: z.string(),
+  age: z.number().optional(),
+})
+
+export const UserEvents = {
+  created: {
+    ...enrichMessageSchemaWithBaseStrict('user.created', USER_SCHEMA, {
+      description: 'User was created',
+    }),
+    schemaVersion: '1.0.1',
+    producedBy: ['user-service'],
+    domain: 'users',
+    snsTopic: 'user',
+  },
+
+  updated: {
+    ...enrichMessageSchemaWithBaseStrict('user.updated', USER_SCHEMA, {
+      description: 'User was updated',
+    }),
+    schemaVersion: '1.0.1',
+    producedBy: ['user-service'],
+    domain: 'users',
+    snsTopic: 'user',
+  },
+} satisfies Record<string, SnsAwareEventDefinition>
+
+export type UserEventsType = (typeof UserEvents)[keyof typeof UserEvents][]
+export type UserEventConsumerPayloadsType = AllConsumerMessageSchemas<UserEventsType>

examples/sns-sqs/lib/common/TestPublisherManager.ts

@@ -0,0 +1,43 @@
+import { CommonMetadataFiller, EventRegistry } from '@message-queue-toolkit/core'
+import type { AllPublisherMessageSchemas } from '@message-queue-toolkit/schemas'
+import { SnsPublisherManager } from '@message-queue-toolkit/sns'
+import type { CommonSnsPublisher } from '@message-queue-toolkit/sns'
+import { CommonSnsPublisherFactory } from '@message-queue-toolkit/sns'
+import { errorReporter, logger, snsClient, stsClient } from './Dependencies.ts'
+import { UserEvents, type UserEventsType } from './TestMessages.ts'
+
+const isTest = true
+
+type PublisherTypes = AllPublisherMessageSchemas<UserEventsType>
+
+export const publisherManager = new SnsPublisherManager<
+  CommonSnsPublisher<PublisherTypes>,
+  UserEventsType
+>(
+  {
+    errorReporter,
+    logger,
+    eventRegistry: new EventRegistry(Object.values(UserEvents)),
+    snsClient,
+    stsClient,
+  },
+  {
+    metadataFiller: new CommonMetadataFiller({
+      serviceId: 'service',
+    }),
+    publisherFactory: new CommonSnsPublisherFactory(),
+    newPublisherOptions: {
+      handlerSpy: true,
+      messageIdField: 'id',
+      messageTypeField: 'type',
+      deletionConfig: {
+        deleteIfExists: isTest, // only enable this in tests
+        // and ensure that the owning side is doing the deletion.
+        // if it is enabled both on consumer and publisher side, you are likely to experience confusing behaviour
+      },
+      creationConfig: {
+        updateAttributesIfExists: true,
+      },
+    },
+  },
+)

examples/sns-sqs/lib/common/UserConsumer.ts

@@ -0,0 +1,55 @@
+import { MessageHandlerConfigBuilder } from '@message-queue-toolkit/core'
+import type { ConsumerMessageSchema } from '@message-queue-toolkit/schemas'
+import { AbstractSnsSqsConsumer, type SNSSQSConsumerDependencies } from '@message-queue-toolkit/sns'
+import { UserEvents } from './TestMessages.ts'
+import { userCreatedHandler } from './handlers/UserCreatedHandler.ts'
+import { userUpdatedHandler } from './handlers/UserUpdatedHandler.ts'
+
+type SupportedMessages = ConsumerMessageSchema<
+  typeof UserEvents.created | typeof UserEvents.updated
+>
+
+// biome-ignore lint/complexity/noBannedTypes: to be expanded later
+type ExecutionContext = {}
+
+const isTest = true
+
+export class UserConsumer extends AbstractSnsSqsConsumer<SupportedMessages, ExecutionContext> {
+  public static readonly CONSUMED_QUEUE_NAME = 'user-my_service'
+  public static readonly SUBSCRIBED_TOPIC_NAME = 'user'
+
+  constructor(dependencies: SNSSQSConsumerDependencies) {
+    super(
+      dependencies,
+      {
+        handlerSpy: true,
+        handlers: new MessageHandlerConfigBuilder<SupportedMessages, ExecutionContext>()
+          .addConfig(UserEvents.created, userCreatedHandler, {})
+          .addConfig(UserEvents.updated, userUpdatedHandler, {})
+          .build(),
+        messageTypeField: 'type',
+        // Consumer creates its own queue
+        creationConfig: {
+          queue: {
+            QueueName: UserConsumer.CONSUMED_QUEUE_NAME,
+          },
+        },
+        deletionConfig: {
+          deleteIfExists: isTest,
+        },
+        locatorConfig: {
+          // Topic is created by a publisher, consumer relies on it already existing.
+          // Note that in order for this to work correctly you need to ensure that
+          // publisher gets initialized first. If consumer will initialize first,
+          // publisher may delete already existing topic and subscription and break the setup
+          topicName: UserConsumer.SUBSCRIBED_TOPIC_NAME,
+        },
+        // consumer creates its own subscription
+        subscriptionConfig: {
+          updateAttributesIfExists: false,
+        },
+      },
+      {},
+    )
+  }
+}

examples/sns-sqs/lib/common/handlers/UserCreatedHandler.ts

@@ -0,0 +1,13 @@
+import type { Either } from '@lokalise/node-core'
+import type { z } from 'zod/v3'
+import type { USER_SCHEMA, UserEvents } from '../TestMessages.ts'
+
+let _latestData: z.infer<typeof USER_SCHEMA>
+
+export function userCreatedHandler(
+  message: z.infer<typeof UserEvents.created.consumerSchema>,
+): Promise<Either<'retryLater', 'success'>> {
+  _latestData = message.payload
+
+  return Promise.resolve({ result: 'success' })
+}