InboxOutboxService

The InboxOutboxService is a NestJS service that handles the processing of messages from an inbox collection, performs aggregation operations, and moves the processed messages to a target collection. It also handles retries and moves failed messages to a dead letter collection and eventually to a failed messages collection if the maximum number of retries is reached.

Installation

To install the package, use npm or yarn:

npm install

or

yarn add

Configuration

The service requires a configuration object to be provided. This configuration object should include the names of the inbox and target collections, as well as the aggregation pipelines for different event types.

export interface InboxOutboxConfig {
  inboxCollectionName: string;
  targetCollectionName: string;
  aggregationPipelines: { [eventType: string]: any[] };
}

Usage

Importing the Service

First, import the InboxOutboxService into your module:

import { Module } from '@nestjs/common';
import { MongooseModule } from '@nestjs/mongoose';
import { ConfigModule, ConfigService } from '@nestjs/config';
import { InboxOutboxService, InboxOutboxConfig } from './inbox-outbox.service';

@Module({
  imports: [
    MongooseModule.forRoot('mongodb://localhost/nest'),
    ConfigModule.forRoot(),
  ],
  providers: [
    {
      provide: 'INBOX_OUTBOX_CONFIG',
      useFactory: (configService: ConfigService): InboxOutboxConfig => ({
        inboxCollectionName: configService.get<string>('INBOX_COLLECTION_NAME'),
        targetCollectionName: configService.get<string>('TARGET_COLLECTION_NAME'),
        aggregationPipelines: {
          // Define your aggregation pipelines here
        },
      }),
      inject: [ConfigService],
    },
    InboxOutboxService,
  ],
})
export class AppModule {}

Using the Service

The InboxOutboxService will automatically start watching the inbox collection for new messages and process them accordingly. You can also manually add messages to the outbox collection using the addToOutbox method.

import { Controller, Post, Body } from '@nestjs/common';
import { InboxOutboxService } from './inbox-outbox.service';

@Controller()
export class AppController {
  constructor(private readonly inboxOutboxService: InboxOutboxService) {}

  @Post('send')
  async sendMessage(@Body('message') message: string) {
    await this.inboxOutboxService.addToOutbox(message);
  }
}

Methods

addToOutbox(message: string): Promise<void>

Adds a message to the outbox collection.

retryFailedMessages(): Promise<void>

Retries processing of failed messages from the dead letter collection.

startRetryLoop(baseDelay = 5000, maxBackoffTime = 10000): Promise<void>

Starts a loop to periodically retry processing of failed messages.

Logging

The service includes logging statements to track status updates, processing steps, cleanup steps, and retry steps. This helps with debugging and provides better visibility into the processing flow.

License

This project is licensed under the MIT License.