nestjs-sqs

Install the library and the required AWS SDK client for SQS using npm:

npm i --save @ssut/nestjs-sqs @aws-sdk/client-sqs

If you need to load configuration asynchronously (e.g., from a config service), use SqsModule.registerAsync(). You can use the useFactory, useClass, or useExisting patterns.

// Use factory
SqsModule.registerAsync({
  useFactory: () => {
    return {
      consumers: [...],
      producers: [...],
    };
  },
});

// Use class
SqsModule.registerAsync({
  useClass: SqsConfigService,
});

// Use existing
SqsModule.registerAsync({
  imports: [ConfigModule],
  useExisting: ConfigService,
});

Use the @SqsConsumerEventHandler decorator to listen for specific lifecycle or error events from an SQS consumer. You must provide the consumer name and the specific eventName (e.g., processing_error).

@SqsConsumerEventHandler(
  /** name: */ "myConsumer1",
  /** eventName: */ "processing_error",
)
public onProcessingError(error: Error, message: Message) {
  // report errors here
}

Use the @SqsMessageHandler decorator on methods within NestJS providers to automatically handle incoming SQS messages. The decorator requires the name of the consumer instance configured in the module. You can optionally specify if the handler should process messages in batches.

import { Message } from "@aws-sdk/client-sqs";

@Injectable()
export class AppMessageHandler {
  @SqsMessageHandler(/** name: */ "myConsumer1", /** batch: */ false)
  public async handleMessage(message: Message) {}

  @SqsConsumerEventHandler(
    /** name: */ "myConsumer1",
    /** eventName: */ "processing_error",
  )
  public onProcessingError(error: Error, message: Message) {
    // report errors here
  }
}

Inject SqsService into your class to send messages to an SQS queue. Use the .send() method, passing the name of the producer instance and a message object. The message object can include id, body, groupId, deduplicationId, messageAttributes, and delaySeconds.

export class AppService {
  public constructor(
    private readonly sqsService: SqsService,
  ) { }

  public async dispatchSomething() {
    await this.sqsService.send(/** name: */ 'myProducer1', {
      id: 'id',
      body: { ... },
      groupId: 'groupId',
      deduplicationId: 'deduplicationId',
      messageAttributes: { ... },
      delaySeconds: 0,
    });
  }
}

When registering the SqsModule, you provide an SqsOptions object to define your consumers, producers, and logging behavior.

Key properties:

• consumers: An array of SqsConsumerOptions defining which queues to listen to.
• producers: An array of SqsProducerOptions defining which queues can send messages.
• logger: An optional NestJS LoggerService for module logging.
• globalStopOptions: Optional StopOptions applied globally.

export interface SqsOptions {
  consumers?: SqsConsumerOptions[];
  producers?: SqsProducerOptions[];
  logger?: LoggerService;
  globalStopOptions?: StopOptions;
}

The @ssut/nestjs-sqs package exports the core components required to integrate Amazon SQS with NestJS. You can import the SqsModule, SqsService, and various decorators (such as @SqsHandler) directly from the package entrypoint.

Key exports include:

• SqsModule: The NestJS module used to configure and provide SQS functionality.
• SqsService: The service used to produce (send) messages to SQS queues.
• Decorators: Used to define message handlers for specific queues.

The SqsService maintains internal maps of all active SQS components. You can access these directly if you need to interact with the underlying sqs-consumer or sqs-producer instances.

• consumers: A Map<QueueName, SqsConsumerMapValues> containing the consumer instances and their stopOptions.
• producers: A Map<QueueName, Producer> containing the producer instances.

The send method allows you to dispatch messages to a specific queue using a registered producer. You can send a single message or an array of messages. If the message body is not a string, the service will automatically stringify it using JSON.stringify before sending.

Parameters:

• name: The QueueName of the registered producer.
• payload: A single Message<T> or an array of Message<T>[].

// Sending a single message
await sqsService.send('my-queue-name', { body: { foo: 'bar' } });

// Sending multiple messages
await sqsService.send('my-queue-name', [
  { body: { id: 1 } },
  { body: { id: 2 } }
]);

The SqsService provides administrative methods to interact with the underlying SQS queues via the AWS SDK. These methods require that a consumer or producer for the specified queue name has already been registered in the module configuration.

• purgeQueue(name): Deletes all messages from the specified queue.
• getQueueAttributes(name): Retrieves all attributes for the specified queue.
• getProducerQueueSize(name): Returns the number of messages currently in the producer's internal queue.

// Purge all messages from a queue
await sqsService.purgeQueue('my-queue-name');

// Get all attributes for a queue
const attributes = await sqsService.getQueueAttributes('my-queue-name');

// Check the size of the producer's internal queue
const size = sqsService.getProducerQueueSize('my-queue-name');

If you need to load SQS configuration from an external source (like a database or a remote config service) or inject other services into your configuration, use SqsModuleAsyncOptions. This follows the standard NestJS asynchronous provider pattern.

Supported patterns:

• useExisting: Use an existing provider that implements SqsModuleOptionsFactory.
• useClass: Use a class that implements SqsModuleOptionsFactory.
• useFactory: A factory function that returns SqsOptions or a Promise<SqsOptions>.

export interface SqsModuleAsyncOptions extends Pick<ModuleMetadata, 'imports'> {
  useExisting?: Type<SqsModuleOptionsFactory>;
  useClass?: Type<SqsModuleOptionsFactory>;
  useFactory?: (...args: any[]) => Promise<SqsOptions> | SqsOptions;
  inject?: any[];
}